STATUSES = [

        (STATUS_NOT_REVIEWED, _("Not reviewed")),  # (no icon) and default

        (STATUS_APPROVED, _("Approved")),

        (STATUS_REJECTED, _("Rejected")),

        (STATUS_UNDER_REVIEW, _("Under review")),

    ]

    changeset_status_id = Column(Integer(), nullable=False, unique=True, primary_key=True)

    repo_id = Column(Integer(), ForeignKey('repositories.repo_id'), nullable=False)

    user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False, unique=None)

    revision = Column(String(40), nullable=False)

    status = Column(String(128), nullable=False, default=DEFAULT)

    changeset_comment_id = Column(Integer(), ForeignKey('changeset_comments.comment_id'), nullable=False)

    modified_at = Column(DateTime(), nullable=False, default=datetime.datetime.now)

    version = Column(Integer(), nullable=False, default=0)

    pull_request_id = Column(Integer(), ForeignKey('pull_requests.pull_request_id'), nullable=True)

    author = relationship('User')

    repo = relationship('Repository')

    comment = relationship('ChangesetComment')

    pull_request = relationship('PullRequest')

    def __unicode__(self):

        return u"<%s('%s:%s')>" % (

            self.__class__.__name__,

            self.status, self.author

        )

    @classmethod

    def get_status_lbl(cls, value):

        return dict(cls.STATUSES).get(value)

    @property

    def status_lbl(self):

        return ChangesetStatus.get_status_lbl(self.status)

class PullRequest(Base, BaseModel):

    __tablename__ = 'pull_requests'

    __table_args__ = (

        Index('pr_org_repo_id_idx', 'org_repo_id'),

        Index('pr_other_repo_id_idx', 'other_repo_id'),

        {'extend_existing': True, 'mysql_engine': 'InnoDB',

         'mysql_charset': 'utf8', 'sqlite_autoincrement': True},

    )

    # values for .status

    STATUS_NEW = u'new'

    STATUS_CLOSED = u'closed'

    pull_request_id = Column(Integer(), nullable=False, unique=True, primary_key=True)

    title = Column(Unicode(255), nullable=True)

    description = Column(UnicodeText(10240))

    status = Column(Unicode(255), nullable=False, default=STATUS_NEW) # only for closedness, not approve/reject/etc

    created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)

    updated_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)

    user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False, unique=None)

    _revisions = Column('revisions', UnicodeText(20500))  # 500 revisions max

    org_repo_id = Column(Integer(), ForeignKey('repositories.repo_id'), nullable=False)

    org_ref = Column(Unicode(255), nullable=False)

    other_repo_id = Column(Integer(), ForeignKey('repositories.repo_id'), nullable=False)

    other_ref = Column(Unicode(255), nullable=False)

    @hybrid_property

    def revisions(self):

        return self._revisions.split(':')

    @revisions.setter

    def revisions(self, val):

        self._revisions = safe_unicode(':'.join(val))

    @property

    def org_ref_parts(self):

        return self.org_ref.split(':')

    @property

    def other_ref_parts(self):

        return self.other_ref.split(':')

    owner = relationship('User')

    reviewers = relationship('PullRequestReviewers',

                             cascade="all, delete-orphan")

    org_repo = relationship('Repository', primaryjoin='PullRequest.org_repo_id==Repository.repo_id')

    other_repo = relationship('Repository', primaryjoin='PullRequest.other_repo_id==Repository.repo_id')

    statuses = relationship('ChangesetStatus')

    comments = relationship('ChangesetComment',

                             cascade="all, delete-orphan")

    def is_closed(self):

        return self.status == self.STATUS_CLOSED

    def user_review_status(self, user_id):

        """Return the user's latest status votes on PR"""

        # note: no filtering on repo - that would be redundant

        status = ChangesetStatus.query()\

            .filter(ChangesetStatus.pull_request == self)\

            .filter(ChangesetStatus.user_id == user_id)\

            .order_by(ChangesetStatus.version)\

            .first()

        return str(status.status) if status else ''

    @classmethod

    def make_nice_id(cls, pull_request_id):

        '''Return pull request id nicely formatted for displaying'''

        return '#%s' % pull_request_id

    def nice_id(self):

        '''Return the id of this pull request, nicely formatted for displaying'''

        return self.make_nice_id(self.pull_request_id)

    def __json__(self):

        return dict(

            revisions=self.revisions

        )

    def url(self, **kwargs):

        canonical = kwargs.pop('canonical', None)

        import kallithea.lib.helpers as h

        b = self.org_ref_parts[1]

        if b != self.other_ref_parts[1]:

            s = '/_/' + b

        else:

            s = '/_/' + self.title

        kwargs['extra'] = urlreadable(s)

        if canonical:

            return h.canonical_url('pullrequest_show', repo_name=self.other_repo.repo_name,

                                   pull_request_id=self.pull_request_id, **kwargs)

        return h.url('pullrequest_show', repo_name=self.other_repo.repo_name,

                     pull_request_id=self.pull_request_id, **kwargs)

