Files @ e1ab82613133
Branch filter:

Location: kallithea/kallithea/tests/functional/test_pullrequests.py

Alessandro Molina
backend: replace Pylons with TurboGears2

Replace the no-longer-supported Pylons application framework by TurboGears2
which is largely compatible/similar to Pylons.
Some interesting history is described at:
https://en.wikipedia.org/wiki/TurboGears

Changes by Dominik Ruf:
- fix sql config in test.ini

Changes by Thomas De Schampheleire:
- set-up of test suite
- tests: 'fix' repo archival test failure
Between Pylons and TurboGears2, there seems to be a small difference in the
headers sent for repository archive files, related to character encoding.
It is assumed that this difference is not important, and that the test
should just align with reality.
- remove need to import helpers/app_globals in lib
TurboGears2 by default expects helpers and app_globals to be available
in lib. For this reason kallithea/lib/__init__.py was originally changed
to include those files. However, this triggered several types of
circular import problems. If module A imported something from lib (e.g.
lib.annotate), and lib.helpers imported (possibly indirectly) module A,
then there was a circular import. Fix this by overruling the relevant
method of tg AppConfig, which is also hinted in the TurboGears2 code.
Hereby, the include of something from lib does not automatically import
helpers, greatly reducing the chances of circular import problems.
- make sure HTTP error '400' uses the custom error pages
TurboGears2 does not by default handle HTTP status code
'400 (Bad Request)' via the custom error page handling, causing a
standard non-styled error page.
- disable transaction manager
Kallithea currently handles its own transactions and does not need the
TurboGears2 transaction manager. However, TurboGears2 tries to enable it
by default and fails, throwing an error during application initialization.
The error itself seemed to be harmless for normal application functioning,
but was nevertheless confusing.
- add backlash as required dependency: backlash is meant as the WebError
replacement in TurboGears2 (originally WebError is part of Pylons). When
debug==true, it provides an interactive debugger in the browser. When
debug==false, backlash is necessary to show backtraces on the console.
- misc fixes
import re
import pytest

from tg.util.webtest import test_context

from kallithea.tests.base import *
from kallithea.tests.fixture import Fixture
from kallithea.model.db import User
from kallithea.model.meta import Session

from kallithea.controllers.pullrequests import PullrequestsController

fixture = Fixture()

