Returns user associated with this ChangesetComment. ie those

        who actually commented

        :param cls:

        :param revision:

        """

        q = Session().query(User)\

                .join(ChangesetComment.author)

        if revision:

            q = q.filter(cls.revision == revision)

        elif pull_request_id:

            q = q.filter(cls.pull_request_id == pull_request_id)

        return q.all()

class ChangesetStatus(Base, BaseModel):

    __tablename__ = 'changeset_statuses'

    __table_args__ = (

        Index('cs_revision_idx', 'revision'),

        Index('cs_version_idx', 'version'),

        UniqueConstraint('repo_id', 'revision', 'version'),

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

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

    )

    STATUS_NOT_REVIEWED = DEFAULT = 'not_reviewed'

    STATUS_APPROVED = 'approved'

    STATUS_REJECTED = 'rejected'

    STATUS_UNDER_REVIEW = 'under_review'

    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('changeset_status_id', Integer(), nullable=False, primary_key=True)

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

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

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

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

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

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

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

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

    author = relationship('User', lazy='joined')

    repo = relationship('Repository')

    comment = relationship('ChangesetComment', lazy='joined')

    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__ = (

        {'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('pull_request_id', Integer(), nullable=False, primary_key=True)

    title = Column('title', Unicode(256), nullable=True)

    description = Column('description', UnicodeText(10240), nullable=True)

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

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

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

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

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

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

    org_ref = Column('org_ref', Unicode(256), nullable=False)

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

    other_ref = Column('other_ref', Unicode(256), nullable=False)

    @hybrid_property

    def revisions(self):

        return self._revisions.split(':')

    @revisions.setter

    def revisions(self, val):

    @property

    def org_ref_parts(self):

        return self.org_ref.split(':')

    @property

    def other_ref_parts(self):

        return self.other_ref.split(':')

    author = relationship('User', lazy='joined')

    reviewers = relationship('PullRequestReviewers',

                             cascade="all, delete, 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, delete-orphan")

    def is_closed(self):

        return self.status == self.STATUS_CLOSED

    @property

    def last_review_status(self):

        return str(self.statuses[-1].status) if self.statuses else ''

    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('pull_requests_reviewers_id', Integer(), nullable=False, primary_key=True)

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

    user_id = Column("user_id", 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('notification_id', Integer(), nullable=False, primary_key=True)

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

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

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

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

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

    created_by_user = relationship('User')

    notifications_to_users = relationship('UserNotification', lazy='joined',

                                          cascade="all, delete, 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()]

                import kallithea.lib.auth_modules

                from kallithea.lib.auth_modules import LazyFormencode

                for module in current_active_modules:

                    plugin = kallithea.lib.auth_modules.loadplugin(module)

                    plugin_name = plugin.name

                    for sv in plugin.plugin_settings():

                        newk = "auth_%s_%s" % (plugin_name, sv["name"])

                        # can be a LazyFormencode object from plugin settings

                        validator = sv["validator"]

                        if isinstance(validator, LazyFormencode):

                            validator = validator()

                        #init all lazy validators from formencode.All

                        if isinstance(validator, All):

                            init_validators = []

                            for validator in validator.validators:

                                if isinstance(validator, LazyFormencode):

                                    validator = validator()

                                init_validators.append(validator)

                            validator.validators = init_validators

                        self.add_field(newk, validator)

            formencode.Schema.__init__(self, *args, **kwargs)

    return _AuthSettingsForm

def LdapSettingsForm(tls_reqcert_choices, search_scope_choices,

                     tls_kind_choices):

    class _LdapSettingsForm(formencode.Schema):

        allow_extra_fields = True

        filter_extra_fields = True

        #pre_validators = [LdapLibValidator]

        ldap_active = v.StringBoolean(if_missing=False)

        ldap_host = v.UnicodeString(strip=True,)

        ldap_port = v.Number(strip=True,)

        ldap_tls_kind = v.OneOf(tls_kind_choices)

        ldap_tls_reqcert = v.OneOf(tls_reqcert_choices)

        ldap_dn_user = v.UnicodeString(strip=True,)

        ldap_dn_pass = v.UnicodeString(strip=True,)

        ldap_base_dn = v.UnicodeString(strip=True,)

        ldap_filter = v.UnicodeString(strip=True,)

        ldap_search_scope = v.OneOf(search_scope_choices)

        ldap_attr_login = v.AttrLoginValidator()(not_empty=True)

        ldap_attr_firstname = v.UnicodeString(strip=True,)

        ldap_attr_lastname = v.UnicodeString(strip=True,)

        ldap_attr_email = v.UnicodeString(strip=True,)

    return _LdapSettingsForm

def UserExtraEmailForm():

    class _UserExtraEmailForm(formencode.Schema):

        email = All(v.UniqSystemEmail(), v.Email(not_empty=True))

    return _UserExtraEmailForm

def UserExtraIpForm():

    class _UserExtraIpForm(formencode.Schema):

        ip = v.ValidIp()(not_empty=True)

    return _UserExtraIpForm

def PullRequestForm(repo_id):

    class _PullRequestForm(formencode.Schema):

        allow_extra_fields = True

        filter_extra_fields = True

        user = v.UnicodeString(strip=True, required=True)

        org_repo = v.UnicodeString(strip=True, required=True)

        org_ref = v.UnicodeString(strip=True, required=True)

        other_repo = v.UnicodeString(strip=True, required=True)

        other_ref = v.UnicodeString(strip=True, required=True)

        review_members = v.Set()

        pullrequest_title = v.UnicodeString(strip=True, required=True)

        pullrequest_desc = v.UnicodeString(strip=True, required=False)

    return _PullRequestForm

def PullRequestPostForm():

    class _PullRequestPostForm(formencode.Schema):

        allow_extra_fields = True

        filter_extra_fields = True

        pullrequest_title = v.UnicodeString(strip=True, required=True)

        pullrequest_desc = v.UnicodeString(strip=True, required=False)

    return _PullRequestPostForm

def GistForm(lifetime_options):

    class _GistForm(formencode.Schema):

        filename = All(v.BasePath()(),

                       v.UnicodeString(strip=True, required=False))

        lifetime = v.OneOf(lifetime_options)

        mimetype = v.UnicodeString(required=False, if_missing=None)

        content = v.UnicodeString(required=True, not_empty=True)

    return _GistForm

            if r:

                return r

        if isinstance(kwargs.get('repo_group'), RepoGroup):

            kwargs['repo_group'] = kwargs['repo_group'].group_id

        form_data = self._get_repo_create_params(repo_name=name, **kwargs)

        cur_user = kwargs.get('cur_user', TEST_USER_ADMIN_LOGIN)

        RepoModel().create(form_data, cur_user)

        Session().commit()

        return Repository.get_by_repo_name(name)

    def create_fork(self, repo_to_fork, fork_name, **kwargs):

        repo_to_fork = Repository.get_by_repo_name(repo_to_fork)

        form_data = self._get_repo_create_params(repo_name=fork_name,

                                            fork_parent_id=repo_to_fork,

                                            repo_type=repo_to_fork.repo_type,

                                            **kwargs)

        form_data['update_after_clone'] = False

        #TODO: fix it !!

        form_data['description'] = form_data['repo_description']

        form_data['private'] = form_data['repo_private']

        form_data['landing_rev'] = form_data['repo_landing_rev']

        owner = kwargs.get('cur_user', TEST_USER_ADMIN_LOGIN)

        RepoModel().create_fork(form_data, cur_user=owner)

        Session().commit()

        r = Repository.get_by_repo_name(fork_name)

        assert r

        return r

    def destroy_repo(self, repo_name, **kwargs):

        RepoModel().delete(repo_name, **kwargs)

        Session().commit()

    def create_repo_group(self, name, **kwargs):

        if 'skip_if_exists' in kwargs:

            del kwargs['skip_if_exists']

            gr = RepoGroup.get_by_group_name(group_name=name)

            if gr:

                return gr

        form_data = self._get_group_create_params(group_name=name, **kwargs)

        owner = kwargs.get('cur_user', TEST_USER_ADMIN_LOGIN)

        gr = RepoGroupModel().create(

            group_name=form_data['group_name'],

            group_description=form_data['group_name'],

            owner=owner, parent=form_data['group_parent_id'])

        Session().commit()

        gr = RepoGroup.get_by_group_name(gr.group_name)

        return gr

    def destroy_repo_group(self, repogroupid):

        RepoGroupModel().delete(repogroupid)

        Session().commit()

    def create_user(self, name, **kwargs):

        if 'skip_if_exists' in kwargs:

            del kwargs['skip_if_exists']

            user = User.get_by_username(name)

            if user:

                return user

        form_data = self._get_user_create_params(name, **kwargs)

        user = UserModel().create(form_data)

        Session().commit()

        user = User.get_by_username(user.username)

        return user

    def destroy_user(self, userid):

        UserModel().delete(userid)

        Session().commit()

    def create_user_group(self, name, **kwargs):

        if 'skip_if_exists' in kwargs:

            del kwargs['skip_if_exists']

            gr = UserGroup.get_by_group_name(group_name=name)

            if gr:

                return gr

        form_data = self._get_user_group_create_params(name, **kwargs)

        owner = kwargs.get('cur_user', TEST_USER_ADMIN_LOGIN)

        user_group = UserGroupModel().create(

            name=form_data['users_group_name'],

            description=form_data['user_group_description'],

            owner=owner, active=form_data['users_group_active'],

            group_data=form_data['user_group_data'])

        Session().commit()

        user_group = UserGroup.get_by_group_name(user_group.users_group_name)

        return user_group

    def destroy_user_group(self, usergroupid):

        UserGroupModel().delete(user_group=usergroupid, force=True)

        Session().commit()

    def create_gist(self, **kwargs):

        form_data = {

            'owner': TEST_USER_ADMIN_LOGIN,

            'gist_type': GistModel.cls.GIST_PUBLIC,

            'lifetime': -1,

            'gist_mapping': {'filename1.txt':{'content':'hello world'},}

        }

        form_data.update(kwargs)

        gist = GistModel().create(

            description=form_data['description'],owner=form_data['owner'],

            gist_mapping=form_data['gist_mapping'], gist_type=form_data['gist_type'],

            lifetime=form_data['lifetime']

        )

        Session().commit()

        return gist

    def destroy_gists(self, gistid=None):

        for g in GistModel.cls.get_all():

            if gistid:

                if gistid == g.gist_access_id:

                    GistModel().delete(g)

            else:

                GistModel().delete(g)

        Session().commit()

    def load_resource(self, resource_name, strip=True):

        with open(os.path.join(FIXTURES, resource_name)) as f:

            source = f.read()

            if strip:

                source = source.strip()

        return source

from kallithea.tests import *

from kallithea.model.gist import GistModel

from kallithea.model.meta import Session

from kallithea.model.db import User, Gist

def _create_gist(f_name, content='some gist', lifetime=-1,

                 owner=TEST_USER_ADMIN_LOGIN):

    gist_mapping = {

        f_name: {'content': content}

    }

    user = User.get_by_username(owner)

    gist = GistModel().create(description, owner=user,

                       gist_mapping=gist_mapping, gist_type=gist_type,

                       lifetime=lifetime)

    Session().commit()

    return gist

class TestGistsController(TestController):

    def tearDown(self):

        for g in Gist.get_all():

            GistModel().delete(g)

        Session().commit()

    def test_index(self):

        self.log_user()

        response = self.app.get(url('gists'))

        # Test response...

        response.mustcontain('There are no gists yet')

        g1 = _create_gist('gist1').gist_access_id

        g2 = _create_gist('gist2', lifetime=1400).gist_access_id

        g4 = _create_gist('gist4', gist_type='private').gist_access_id

        response = self.app.get(url('gists'))

        # Test response...

        response.mustcontain('gist: %s' % g1)

        response.mustcontain('gist: %s' % g2)

        response.mustcontain('Expires: in 23 hours')  # we don't care about the end

        response.mustcontain('gist: %s' % g3)

        response.mustcontain('gist3-desc')

        response.mustcontain(no=['gist: %s' % g4])

    def test_index_private_gists(self):

        self.log_user()

        gist = _create_gist('gist5', gist_type='private')

        response = self.app.get(url('gists', private=1))

        # Test response...

        #and privates

        response.mustcontain('gist: %s' % gist.gist_access_id)

    def test_create_missing_description(self):

        self.log_user()

        response = self.app.post(url('gists'),

                                 params={'lifetime': -1}, status=200)

        response.mustcontain('Missing value')

    def test_create(self):

        self.log_user()

        response = self.app.post(url('gists'),

                                 params={'lifetime': -1,

                                         'content': 'gist test',

                                         'filename': 'foo',

                                         'public': 'public'},

                                 status=302)

        response = response.follow()

        response.mustcontain('added file: foo')

        response.mustcontain('gist test')

        response.mustcontain('<div class="btn btn-mini btn-success disabled">Public Gist</div>')

    def test_create_with_path_with_dirs(self):

        self.log_user()

        response = self.app.post(url('gists'),

                                 params={'lifetime': -1,

                                         'content': 'gist test',

                                         'filename': '/home/foo',

                                         'public': 'public'},

                                 status=200)

        response.mustcontain('Filename cannot be inside a directory')

    def test_access_expired_gist(self):

        self.log_user()

        gist = _create_gist('never-see-me')

        gist.gist_expires = 0  # 1970

        Session().add(gist)

        Session().commit()

        response = self.app.get(url('gist', gist_id=gist.gist_access_id), status=404)

    def test_create_private(self):

        self.log_user()

        response = self.app.post(url('gists'),

                                 params={'lifetime': -1,

                                         'content': 'private gist test',

                                         'filename': 'private-foo',

                                         'private': 'private'},

                                 status=302)

        response = response.follow()

        response.mustcontain('added file: private-foo<')

        response.mustcontain('private gist test')

        response.mustcontain('<div class="btn btn-mini btn-warning disabled">Private Gist</div>')

    def test_create_with_description(self):

        self.log_user()

        response = self.app.post(url('gists'),

                                 params={'lifetime': -1,

                                         'content': 'gist test',

                                         'filename': 'foo-desc',

                                         'description': 'gist-desc',

                                         'public': 'public'},

                                 status=302)

        response = response.follow()

        response.mustcontain('added file: foo-desc')

        response.mustcontain('gist test')

        response.mustcontain('gist-desc')

        response.mustcontain('<div class="btn btn-mini btn-success disabled">Public Gist</div>')

    def test_new(self):

        self.log_user()

        response = self.app.get(url('new_gist'))

    def test_update(self):

        self.skipTest('not implemented')

        response = self.app.put(url('gist', gist_id=1))

    def test_delete(self):

        self.log_user()

        name = 'passwd'

        lastname = 'reset'

        new = User()

        new.username = username

        new.password = password

        new.email = email

        new.name = name

        new.lastname = lastname

        new.api_key = generate_api_key(username)

        Session().add(new)

        Session().commit()

        response = self.app.post(url(controller='login',

                                     action='password_reset'),

                                 {'email': email, })

        self.checkSessionFlash(response, 'Your password reset link was sent')

        response = response.follow()

        # BAD KEY

        key = "bad"

        response = self.app.get(url(controller='login',

                                    action='password_reset_confirmation',

                                    key=key))

        self.assertEqual(response.status, '302 Found')

        self.assertTrue(response.location.endswith(url('reset_password')))

        # GOOD KEY

        key = User.get_by_username(username).api_key

        response = self.app.get(url(controller='login',

                                    action='password_reset_confirmation',

                                    key=key))

        self.assertEqual(response.status, '302 Found')

        self.assertTrue(response.location.endswith(url('login_home')))

        self.checkSessionFlash(response,

                               ('Your password reset was successful, '

                                'new password has been sent to your email'))

        response = response.follow()

    def _get_api_whitelist(self, values=None):

        config = {'api_access_controllers_whitelist': values or []}

        return config

    @parameterized.expand([

        ('none', None),

        ('empty_string', ''),

        ('fake_number', '123456'),

        ('proper_api_key', None)

    ])

    def test_access_not_whitelisted_page_via_api_key(self, test_name, api_key):

        whitelist = self._get_api_whitelist([])

        with mock.patch('kallithea.CONFIG', whitelist):

            self.assertEqual([],

                             whitelist['api_access_controllers_whitelist'])

            if test_name == 'proper_api_key':

                #use builtin if api_key is None

                api_key = User.get_first_admin().api_key

            with fixture.anon_access(False):

                self.app.get(url(controller='changeset',

                                 action='changeset_raw',

                                 repo_name=HG_REPO, revision='tip', api_key=api_key),

                             status=302)

    @parameterized.expand([

        ('none', None, 302),

        ('empty_string', '', 302),

        ('fake_number', '123456', 302),

        ('proper_api_key', None, 200)

    ])

    def test_access_whitelisted_page_via_api_key(self, test_name, api_key, code):

        whitelist = self._get_api_whitelist(['ChangesetController:changeset_raw'])

        with mock.patch('kallithea.CONFIG', whitelist):

            self.assertEqual(['ChangesetController:changeset_raw'],

                             whitelist['api_access_controllers_whitelist'])

            if test_name == 'proper_api_key':

                api_key = User.get_first_admin().api_key

            with fixture.anon_access(False):

                self.app.get(url(controller='changeset',

                                 action='changeset_raw',

                                 repo_name=HG_REPO, revision='tip', api_key=api_key),

                             status=code)

    def test_access_page_via_extra_api_key(self):

        whitelist = self._get_api_whitelist(['ChangesetController:changeset_raw'])

        with mock.patch('kallithea.CONFIG', whitelist):

            self.assertEqual(['ChangesetController:changeset_raw'],

                             whitelist['api_access_controllers_whitelist'])

            Session().commit()

            with fixture.anon_access(False):

                self.app.get(url(controller='changeset',

                                 action='changeset_raw',

                                 repo_name=HG_REPO, revision='tip', api_key=new_api_key.api_key),

                             status=200)

    def test_access_page_via_expired_api_key(self):

        whitelist = self._get_api_whitelist(['ChangesetController:changeset_raw'])

        with mock.patch('kallithea.CONFIG', whitelist):

            self.assertEqual(['ChangesetController:changeset_raw'],

                             whitelist['api_access_controllers_whitelist'])

            Session().commit()

            #patch the api key and make it expired

            new_api_key.expires = 0

            Session().add(new_api_key)

            Session().commit()

            with fixture.anon_access(False):

                self.app.get(url(controller='changeset',

                                 action='changeset_raw',

                                 repo_name=HG_REPO, revision='tip',

                                 api_key=new_api_key.api_key),

                             status=302)