class PullRequestReviewers(Base, BaseModel):

    __tablename__ = 'pull_request_reviewers'

    __table_args__ = (

        {'extend_existing': True, 'mysql_engine': 'InnoDB',

         'mysql_charset': 'utf8', 'sqlite_autoincrement': True},

    )

    def __init__(self, user=None, pull_request=None):

        self.user = user

        self.pull_request = pull_request

    pull_requests_reviewers_id = Column(Integer(), nullable=False, unique=True, primary_key=True)

    pull_request_id = Column(Integer(), ForeignKey('pull_requests.pull_request_id'), nullable=False)

    user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=True)

    user = relationship('User')

    pull_request = relationship('PullRequest')

class Notification(Base, BaseModel):

    __tablename__ = 'notifications'

    __table_args__ = (

        Index('notification_type_idx', 'type'),

        {'extend_existing': True, 'mysql_engine': 'InnoDB',

         'mysql_charset': 'utf8', 'sqlite_autoincrement': True},

    )

    TYPE_CHANGESET_COMMENT = u'cs_comment'

    TYPE_MESSAGE = u'message'

    TYPE_MENTION = u'mention'

    TYPE_REGISTRATION = u'registration'

    TYPE_PULL_REQUEST = u'pull_request'

    TYPE_PULL_REQUEST_COMMENT = u'pull_request_comment'

    notification_id = Column(Integer(), nullable=False, unique=True, primary_key=True)

    subject = Column(Unicode(512), nullable=True)

    body = Column(UnicodeText(50000), nullable=True)

    created_by = Column(Integer(), ForeignKey('users.user_id'), nullable=True)

    created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)

    type_ = Column('type', Unicode(255))

    created_by_user = relationship('User')

    notifications_to_users = relationship('UserNotification', cascade="all, delete-orphan")

    @property

    def recipients(self):

        return [x.user for x in UserNotification.query()

                .filter(UserNotification.notification == self)

                .order_by(UserNotification.user_id.asc()).all()]

    @classmethod

    def create(cls, created_by, subject, body, recipients, type_=None):

        if type_ is None:

            type_ = Notification.TYPE_MESSAGE

        notification = cls()

        notification.created_by_user = created_by

        notification.subject = subject

        notification.body = body

        notification.type_ = type_

        notification.created_on = datetime.datetime.now()

        Session().add(notification)

        Session().flush() # assign notificaiton.notification_id

        return notification

    @property

    def description(self):

        from kallithea.model.notification import NotificationModel

        return NotificationModel().make_description(self)

class UserNotification(Base, BaseModel):

    __tablename__ = 'user_to_notification'

    __table_args__ = (

        UniqueConstraint('user_id', 'notification_id'),

        {'extend_existing': True, 'mysql_engine': 'InnoDB',

         'mysql_charset': 'utf8', 'sqlite_autoincrement': True}

    )

    user_id = Column(Integer(), ForeignKey('users.user_id'), primary_key=True)

    notification_id = Column(Integer(), ForeignKey('notifications.notification_id'), primary_key=True)

    read = Column(Boolean, default=False)

    sent_on = Column(DateTime(timezone=False), nullable=True, unique=None)

    user = relationship('User')

    notification = relationship('Notification')

    def mark_as_read(self):

        self.read = True

        Session().add(self)