class TestPullrequestsController(TestController):

    def test_index(self):
        self.log_user()
        response = self.app.get(url(controller='pullrequests', action='index',
                                    repo_name=HG_REPO))

    def test_create_trivial(self):
        self.log_user()
        response = self.app.post(url(controller='pullrequests', action='create',
                                     repo_name=HG_REPO),
                                 {'org_repo': HG_REPO,
                                  'org_ref': 'branch:stable:4f7e2131323e0749a740c0a56ab68ae9269c562a',
                                  'other_repo': HG_REPO,
                                  'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',
                                  'pullrequest_title': 'title',
                                  'pullrequest_desc': 'description',
                                  '_authentication_token': self.authentication_token(),
                                 },
                                 status=302)
        response = response.follow()
        assert response.status == '200 OK'
        response.mustcontain('Successfully opened new pull request')
        response.mustcontain('No additional changesets found for iterating on this pull request')
        response.mustcontain('href="/vcs_test_hg/changeset/4f7e2131323e0749a740c0a56ab68ae9269c562a"')

    def test_available(self):
        self.log_user()
        response = self.app.post(url(controller='pullrequests', action='create',
                                     repo_name=HG_REPO),
                                 {'org_repo': HG_REPO,
                                  'org_ref': 'rev:94f45ed825a1:94f45ed825a113e61af7e141f44ca578374abef0',
                                  'other_repo': HG_REPO,
                                  'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',
                                  'pullrequest_title': 'title',
                                  'pullrequest_desc': 'description',
                                  '_authentication_token': self.authentication_token(),
                                 },
                                 status=302)
        response = response.follow()
        assert response.status == '200 OK'
        response.mustcontain(no='No additional changesets found for iterating on this pull request')
        response.mustcontain('The following additional changes are available on stable:')
        response.mustcontain('<input id="updaterev_4f7e2131323e0749a740c0a56ab68ae9269c562a" name="updaterev" type="radio" value="4f7e2131323e0749a740c0a56ab68ae9269c562a" />')
        response.mustcontain('href="/vcs_test_hg/changeset/4f7e2131323e0749a740c0a56ab68ae9269c562a"') # as update

    def test_range(self):
        self.log_user()
        response = self.app.post(url(controller='pullrequests', action='create',
                                     repo_name=HG_REPO),
                                 {'org_repo': HG_REPO,
                                  'org_ref': 'branch:stable:4f7e2131323e0749a740c0a56ab68ae9269c562a',
                                  'other_repo': HG_REPO,
                                  'other_ref': 'rev:94f45ed825a1:94f45ed825a113e61af7e141f44ca578374abef0',
                                  'pullrequest_title': 'title',
                                  'pullrequest_desc': 'description',
                                  '_authentication_token': self.authentication_token(),
                                 },
                                 status=302)
        response = response.follow()
        assert response.status == '200 OK'
        response.mustcontain('No additional changesets found for iterating on this pull request')
        response.mustcontain('href="/vcs_test_hg/changeset/4f7e2131323e0749a740c0a56ab68ae9269c562a"')

    def test_update_reviewers(self):
        self.log_user()
        regular_user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
        regular_user2 = User.get_by_username(TEST_USER_REGULAR2_LOGIN)
        admin_user = User.get_by_username(TEST_USER_ADMIN_LOGIN)

        # create initial PR
        response = self.app.post(url(controller='pullrequests', action='create',
                                     repo_name=HG_REPO),
                                 {'org_repo': HG_REPO,
                                  'org_ref': 'rev:94f45ed825a1:94f45ed825a113e61af7e141f44ca578374abef0',
                                  'other_repo': HG_REPO,
                                  'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',
                                  'pullrequest_title': 'title',
                                  'pullrequest_desc': 'description',
                                  '_authentication_token': self.authentication_token(),
                                 },
                                 status=302)
        pull_request1_id = re.search('/pull-request/(\d+)/', response.location).group(1)
        assert response.location == 'http://localhost/%s/pull-request/%s/_/stable' % (HG_REPO, pull_request1_id)

        # create new iteration
        response = self.app.post(url(controller='pullrequests', action='post',
                                     repo_name=HG_REPO, pull_request_id=pull_request1_id),
                                 {
                                  'updaterev': '4f7e2131323e0749a740c0a56ab68ae9269c562a',
                                  'pullrequest_title': 'title',
                                  'pullrequest_desc': 'description',
                                  'owner': TEST_USER_ADMIN_LOGIN,
                                  '_authentication_token': self.authentication_token(),
                                  'review_members': [regular_user.user_id],
                                 },
                                 status=302)
        pull_request2_id = re.search('/pull-request/(\d+)/', response.location).group(1)
        assert pull_request2_id != pull_request1_id
        assert response.location == 'http://localhost/%s/pull-request/%s/_/stable' % (HG_REPO, pull_request2_id)
        response = response.follow()
        # verify reviewer was added
        response.mustcontain('<input type="hidden" value="%s" name="review_members" />' % regular_user.user_id)

        # update without creating new iteration
        response = self.app.post(url(controller='pullrequests', action='post',
                                     repo_name=HG_REPO, pull_request_id=pull_request2_id),
                                 {
                                  'pullrequest_title': 'Title',
                                  'pullrequest_desc': 'description',
                                  'owner': TEST_USER_ADMIN_LOGIN,
                                  '_authentication_token': self.authentication_token(),
                                  'org_review_members': [admin_user.user_id], # fake - just to get some 'meanwhile' warning ... but it is also added ...
                                  'review_members': [regular_user2.user_id, admin_user.user_id],
                                 },
                                 status=302)
        assert response.location == 'http://localhost/%s/pull-request/%s/_/stable' % (HG_REPO, pull_request2_id)
        response = response.follow()
        # verify reviewers were added / removed
        response.mustcontain('Meanwhile, the following reviewers have been added: test_regular')
        response.mustcontain('Meanwhile, the following reviewers have been removed: test_admin')
        response.mustcontain('<input type="hidden" value="%s" name="review_members" />' % regular_user.user_id)
        response.mustcontain('<input type="hidden" value="%s" name="review_members" />' % regular_user2.user_id)
        response.mustcontain(no='<input type="hidden" value="%s" name="review_members" />' % admin_user.user_id)

    def test_update_with_invalid_reviewer(self):
        invalid_user_id = 99999
        self.log_user()
        # create a valid pull request
        response = self.app.post(url(controller='pullrequests', action='create',
                                     repo_name=HG_REPO),
                                 {
                                  'org_repo': HG_REPO,
                                  'org_ref': 'rev:94f45ed825a1:94f45ed825a113e61af7e141f44ca578374abef0',
                                  'other_repo': HG_REPO,
                                  'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',
                                  'pullrequest_title': 'title',
                                  'pullrequest_desc': 'description',
                                  '_authentication_token': self.authentication_token(),
                                 },
                                status=302)
        # location is of the form:
        # http://localhost/vcs_test_hg/pull-request/54/_/title
        m = re.search('/pull-request/(\d+)/', response.location)
        assert m != None
        pull_request_id = m.group(1)

        # update it
        response = self.app.post(url(controller='pullrequests', action='post',
                                     repo_name=HG_REPO, pull_request_id=pull_request_id),
                                 {
                                  'updaterev': '4f7e2131323e0749a740c0a56ab68ae9269c562a',
                                  'pullrequest_title': 'title',
                                  'pullrequest_desc': 'description',
                                  'owner': TEST_USER_ADMIN_LOGIN,
                                  '_authentication_token': self.authentication_token(),
                                  'review_members': [str(invalid_user_id)],
                                 },
                                 status=400)
        response.mustcontain('Invalid reviewer &#34;%s&#34; specified' % invalid_user_id)

    def test_edit_with_invalid_reviewer(self):
        invalid_user_id = 99999
        self.log_user()
        # create a valid pull request
        response = self.app.post(url(controller='pullrequests', action='create',
                                     repo_name=HG_REPO),
                                 {
                                  'org_repo': HG_REPO,
                                  'org_ref': 'branch:stable:4f7e2131323e0749a740c0a56ab68ae9269c562a',
                                  'other_repo': HG_REPO,
                                  'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',
                                  'pullrequest_title': 'title',
                                  'pullrequest_desc': 'description',
                                  '_authentication_token': self.authentication_token(),
                                 },
                                status=302)
        # location is of the form:
        # http://localhost/vcs_test_hg/pull-request/54/_/title
        m = re.search('/pull-request/(\d+)/', response.location)
        assert m != None
        pull_request_id = m.group(1)

        # edit it
        response = self.app.post(url(controller='pullrequests', action='post',
                                     repo_name=HG_REPO, pull_request_id=pull_request_id),
                                 {
                                  'pullrequest_title': 'title',
                                  'pullrequest_desc': 'description',
                                  'owner': TEST_USER_ADMIN_LOGIN,
                                  '_authentication_token': self.authentication_token(),
                                  'review_members': [str(invalid_user_id)],
                                 },
                                 status=400)
        response.mustcontain('Invalid reviewer &#34;%s&#34; specified' % invalid_user_id)

