# 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 ''

        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

    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()

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)

        ('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)