class Gist(Base, BaseModel):

    __tablename__ = 'gists'

    __table_args__ = (

        Index('g_gist_access_id_idx', 'gist_access_id'),

        Index('g_created_on_idx', 'created_on'),

        {'extend_existing': True, 'mysql_engine': 'InnoDB',

         'mysql_charset': 'utf8', 'sqlite_autoincrement': True}

    )

    GIST_PUBLIC = u'public'

    GIST_PRIVATE = u'private'

    DEFAULT_FILENAME = u'gistfile1.txt'

    gist_id = Column(Integer(), nullable=False, unique=True, primary_key=True)

    gist_access_id = Column(Unicode(250))

    gist_description = Column(UnicodeText(1024))

    gist_owner = Column('user_id', Integer(), ForeignKey('users.user_id'), nullable=True)

    gist_expires = Column(Float(53), nullable=False)

    gist_type = Column(Unicode(128), nullable=False)

    created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)

    modified_at = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)

    owner = relationship('User')

    def __repr__(self):

        return '<Gist:[%s]%s>' % (self.gist_type, self.gist_access_id)

    @classmethod

    def get_or_404(cls, id_):

        res = cls.query().filter(cls.gist_access_id == id_).scalar()

        if res is None:

            raise HTTPNotFound

        return res

    @classmethod

    def get_by_access_id(cls, gist_access_id):

        return cls.query().filter(cls.gist_access_id == gist_access_id).scalar()

    def gist_url(self):

        import kallithea

        alias_url = kallithea.CONFIG.get('gist_alias_url')

        if alias_url:

            return alias_url.replace('{gistid}', self.gist_access_id)

        import kallithea.lib.helpers as h

        return h.canonical_url('gist', gist_id=self.gist_access_id)

    @classmethod

    def base_path(cls):

        """

        Returns base path where all gists are stored

        :param cls:

        """

        from kallithea.model.gist import GIST_STORE_LOC

        q = Session().query(Ui)\

            .filter(Ui.ui_key == URL_SEP)

        q = q.options(FromCache("sql_cache_short", "repository_repo_path"))

        return os.path.join(q.one().ui_value, GIST_STORE_LOC)

    def get_api_data(self):

        """

        Common function for generating gist related data for API

        """

        gist = self

        data = dict(

            gist_id=gist.gist_id,

            type=gist.gist_type,

            access_id=gist.gist_access_id,

            description=gist.gist_description,

            url=gist.gist_url(),

            expires=gist.gist_expires,

            created_on=gist.created_on,

        )

        return data

    def __json__(self):

        data = dict(

        )

        data.update(self.get_api_data())

        return data

    ## SCM functions

    @property

    def scm_instance(self):

        from kallithea.lib.vcs import get_repo

        base_path = self.base_path()

        return get_repo(os.path.join(*map(safe_str,

                                          [base_path, self.gist_access_id])))

class DbMigrateVersion(Base, BaseModel):

    __tablename__ = 'db_migrate_version'

    __table_args__ = (

        {'extend_existing': True, 'mysql_engine': 'InnoDB',

         'mysql_charset': 'utf8', 'sqlite_autoincrement': True},

    )

    repository_id = Column(String(250), nullable=False, unique=True, primary_key=True)

    repository_path = Column(Text)

    version = Column(Integer)

from kallithea.tests import *

from kallithea.model.db import User, Notification, UserNotification

from kallithea.model.user import UserModel

from kallithea.model.meta import Session

from kallithea.model.notification import NotificationModel