@pytest.mark.usefixtures("test_context_fixture") # apply fixture for all test methods
class TestPullrequestsGetRepoRefs(TestController):

    def setup_method(self, method):
        self.repo_name = u'main'
        repo = fixture.create_repo(self.repo_name, repo_type='hg')
        self.repo_scm_instance = repo.scm_instance
        Session.commit()
        self.c = PullrequestsController()

    def teardown_method(self, method):
        fixture.destroy_repo(u'main')
        Session.commit()
        Session.remove()

    def test_repo_refs_empty_repo(self):
        # empty repo with no commits, no branches, no bookmarks, just one tag
        refs, default = self.c._get_repo_refs(self.repo_scm_instance)
        assert default == 'tag:null:0000000000000000000000000000000000000000'

    def test_repo_refs_one_commit_no_hints(self):
        cs0 = fixture.commit_change(self.repo_name, filename='file1',
                content='line1\n', message='commit1', vcs_type='hg',
                parent=None, newfile=True)

        refs, default = self.c._get_repo_refs(self.repo_scm_instance)
        assert default == 'branch:default:%s' % cs0.raw_id
        assert ([('branch:default:%s' % cs0.raw_id, 'default (current tip)')],
                'Branches') in refs

    def test_repo_refs_one_commit_rev_hint(self):
        cs0 = fixture.commit_change(self.repo_name, filename='file1',
                content='line1\n', message='commit1', vcs_type='hg',
                parent=None, newfile=True)

        refs, default = self.c._get_repo_refs(self.repo_scm_instance, rev=cs0.raw_id)
        expected = 'branch:default:%s' % cs0.raw_id
        assert default == expected
        assert ([(expected, 'default (current tip)')], 'Branches') in refs

    def test_repo_refs_two_commits_no_hints(self):
        cs0 = fixture.commit_change(self.repo_name, filename='file1',
                content='line1\n', message='commit1', vcs_type='hg',
                parent=None, newfile=True)
        cs1 = fixture.commit_change(self.repo_name, filename='file2',
                content='line2\n', message='commit2', vcs_type='hg',
                parent=None, newfile=True)

        refs, default = self.c._get_repo_refs(self.repo_scm_instance)
        expected = 'branch:default:%s' % cs1.raw_id
        assert default == expected
        assert ([(expected, 'default (current tip)')], 'Branches') in refs

    def test_repo_refs_two_commits_rev_hints(self):
        cs0 = fixture.commit_change(self.repo_name, filename='file1',
                content='line1\n', message='commit1', vcs_type='hg',
                parent=None, newfile=True)
        cs1 = fixture.commit_change(self.repo_name, filename='file2',
                content='line2\n', message='commit2', vcs_type='hg',
                parent=None, newfile=True)

        refs, default = self.c._get_repo_refs(self.repo_scm_instance, rev=cs0.raw_id)
        expected = 'rev:%s:%s' % (cs0.raw_id, cs0.raw_id)
        assert default == expected
        assert ([(expected, 'Changeset: %s' % cs0.raw_id[0:12])], 'Special') in refs
        assert ([('branch:default:%s' % cs1.raw_id, 'default (current tip)')], 'Branches') in refs

        refs, default = self.c._get_repo_refs(self.repo_scm_instance, rev=cs1.raw_id)
        expected = 'branch:default:%s' % cs1.raw_id
        assert default == expected
        assert ([(expected, 'default (current tip)')], 'Branches') in refs

    def test_repo_refs_two_commits_branch_hint(self):
        cs0 = fixture.commit_change(self.repo_name, filename='file1',
                content='line1\n', message='commit1', vcs_type='hg',
                parent=None, newfile=True)
        cs1 = fixture.commit_change(self.repo_name, filename='file2',
                content='line2\n', message='commit2', vcs_type='hg',
                parent=None, newfile=True)

        refs, default = self.c._get_repo_refs(self.repo_scm_instance, branch='default')
        expected = 'branch:default:%s' % cs1.raw_id
        assert default == expected
        assert ([(expected, 'default (current tip)')], 'Branches') in refs

    def test_repo_refs_one_branch_no_hints(self):
        cs0 = fixture.commit_change(self.repo_name, filename='file1',
                content='line1\n', message='commit1', vcs_type='hg',
                parent=None, newfile=True)
        # TODO