class TestNotifications(BaseTestCase):

    def __init__(self, methodName='runTest'):

        Session.remove()

        self.u1 = UserModel().create_or_update(username=u'u1',

                                        password=u'qweqwe',

                                        email=u'u1@example.com',

                                        firstname=u'u1', lastname=u'u1')

        Session().commit()

        self.u1 = self.u1.user_id

        self.u2 = UserModel().create_or_update(username=u'u2',

                                        password=u'qweqwe',

                                        email=u'u2@example.com',

                                        firstname=u'u2', lastname=u'u3')

        Session().commit()

        self.u2 = self.u2.user_id

        self.u3 = UserModel().create_or_update(username=u'u3',

                                        password=u'qweqwe',

                                        email=u'u3@example.com',

                                        firstname=u'u3', lastname=u'u3')

        Session().commit()

        self.u3 = self.u3.user_id

        super(TestNotifications, self).__init__(methodName=methodName)

    def setUp(self):

        self.remove_all_notifications()

        self.assertEqual([], Notification.query().all())

        self.assertEqual([], UserNotification.query().all())

    def test_create_notification(self):

        usrs = [self.u1, self.u2]

        notification = NotificationModel().create(created_by=self.u1,

                                           subject=u'subj', body=u'hi there',

                                           recipients=usrs)

        Session().commit()

        u1 = User.get(self.u1)

        u2 = User.get(self.u2)

        u3 = User.get(self.u3)

        notifications = Notification.query().all()

        self.assertEqual(len(notifications), 1)

        self.assertEqual(notifications[0].recipients, [u1, u2])

        self.assertEqual(notification.notification_id,

                         notifications[0].notification_id)

        unotification = UserNotification.query()\

            .filter(UserNotification.notification == notification).all()

        self.assertEqual(len(unotification), len(usrs))

        self.assertEqual(set([x.user.user_id for x in unotification]),

                         set(usrs))

    def test_user_notifications(self):

        notification1 = NotificationModel().create(created_by=self.u1,

                                            subject=u'subj', body=u'hi there1',

                                            recipients=[self.u3])

        Session().commit()

        notification2 = NotificationModel().create(created_by=self.u1,

                                            subject=u'subj', body=u'hi there2',

                                            recipients=[self.u3])

        Session().commit()

        u3 = Session().query(User).get(self.u3)

        self.assertEqual(sorted([x.notification for x in u3.notifications]),

                         sorted([notification2, notification1]))

    def test_delete_notifications(self):

        notification = NotificationModel().create(created_by=self.u1,

                                           subject=u'title', body=u'hi there3',

                                    recipients=[self.u3, self.u1, self.u2])

        Session().commit()

        notifications = Notification.query().all()

        self.assertTrue(notification in notifications)

        Notification.delete(notification.notification_id)

        Session().commit()

        notifications = Notification.query().all()

        self.assertFalse(notification in notifications)

        un = UserNotification.query().filter(UserNotification.notification

                                             == notification).all()

        self.assertEqual(un, [])

    def test_delete_association(self):

        notification = NotificationModel().create(created_by=self.u1,

                                           subject=u'title', body=u'hi there3',

                                    recipients=[self.u3, self.u1, self.u2])

        Session().commit()

        unotification = UserNotification.query()\

                            .filter(UserNotification.notification ==

                                    notification)\

                            .filter(UserNotification.user_id == self.u3)\

                            .scalar()

        self.assertEqual(unotification.user_id, self.u3)

        NotificationModel().delete(self.u3,

                                   notification.notification_id)

        Session().commit()

        u3notification = UserNotification.query()\

                            .filter(UserNotification.notification ==

                                    notification)\

                            .filter(UserNotification.user_id == self.u3)\

                            .scalar()

        self.assertEqual(u3notification, None)

        # notification object is still there

        self.assertEqual(Notification.query().all(), [notification])

        #u1 and u2 still have assignments

        u1notification = UserNotification.query()\

                            .filter(UserNotification.notification ==

                                    notification)\

                            .filter(UserNotification.user_id == self.u1)\

                            .scalar()

        self.assertNotEqual(u1notification, None)

        u2notification = UserNotification.query()\

                            .filter(UserNotification.notification ==

                                    notification)\

                            .filter(UserNotification.user_id == self.u2)\

                            .scalar()

        self.assertNotEqual(u2notification, None)

    def test_notification_counter(self):

        NotificationModel().create(created_by=self.u1,

                            subject=u'title', body=u'hi there_delete',

                            recipients=[self.u3, self.u1])

        Session().commit()

        self.assertEqual(NotificationModel()

        self.assertEqual(NotificationModel()

                         .get_unread_cnt_for_user(self.u2), 0)

        self.assertEqual(NotificationModel()

                         .get_unread_cnt_for_user(self.u3), 1)

        notification = NotificationModel().create(created_by=self.u1,

                                           subject=u'title', body=u'hi there3',

                                    recipients=[self.u3, self.u1, self.u2])

        Session().commit()

        self.assertEqual(NotificationModel()

        self.assertEqual(NotificationModel()

                         .get_unread_cnt_for_user(self.u2), 1)

        self.assertEqual(NotificationModel()

                         .get_unread_cnt_for_user(self.u3), 2)