kallithea Changeset - 15c40f8a3510

Changeset - 15c40f8a3510

Parent rev.

Child rev.

[Not reviewed]

default

0 8 0

Thomas De Schampheleire - 10 years ago 2016-04-26 17:35:13
thomas.de.schampheleire@gmail.com

pytest migration: convert functional tests with setup/teardown methods

pytest also supports setup/teardown methods like unittest (even though
pytest fixtures are more powerful and should be the end goal).
Only difference is the naming and signature of setUp (setup_method) and
tearDown (teardown_method).

8 files changed with 20 insertions and 23 deletions:

kallithea/tests/functional/test_admin.py

kallithea/tests/functional/test_admin_gists.py

kallithea/tests/functional/test_admin_notifications.py

kallithea/tests/functional/test_admin_repos.py

kallithea/tests/functional/test_changeset_comments.py

kallithea/tests/functional/test_compare.py

kallithea/tests/functional/test_forks.py

kallithea/tests/functional/test_pullrequests.py

0 comments (0 inline, 0 general)

kallithea/tests/functional/test_admin.py

➞

Show inline comments

 import os
 import csv
 import datetime
 from kallithea.tests import *
 from kallithea.model.db import UserLog
 from kallithea.model.meta import Session
 from kallithea.lib.utils2 import safe_unicode
 dn = os.path.dirname
 FIXTURES = os.path.join(dn(dn(os.path.abspath(__file__))), 'fixtures')
-class TestAdminController(TestController):
+class TestAdminController(TestControllerPytest):
     @classmethod
     def setup_class(cls):
         UserLog.query().delete()
         Session().commit()
         def strptime(val):
             fmt = '%Y-%m-%d %H:%M:%S'
             if '.' not in val:
                 return datetime.datetime.strptime(val, fmt)
             nofrag, frag = val.split(".")
             date = datetime.datetime.strptime(nofrag, fmt)
             frag = frag[:6]  # truncate to microseconds
             frag += (6 - len(frag)) * '0'  # add 0s
             return date.replace(microsecond=int(frag))
         with open(os.path.join(FIXTURES, 'journal_dump.csv')) as f:
             for row in csv.DictReader(f):
                 ul = UserLog()
                 for k, v in row.iteritems():
                     v = safe_unicode(v)
                     if k == 'action_date':
                         v = strptime(v)
                     if k in ['user_id', 'repository_id']:
                         # nullable due to FK problems
                         v = None
                     setattr(ul, k, v)
                 Session().add(ul)
             Session().commit()
     @classmethod
     def teardown_class(cls):
         UserLog.query().delete()
         Session().commit()
     def test_index(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index'))
         response.mustcontain('Admin Journal')
     def test_filter_all_entries(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',))
         response.mustcontain('2034 Entries')
     def test_filter_journal_filter_exact_match_on_repository(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',
                                     filter='repository:xxx'))
         response.mustcontain('3 Entries')
     def test_filter_journal_filter_exact_match_on_repository_CamelCase(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',
                                     filter='repository:XxX'))
         response.mustcontain('3 Entries')
     def test_filter_journal_filter_wildcard_on_repository(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',
                                     filter='repository:*test*'))
         response.mustcontain('862 Entries')
     def test_filter_journal_filter_prefix_on_repository(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',
                                     filter='repository:test*'))
         response.mustcontain('257 Entries')
     def test_filter_journal_filter_prefix_on_repository_CamelCase(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',
                                     filter='repository:Test*'))
         response.mustcontain('257 Entries')
     def test_filter_journal_filter_prefix_on_repository_and_user(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',
                                     filter='repository:test* AND username:demo'))
         response.mustcontain('130 Entries')
     def test_filter_journal_filter_prefix_on_repository_or_other_repo(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',
                                     filter='repository:test* OR repository:xxx'))
         response.mustcontain('260 Entries')  # 257 + 3
     def test_filter_journal_filter_exact_match_on_username(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',
                                     filter='username:demo'))
         response.mustcontain('1087 Entries')
     def test_filter_journal_filter_exact_match_on_username_camelCase(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',
                                     filter='username:DemO'))
         response.mustcontain('1087 Entries')
     def test_filter_journal_filter_wildcard_on_username(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',
                                     filter='username:*test*'))
         response.mustcontain('100 Entries')
     def test_filter_journal_filter_prefix_on_username(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',
                                     filter='username:demo*'))
         response.mustcontain('1101 Entries')
     def test_filter_journal_filter_prefix_on_user_or_other_user(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',
                                     filter='username:demo OR username:volcan'))
         response.mustcontain('1095 Entries')  # 1087 + 8
     def test_filter_journal_filter_wildcard_on_action(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',
                                     filter='action:*pull_request*'))
         response.mustcontain('187 Entries')
     def test_filter_journal_filter_on_date(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',
                                     filter='date:20121010'))
         response.mustcontain('47 Entries')
     def test_filter_journal_filter_on_date_2(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',
                                     filter='date:20121020'))
         response.mustcontain('17 Entries')

kallithea/tests/functional/test_admin_gists.py

➞

Show inline comments

 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,
                  description=u'gist-desc', gist_type='public',
                  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):
+class TestGistsController(TestControllerPytest):
-    def tearDown(self):
+    def teardown_method(self, method):
         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
         g3 = _create_gist('gist3', description=u'gist3-desc').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, '_authentication_token': self.authentication_token()},
                                  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',
                                          '_authentication_token': self.authentication_token()},
                                  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',
                                          '_authentication_token': self.authentication_token()},
                                  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',
                                          '_authentication_token': self.authentication_token()},
                                  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',
                                          '_authentication_token': self.authentication_token()},
                                  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'))
     # Functionality currently not implemented
     #def test_update(self):
     #    response = self.app.put(url('gist', gist_id=1))
     def test_delete(self):
         self.log_user()
         gist = _create_gist('delete-me')
         response = self.app.post(url('gist', gist_id=gist.gist_id),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
     def test_delete_normal_user_his_gist(self):
         self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
         gist = _create_gist('delete-me', owner=TEST_USER_REGULAR_LOGIN)
         response = self.app.post(url('gist', gist_id=gist.gist_id),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
     def test_delete_normal_user_not_his_own_gist(self):
         self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
         gist = _create_gist('delete-me')
         response = self.app.post(url('gist', gist_id=gist.gist_id), status=403,
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
     def test_show(self):
         gist = _create_gist('gist-show-me')
         response = self.app.get(url('gist', gist_id=gist.gist_access_id))
         response.mustcontain('added file: gist-show-me<')
         response.mustcontain('%s - created' % TEST_USER_ADMIN_LOGIN)
         response.mustcontain('gist-desc')
         response.mustcontain('<div class="btn btn-mini btn-success disabled">Public Gist</div>')
     def test_show_as_raw(self):
         gist = _create_gist('gist-show-me', content='GIST CONTENT')
         response = self.app.get(url('formatted_gist',
                                     gist_id=gist.gist_access_id, format='raw'))
         self.assertEqual(response.body, 'GIST CONTENT')
     def test_show_as_raw_individual_file(self):
         gist = _create_gist('gist-show-me-raw', content='GIST BODY')
         response = self.app.get(url('formatted_gist_file',
                                     gist_id=gist.gist_access_id, format='raw',
                                     revision='tip', f_path='gist-show-me-raw'))
         self.assertEqual(response.body, 'GIST BODY')
     def test_edit(self):
         response = self.app.get(url('edit_gist', gist_id=1))

kallithea/tests/functional/test_admin_notifications.py

➞

Show inline comments

 from kallithea.tests import *
 from kallithea.model.db import User
 from kallithea.model.user import UserModel
 from kallithea.model.notification import NotificationModel
 from kallithea.model.meta import Session
 from kallithea.lib import helpers as h
 class TestNotificationsController(TestController):
     def setUp(self):
 class TestNotificationsController(TestControllerPytest):
     def setup_method(self, method):
         remove_all_notifications()
     def test_index(self):
         self.log_user()
         u1 = UserModel().create_or_update(username='u1', password='qweqwe',
                                           email='u1@example.com',
                                           firstname=u'u1', lastname=u'u1')
         u1 = u1.user_id
         Session().commit()
         response = self.app.get(url('notifications'))
         response.mustcontain('<div class="table">No notifications here yet</div>')
         cur_user = self._get_logged_user()
         notif = NotificationModel().create(created_by=u1, subject=u'test_notification_1',
                                            body=u'notification_1', recipients=[cur_user])
         Session().commit()
         response = self.app.get(url('notifications'))
         response.mustcontain('id="notification_%s"' % notif.notification_id)
     def test_delete(self):
         self.log_user()
         cur_user = self._get_logged_user()
         u1 = UserModel().create_or_update(username='u1', password='qweqwe',
                                                email='u1@example.com',
                                                firstname=u'u1', lastname=u'u1')
         u2 = UserModel().create_or_update(username='u2', password='qweqwe',
                                                email='u2@example.com',
                                                firstname=u'u2', lastname=u'u2')
         # make notifications
         notification = NotificationModel().create(created_by=cur_user,
                                                   subject=u'test',
                                                   body=u'hi there',
                                                   recipients=[cur_user, u1, u2])
         Session().commit()
         u1 = User.get(u1.user_id)
         u2 = User.get(u2.user_id)
         # check DB
         get_notif = lambda un: [x.notification for x in un]
         self.assertEqual(get_notif(cur_user.notifications), [notification])
         self.assertEqual(get_notif(u1.notifications), [notification])
         self.assertEqual(get_notif(u2.notifications), [notification])
         cur_usr_id = cur_user.user_id
         response = self.app.post(
             url('notification', notification_id=notification.notification_id),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         self.assertEqual(response.body, 'ok')
         cur_user = User.get(cur_usr_id)
         self.assertEqual(cur_user.notifications, [])
     def test_show(self):
         self.log_user()
         cur_user = self._get_logged_user()
         u1 = UserModel().create_or_update(username='u1', password='qweqwe',
                                           email='u1@example.com',
                                           firstname=u'u1', lastname=u'u1')
         u2 = UserModel().create_or_update(username='u2', password='qweqwe',
                                           email='u2@example.com',
                                           firstname=u'u2', lastname=u'u2')
         subject = u'test'
         notif_body = u'hi there'
         notification = NotificationModel().create(created_by=cur_user,
                                                   subject=subject,
                                                   body=notif_body,
                                                   recipients=[cur_user, u1, u2])
         response = self.app.get(url('notification',
                                     notification_id=notification.notification_id))
         response.mustcontain(subject)
         response.mustcontain(notif_body)
     def test_description_with_age(self):
         self.log_user()
         cur_user = self._get_logged_user()
         subject = u'test'
         notify_body = u'hi there'
         notification = NotificationModel().create(created_by = cur_user,
                                                   subject    = subject,
                                                   body       = notify_body)
         description = NotificationModel().make_description(notification)
         self.assertEqual(
             description,
             "{0} sent message {1}".format(
                 cur_user.username,
                 h.age(notification.created_on)
+                )
+            )
     def test_description_with_datetime(self):
         self.log_user()
         cur_user = self._get_logged_user()
         subject = u'test'
         notify_body = u'hi there'
         notification = NotificationModel().create(created_by = cur_user,
                                                   subject    = subject,
                                                   body       = notify_body)
         description = NotificationModel().make_description(notification, False)
         self.assertEqual(
             description,
             "{0} sent message at {1}".format(
                 cur_user.username,
                 h.fmt_date(notification.created_on)
+                )
+            )

kallithea/tests/functional/test_admin_repos.py

➞

Show inline comments

@@ @@ -415,206 +415,206 @@ class _BaseTest(object): @@
         except vcs.exceptions.VCSError:
             pytest.fail('no repo %s in filesystem' % repo_name)
         response = self.app.post(url('delete_repo', repo_name=repo_name),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'Deleted repository %s' % (repo_name_unicode))
         response.follow()
         #check if repo was deleted from db
         deleted_repo = Session().query(Repository) \
             .filter(Repository.repo_name == repo_name_unicode).scalar()
         self.assertEqual(deleted_repo, None)
         self.assertEqual(os.path.isdir(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name_unicode)),
                                   False)
     def test_delete_repo_with_group(self):
         #TODO:
         pass
     def test_delete_browser_fakeout(self):
         response = self.app.post(url('delete_repo', repo_name=self.REPO),
                                  params=dict(_method='delete', _authentication_token=self.authentication_token()))
     def test_show(self):
         self.log_user()
         response = self.app.get(url('summary_home', repo_name=self.REPO))
     def test_edit(self):
         response = self.app.get(url('edit_repo', repo_name=self.REPO))
     def test_set_private_flag_sets_default_to_none(self):
         self.log_user()
         #initially repository perm should be read
         perm = _get_permission_for_user(user='default', repo=self.REPO)
         self.assertTrue(len(perm), 1)
         self.assertEqual(perm[0].permission.permission_name, 'repository.read')
         self.assertEqual(Repository.get_by_repo_name(self.REPO).private, False)
         response = self.app.put(url('put_repo', repo_name=self.REPO),
                         fixture._get_repo_create_params(repo_private=1,
                                                 repo_name=self.REPO,
                                                 repo_type=self.REPO_TYPE,
                                                 user=TEST_USER_ADMIN_LOGIN,
                                                 _authentication_token=self.authentication_token()))
         self.checkSessionFlash(response,
                                msg='Repository %s updated successfully' % (self.REPO))
         self.assertEqual(Repository.get_by_repo_name(self.REPO).private, True)
         #now the repo default permission should be None
         perm = _get_permission_for_user(user='default', repo=self.REPO)
         self.assertTrue(len(perm), 1)
         self.assertEqual(perm[0].permission.permission_name, 'repository.none')
         response = self.app.put(url('put_repo', repo_name=self.REPO),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=self.REPO,
                                                 repo_type=self.REPO_TYPE,
                                                 user=TEST_USER_ADMIN_LOGIN,
                                                 _authentication_token=self.authentication_token()))
         self.checkSessionFlash(response,
                                msg='Repository %s updated successfully' % (self.REPO))
         self.assertEqual(Repository.get_by_repo_name(self.REPO).private, False)
         #we turn off private now the repo default permission should stay None
         perm = _get_permission_for_user(user='default', repo=self.REPO)
         self.assertTrue(len(perm), 1)
         self.assertEqual(perm[0].permission.permission_name, 'repository.none')
         #update this permission back
         perm[0].permission = Permission.get_by_key('repository.read')
         Session().add(perm[0])
         Session().commit()
     def test_set_repo_fork_has_no_self_id(self):
         self.log_user()
         repo = Repository.get_by_repo_name(self.REPO)
         response = self.app.get(url('edit_repo_advanced', repo_name=self.REPO))
         opt = """<option value="%s">%s</option>""" % (repo.repo_id, self.REPO)
         response.mustcontain(no=[opt])
     def test_set_fork_of_other_repo(self):
         self.log_user()
         other_repo = u'other_%s' % self.REPO_TYPE
         fixture.create_repo(other_repo, repo_type=self.REPO_TYPE)
         repo = Repository.get_by_repo_name(self.REPO)
         repo2 = Repository.get_by_repo_name(other_repo)
         response = self.app.put(url('edit_repo_advanced_fork', repo_name=self.REPO),
                                 params=dict(id_fork_of=repo2.repo_id, _authentication_token=self.authentication_token()))
         repo = Repository.get_by_repo_name(self.REPO)
         repo2 = Repository.get_by_repo_name(other_repo)
         self.checkSessionFlash(response,
             'Marked repository %s as fork of %s' % (repo.repo_name, repo2.repo_name))
         assert repo.fork == repo2
         response = response.follow()
         # check if given repo is selected
         opt = """<option value="%s" selected="selected">%s</option>""" % (
                     repo2.repo_id, repo2.repo_name)
         response.mustcontain(opt)
         fixture.destroy_repo(other_repo, forks='detach')
     def test_set_fork_of_other_type_repo(self):
         self.log_user()
         repo = Repository.get_by_repo_name(self.REPO)
         repo2 = Repository.get_by_repo_name(self.OTHER_TYPE_REPO)
         response = self.app.put(url('edit_repo_advanced_fork', repo_name=self.REPO),
                                 params=dict(id_fork_of=repo2.repo_id, _authentication_token=self.authentication_token()))
         repo = Repository.get_by_repo_name(self.REPO)
         repo2 = Repository.get_by_repo_name(self.OTHER_TYPE_REPO)
         self.checkSessionFlash(response,
             'Cannot set repository as fork of repository with other type')
     def test_set_fork_of_none(self):
         self.log_user()
         ## mark it as None
         response = self.app.put(url('edit_repo_advanced_fork', repo_name=self.REPO),
                                 params=dict(id_fork_of=None, _authentication_token=self.authentication_token()))
         repo = Repository.get_by_repo_name(self.REPO)
         repo2 = Repository.get_by_repo_name(self.OTHER_TYPE_REPO)
         self.checkSessionFlash(response,
                                'Marked repository %s as fork of %s'
                                % (repo.repo_name, "Nothing"))
         assert repo.fork is None
     def test_set_fork_of_same_repo(self):
         self.log_user()
         repo = Repository.get_by_repo_name(self.REPO)
         response = self.app.put(url('edit_repo_advanced_fork', repo_name=self.REPO),
                                 params=dict(id_fork_of=repo.repo_id, _authentication_token=self.authentication_token()))
         self.checkSessionFlash(response,
                                'An error occurred during this operation')
     def test_create_on_top_level_without_permissions(self):
         usr = self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
         # revoke
         user_model = UserModel()
         # disable fork and create on default user
         user_model.revoke_perm(User.DEFAULT_USER, 'hg.create.repository')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.create.none')
         user_model.revoke_perm(User.DEFAULT_USER, 'hg.fork.repository')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.fork.none')
         # disable on regular user
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.none')
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.none')
         Session().commit()
         user = User.get(usr['user_id'])
         repo_name = self.NEW_REPO + u'no_perms'
         description = 'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 _authentication_token=self.authentication_token()))
         response.mustcontain('<span class="error-message">Invalid value</span>')
         RepoModel().delete(repo_name)
         Session().commit()
     @mock.patch.object(RepoModel, '_create_filesystem_repo', error_function)
     def test_create_repo_when_filesystem_op_fails(self):
         self.log_user()
         repo_name = self.NEW_REPO
         description = 'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 _authentication_token=self.authentication_token()))
         self.checkSessionFlash(response,
                                'Error creating repository %s' % repo_name)
         # repo must not be in db
         repo = Repository.get_by_repo_name(repo_name)
         self.assertEqual(repo, None)
         # repo must not be in filesystem !
         self.assertFalse(os.path.isdir(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name)))
-class TestAdminReposControllerGIT(TestController, _BaseTest):
+class TestAdminReposControllerGIT(TestControllerPytest, _BaseTest):
     REPO = GIT_REPO
     REPO_TYPE = 'git'
     NEW_REPO = NEW_GIT_REPO
     OTHER_TYPE_REPO = HG_REPO
     OTHER_TYPE = 'hg'
-class TestAdminReposControllerHG(TestController, _BaseTest):
+class TestAdminReposControllerHG(TestControllerPytest, _BaseTest):
     REPO = HG_REPO
     REPO_TYPE = 'hg'
     NEW_REPO = NEW_HG_REPO
     OTHER_TYPE_REPO = GIT_REPO
     OTHER_TYPE = 'git'

kallithea/tests/functional/test_changeset_comments.py

➞

Show inline comments

 from kallithea.tests import *
 from kallithea.model.db import ChangesetComment, Notification, \
     UserNotification
 from kallithea.model.meta import Session
-class TestChangeSetCommentsController(TestController):
+class TestChangeSetCommentsController(TestControllerPytest):
-    def setUp(self):
+    def setup_method(self, method):
         for x in ChangesetComment.query().all():
             Session().delete(x)
         Session().commit()
         remove_all_notifications()
     def test_create(self):
         self.log_user()
         rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
         text = u'CommentOnRevision'
         params = {'text': text, '_authentication_token': self.authentication_token()}
         response = self.app.post(url(controller='changeset', action='comment',
                                      repo_name=HG_REPO, revision=rev),
                                      params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         # Test response...
         self.assertEqual(response.status, '200 OK')
         response = self.app.get(url(controller='changeset', action='index',
                                 repo_name=HG_REPO, revision=rev))
         # test DB
         self.assertEqual(ChangesetComment.query().count(), 1)
         response.mustcontain(
             '''<div class="comments-number">'''
             ''' 1 comment (0 inline, 1 general)'''
+        )
         self.assertEqual(Notification.query().count(), 1)
         self.assertEqual(ChangesetComment.query().count(), 1)
         notification = Notification.query().all()[0]
         ID = ChangesetComment.query().first().comment_id
         self.assertEqual(notification.type_,
                          Notification.TYPE_CHANGESET_COMMENT)
         sbj = (u'/%s/changeset/'
                '27cd5cce30c96924232dffcd24178a07ffeb5dfc#comment-%s' % (HG_REPO, ID))
         print "%s vs %s" % (sbj, notification.subject)
         self.assertTrue(sbj in notification.subject)
     def test_create_inline(self):
         self.log_user()
         rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
         text = u'CommentOnRevision'
         f_path = 'vcs/web/simplevcs/views/repository.py'
         line = 'n1'
         params = {'text': text, 'f_path': f_path, 'line': line, '_authentication_token': self.authentication_token()}
         response = self.app.post(url(controller='changeset', action='comment',
                                      repo_name=HG_REPO, revision=rev),
                                      params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         # Test response...
         self.assertEqual(response.status, '200 OK')
         response = self.app.get(url(controller='changeset', action='index',
                                 repo_name=HG_REPO, revision=rev))
         #test DB
         self.assertEqual(ChangesetComment.query().count(), 1)
         response.mustcontain(
             '''<div class="comments-number">'''
             ''' 1 comment (1 inline, 0 general)'''
+        )
         response.mustcontain(
             '''<div class="comments-list-chunk" '''
             '''data-f_path="vcs/web/simplevcs/views/repository.py" '''
             '''data-line_no="n1" data-target-id="vcswebsimplevcsviewsrepositorypy_n1">'''
+        )
         self.assertEqual(Notification.query().count(), 1)
         self.assertEqual(ChangesetComment.query().count(), 1)
         notification = Notification.query().all()[0]
         ID = ChangesetComment.query().first().comment_id
         self.assertEqual(notification.type_,
                          Notification.TYPE_CHANGESET_COMMENT)
         sbj = (u'/%s/changeset/'
                '27cd5cce30c96924232dffcd24178a07ffeb5dfc#comment-%s' % (HG_REPO, ID))
         print "%s vs %s" % (sbj, notification.subject)
         self.assertTrue(sbj in notification.subject)
     def test_create_with_mention(self):
         self.log_user()
         rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
         text = u'@%s check CommentOnRevision' % TEST_USER_REGULAR_LOGIN
         params = {'text': text, '_authentication_token': self.authentication_token()}
         response = self.app.post(url(controller='changeset', action='comment',
                                      repo_name=HG_REPO, revision=rev),
                                      params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         # Test response...
         self.assertEqual(response.status, '200 OK')
         response = self.app.get(url(controller='changeset', action='index',
                                 repo_name=HG_REPO, revision=rev))
         # test DB
         self.assertEqual(ChangesetComment.query().count(), 1)
         response.mustcontain(
             '''<div class="comments-number">'''
             ''' 1 comment (0 inline, 1 general)'''
+        )
         self.assertEqual(Notification.query().count(), 2)
         users = [x.user.username for x in UserNotification.query().all()]
         # test_regular gets notification by @mention
         self.assertEqual(sorted(users), [TEST_USER_ADMIN_LOGIN, TEST_USER_REGULAR_LOGIN])
     def test_delete(self):
         self.log_user()
         rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
         text = u'CommentOnRevision'
         params = {'text': text, '_authentication_token': self.authentication_token()}
         response = self.app.post(url(controller='changeset', action='comment',
                                      repo_name=HG_REPO, revision=rev),
                                      params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         comments = ChangesetComment.query().all()
         self.assertEqual(len(comments), 1)
         comment_id = comments[0].comment_id
         self.app.post(url(controller='changeset',
                                     action='delete_comment',
                                     repo_name=HG_REPO,
                                     comment_id=comment_id),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         comments = ChangesetComment.query().all()
         self.assertEqual(len(comments), 0)
         response = self.app.get(url(controller='changeset', action='index',
                                 repo_name=HG_REPO, revision=rev))
         response.mustcontain(
             '''<div class="comments-number">'''
             ''' 0 comments (0 inline, 0 general)'''
+        )

kallithea/tests/functional/test_compare.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 from kallithea.tests import *
 from kallithea.model.repo import RepoModel
 from kallithea.model.meta import Session
 from kallithea.tests.fixture import Fixture
 fixture = Fixture()
 def _commit_div(sha, msg):
     return """<div id="C-%s" class="message">%s</div>""" % (sha, msg)
-class TestCompareController(TestController):
+class TestCompareController(TestControllerPytest):
-    def setUp(self):
+    def setup_method(self, method):
         self.r1_id = None
         self.r2_id = None
-    def tearDown(self):
+    def teardown_method(self, method):
         if self.r2_id:
             RepoModel().delete(self.r2_id)
         if self.r1_id:
             RepoModel().delete(self.r1_id)
         Session().commit()
         Session.remove()
     def test_compare_forks_on_branch_extra_commits_hg(self):
         self.log_user()
         repo1 = fixture.create_repo(u'one', repo_type='hg',
                                     repo_description='diff-test',
                                     cur_user=TEST_USER_ADMIN_LOGIN)
         self.r1_id = repo1.repo_id
         #commit something !
         cs0 = fixture.commit_change(repo1.repo_name, filename='file1',
                 content='line1\n', message='commit1', vcs_type='hg',
                 parent=None, newfile=True)
         #fork this repo
         repo2 = fixture.create_fork(u'one', u'one-fork')
         self.r2_id = repo2.repo_id
         #add two extra commit into fork
         cs1 = fixture.commit_change(repo2.repo_name, filename='file1',
                 content='line1\nline2\n', message='commit2', vcs_type='hg',
                 parent=cs0)
         cs2 = fixture.commit_change(repo2.repo_name, filename='file1',
                 content='line1\nline2\nline3\n', message='commit3',
                 vcs_type='hg', parent=cs1)
         rev1 = 'default'
         rev2 = 'default'
         response = self.app.get(url('compare_url',
                                     repo_name=repo1.repo_name,
                                     org_ref_type="branch",
                                     org_ref_name=rev2,
                                     other_repo=repo2.repo_name,
                                     other_ref_type="branch",
                                     other_ref_name=rev1,
                                     merge='1',))
         response.mustcontain('%s@%s' % (repo1.repo_name, rev2))
         response.mustcontain('%s@%s' % (repo2.repo_name, rev1))
         response.mustcontain("""Showing 2 commits""")
         response.mustcontain("""1 file changed with 2 insertions and 0 deletions""")
         response.mustcontain(_commit_div(cs1.raw_id, 'commit2'))
         response.mustcontain(_commit_div(cs2.raw_id, 'commit3'))
         response.mustcontain("""<a href="/%s/changeset/%s">r1:%s</a>""" % (repo2.repo_name, cs1.raw_id, cs1.short_id))
         response.mustcontain("""<a href="/%s/changeset/%s">r2:%s</a>""" % (repo2.repo_name, cs2.raw_id, cs2.short_id))
         ## files
         response.mustcontain("""<a href="#C--826e8142e6ba">file1</a>""")
         #swap
         response.mustcontain("""<a class="btn btn-small" href="/%s/compare/branch@%s...branch@%s?other_repo=%s&amp;merge=True"><i class="icon-arrows-cw"></i> Swap</a>""" % (repo2.repo_name, rev1, rev2, repo1.repo_name))
     def test_compare_forks_on_branch_extra_commits_git(self):
         self.log_user()
         repo1 = fixture.create_repo(u'one-git', repo_type='git',
                                     repo_description='diff-test',
                                     cur_user=TEST_USER_ADMIN_LOGIN)
         self.r1_id = repo1.repo_id
         #commit something !
         cs0 = fixture.commit_change(repo1.repo_name, filename='file1',
                 content='line1\n', message='commit1', vcs_type='git',
                 parent=None, newfile=True)
         #fork this repo
         repo2 = fixture.create_fork(u'one-git', u'one-git-fork')
         self.r2_id = repo2.repo_id
         #add two extra commit into fork
         cs1 = fixture.commit_change(repo2.repo_name, filename='file1',
                 content='line1\nline2\n', message='commit2', vcs_type='git',
                 parent=cs0)
         cs2 = fixture.commit_change(repo2.repo_name, filename='file1',
                 content='line1\nline2\nline3\n', message='commit3',
                 vcs_type='git', parent=cs1)
         rev1 = 'master'
         rev2 = 'master'
         response = self.app.get(url('compare_url',
                                     repo_name=repo1.repo_name,
                                     org_ref_type="branch",
                                     org_ref_name=rev2,
                                     other_repo=repo2.repo_name,
                                     other_ref_type="branch",
                                     other_ref_name=rev1,
                                     merge='1',))
         response.mustcontain('%s@%s' % (repo1.repo_name, rev2))
         response.mustcontain('%s@%s' % (repo2.repo_name, rev1))
         response.mustcontain("""Showing 2 commits""")
         response.mustcontain("""1 file changed with 2 insertions and 0 deletions""")
         response.mustcontain(_commit_div(cs1.raw_id, 'commit2'))
         response.mustcontain(_commit_div(cs2.raw_id, 'commit3'))
         response.mustcontain("""<a href="/%s/changeset/%s">r1:%s</a>""" % (repo2.repo_name, cs1.raw_id, cs1.short_id))
         response.mustcontain("""<a href="/%s/changeset/%s">r2:%s</a>""" % (repo2.repo_name, cs2.raw_id, cs2.short_id))
         ## files
         response.mustcontain("""<a href="#C--826e8142e6ba">file1</a>""")
         #swap
         response.mustcontain("""<a class="btn btn-small" href="/%s/compare/branch@%s...branch@%s?other_repo=%s&amp;merge=True"><i class="icon-arrows-cw"></i> Swap</a>""" % (repo2.repo_name, rev1, rev2, repo1.repo_name))
     def test_compare_forks_on_branch_extra_commits_origin_has_incomming_hg(self):
         self.log_user()
         repo1 = fixture.create_repo(u'one', repo_type='hg',
                                     repo_description='diff-test',
                                     cur_user=TEST_USER_ADMIN_LOGIN)
         self.r1_id = repo1.repo_id
         #commit something !
         cs0 = fixture.commit_change(repo1.repo_name, filename='file1',
                 content='line1\n', message='commit1', vcs_type='hg',
                 parent=None, newfile=True)
         #fork this repo
         repo2 = fixture.create_fork(u'one', u'one-fork')
         self.r2_id = repo2.repo_id
         #now commit something to origin repo
         cs1_prim = fixture.commit_change(repo1.repo_name, filename='file2',
                 content='line1file2\n', message='commit2', vcs_type='hg',
                 parent=cs0, newfile=True)
         #add two extra commit into fork
         cs1 = fixture.commit_change(repo2.repo_name, filename='file1',
                 content='line1\nline2\n', message='commit2', vcs_type='hg',
                 parent=cs0)
         cs2 = fixture.commit_change(repo2.repo_name, filename='file1',
                 content='line1\nline2\nline3\n', message='commit3',
                 vcs_type='hg', parent=cs1)
         rev1 = 'default'
         rev2 = 'default'
         response = self.app.get(url('compare_url',
                                     repo_name=repo1.repo_name,
                                     org_ref_type="branch",
                                     org_ref_name=rev2,
                                     other_repo=repo2.repo_name,
                                     other_ref_type="branch",
                                     other_ref_name=rev1,
                                     merge='1',))
         response.mustcontain('%s@%s' % (repo1.repo_name, rev2))
         response.mustcontain('%s@%s' % (repo2.repo_name, rev1))
         response.mustcontain("""Showing 2 commits""")
         response.mustcontain("""1 file changed with 2 insertions and 0 deletions""")
         response.mustcontain(_commit_div(cs1.raw_id, 'commit2'))
         response.mustcontain(_commit_div(cs2.raw_id, 'commit3'))
         response.mustcontain("""<a href="/%s/changeset/%s">r1:%s</a>""" % (repo2.repo_name, cs1.raw_id, cs1.short_id))
         response.mustcontain("""<a href="/%s/changeset/%s">r2:%s</a>""" % (repo2.repo_name, cs2.raw_id, cs2.short_id))
         ## files
         response.mustcontain("""<a href="#C--826e8142e6ba">file1</a>""")
         #swap
         response.mustcontain("""<a class="btn btn-small" href="/%s/compare/branch@%s...branch@%s?other_repo=%s&amp;merge=True"><i class="icon-arrows-cw"></i> Swap</a>""" % (repo2.repo_name, rev1, rev2, repo1.repo_name))
     def test_compare_forks_on_branch_extra_commits_origin_has_incomming_git(self):
         self.log_user()
         repo1 = fixture.create_repo(u'one-git', repo_type='git',
                                     repo_description='diff-test',
                                     cur_user=TEST_USER_ADMIN_LOGIN)
         self.r1_id = repo1.repo_id
         #commit something !
         cs0 = fixture.commit_change(repo1.repo_name, filename='file1',
                 content='line1\n', message='commit1', vcs_type='git',
                 parent=None, newfile=True)
         #fork this repo
         repo2 = fixture.create_fork(u'one-git', u'one-git-fork')
         self.r2_id = repo2.repo_id
         #now commit something to origin repo
         cs1_prim = fixture.commit_change(repo1.repo_name, filename='file2',
                 content='line1file2\n', message='commit2', vcs_type='git',
                 parent=cs0, newfile=True)
         #add two extra commit into fork

kallithea/tests/functional/test_forks.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 import unittest
 from kallithea.tests import *
 from kallithea.tests.fixture import Fixture
 from kallithea.model.db import Repository
 from kallithea.model.repo import RepoModel
 from kallithea.model.user import UserModel
 from kallithea.model.meta import Session
 fixture = Fixture()
 from kallithea.tests import *
 class _BaseFixture(unittest.TestCase):
 class _BaseFixture(object):
     @classmethod
     def setup_class(cls):
         pass
     @classmethod
     def teardown_class(cls):
         pass
-    def setUp(self):
+    def setup_method(self, method):
         self.username = u'forkuser'
         self.password = u'qweqwe'
         self.u1 = fixture.create_user(self.username, password=self.password,
                                       email=u'fork_king@example.com')
         Session().commit()
-    def tearDown(self):
+    def teardown_method(self, method):
         Session().delete(self.u1)
         Session().commit()
 class _BaseTestCase(object):
     """
     Write all tests here
     """
     REPO = None
     REPO_TYPE = None
     NEW_REPO = None
     REPO_FORK = None
     def test_index(self):
         self.log_user()
         repo_name = self.REPO
         response = self.app.get(url(controller='forks', action='forks',
                                     repo_name=repo_name))
         response.mustcontain("""There are no forks yet""")
     def test_no_permissions_to_fork(self):
         usr = self.log_user(TEST_USER_REGULAR_LOGIN,
                             TEST_USER_REGULAR_PASS)['user_id']
         user_model = UserModel()
         user_model.revoke_perm(usr, 'hg.fork.repository')
         user_model.grant_perm(usr, 'hg.fork.none')
         u = UserModel().get(usr)
         u.inherit_default_permissions = False
         Session().commit()
         # try create a fork
         repo_name = self.REPO
         self.app.post(url(controller='forks', action='fork_create',
                           repo_name=repo_name), {'_authentication_token': self.authentication_token()}, status=403)
     def test_index_with_fork(self):
         self.log_user()
         # create a fork
         fork_name = self.REPO_FORK
         description = 'fork of vcs test'
         repo_name = self.REPO
         org_repo = Repository.get_by_repo_name(repo_name)
         creation_args = {
             'repo_name': fork_name,
             'repo_group': u'-1',
             'fork_parent_id': org_repo.repo_id,
             'repo_type': self.REPO_TYPE,
             'description': description,
             'private': 'False',
             'landing_rev': 'rev:tip',
             '_authentication_token': self.authentication_token()}
         self.app.post(url(controller='forks', action='fork_create',
                           repo_name=repo_name), creation_args)
         response = self.app.get(url(controller='forks', action='forks',
                                     repo_name=repo_name))
         response.mustcontain(
             """<a href="/%s">%s</a>""" % (fork_name, fork_name)
+        )
         # remove this fork
         response = self.app.post(url('delete_repo', repo_name=fork_name),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
     def test_fork_create_into_group(self):
         self.log_user()
         group = fixture.create_repo_group(u'vc')
         group_id = group.group_id
         fork_name = self.REPO_FORK
         fork_name_full = 'vc/%s' % fork_name
         description = 'fork of vcs test'
         repo_name = self.REPO
         org_repo = Repository.get_by_repo_name(repo_name)
         creation_args = {
             'repo_name': fork_name,
             'repo_group': group_id,
             'fork_parent_id': org_repo.repo_id,
             'repo_type': self.REPO_TYPE,
             'description': description,
             'private': 'False',
             'landing_rev': 'rev:tip',
             '_authentication_token': self.authentication_token()}
         self.app.post(url(controller='forks', action='fork_create',
                           repo_name=repo_name), creation_args)
         repo = Repository.get_by_repo_name(fork_name_full)
         assert repo.fork.repo_name == self.REPO
         ## run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=fork_name_full))
         #test if we have a message that fork is ok
         self.checkSessionFlash(response,
                 'Forked repository %s as <a href="/%s">%s</a>'
                 % (repo_name, fork_name_full, fork_name_full))
         #test if the fork was created in the database
         fork_repo = Session().query(Repository) \
             .filter(Repository.repo_name == fork_name_full).one()
         self.assertEqual(fork_repo.repo_name, fork_name_full)
         self.assertEqual(fork_repo.fork.repo_name, repo_name)
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=fork_name_full))
         response.mustcontain(fork_name_full)
         response.mustcontain(self.REPO_TYPE)
         response.mustcontain('Fork of "<a href="/%s">%s</a>"' % (repo_name, repo_name))
         fixture.destroy_repo(fork_name_full)
         fixture.destroy_repo_group(group_id)
     def test_z_fork_create(self):
         self.log_user()
         fork_name = self.REPO_FORK
         description = 'fork of vcs test'
         repo_name = self.REPO
         org_repo = Repository.get_by_repo_name(repo_name)
         creation_args = {
             'repo_name': fork_name,
             'repo_group': u'-1',
             'fork_parent_id': org_repo.repo_id,
             'repo_type': self.REPO_TYPE,
             'description': description,
             'private': 'False',
             'landing_rev': 'rev:tip',
             '_authentication_token': self.authentication_token()}
         self.app.post(url(controller='forks', action='fork_create',
                           repo_name=repo_name), creation_args)
         repo = Repository.get_by_repo_name(self.REPO_FORK)
         assert repo.fork.repo_name == self.REPO
         ## run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=fork_name))
         #test if we have a message that fork is ok
         self.checkSessionFlash(response,
                 'Forked repository %s as <a href="/%s">%s</a>'
                 % (repo_name, fork_name, fork_name))
         #test if the fork was created in the database
         fork_repo = Session().query(Repository) \
             .filter(Repository.repo_name == fork_name).one()
         self.assertEqual(fork_repo.repo_name, fork_name)
         self.assertEqual(fork_repo.fork.repo_name, repo_name)
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=fork_name))
         response.mustcontain(fork_name)
         response.mustcontain(self.REPO_TYPE)
         response.mustcontain('Fork of "<a href="/%s">%s</a>"' % (repo_name, repo_name))
     def test_zz_fork_permission_page(self):
         usr = self.log_user(self.username, self.password)['user_id']
         repo_name = self.REPO
         forks = Repository.query() \
             .filter(Repository.repo_type == self.REPO_TYPE) \
             .filter(Repository.fork_id != None).all()
         self.assertEqual(1, len(forks))
         # set read permissions for this
         RepoModel().grant_user_permission(repo=forks[0],
                                           user=usr,
                                           perm='repository.read')
         Session().commit()
         response = self.app.get(url(controller='forks', action='forks',
                                     repo_name=repo_name))
         response.mustcontain('<div style="padding:5px 3px 3px 42px;">fork of vcs test</div>')
     def test_zzz_fork_permission_page(self):
         usr = self.log_user(self.username, self.password)['user_id']
         repo_name = self.REPO
         forks = Repository.query() \
             .filter(Repository.repo_type == self.REPO_TYPE) \
             .filter(Repository.fork_id != None).all()
         self.assertEqual(1, len(forks))
         # set none
         RepoModel().grant_user_permission(repo=forks[0],
                                           user=usr, perm='repository.none')
         Session().commit()
         # fork shouldn't be there
         response = self.app.get(url(controller='forks', action='forks',
                                     repo_name=repo_name))
         response.mustcontain('There are no forks yet')
-class TestGIT(TestController, _BaseTestCase, _BaseFixture):
+class TestGIT(TestControllerPytest, _BaseTestCase, _BaseFixture):
     REPO = GIT_REPO
     NEW_REPO = NEW_GIT_REPO
     REPO_TYPE = 'git'
     REPO_FORK = GIT_FORK
-class TestHG(TestController, _BaseTestCase, _BaseFixture):
+class TestHG(TestControllerPytest, _BaseTestCase, _BaseFixture):
     REPO = HG_REPO
     NEW_REPO = NEW_HG_REPO
     REPO_TYPE = 'hg'
     REPO_FORK = HG_FORK

kallithea/tests/functional/test_pullrequests.py

➞

Show inline comments

 import re
 from kallithea.tests import *
 from kallithea.tests.fixture import Fixture
 from kallithea.model.meta import Session
 from kallithea.controllers.pullrequests import PullrequestsController
 fixture = Fixture()
 class TestPullrequestsController(TestControllerPytest):
     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:default:default',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'branch:default:default',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   '_authentication_token': self.authentication_token(),
+                                 }
+                                )
         self.assertEqual(response.status, '302 Found')
         response = response.follow()
         self.assertEqual(response.status, '200 OK')
         response.mustcontain('This pull request has already been merged to default.')
     def test_create_with_existing_reviewer(self):
         self.log_user()
         response = self.app.post(url(controller='pullrequests', action='create',
                                      repo_name=HG_REPO),
                                  {'org_repo': HG_REPO,
                                   'org_ref': 'branch:default:default',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'branch:default:default',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   '_authentication_token': self.authentication_token(),
                                   'review_members': TEST_USER_ADMIN_LOGIN,
+                                 }
+                                )
         self.assertEqual(response.status, '302 Found')
         response = response.follow()
         self.assertEqual(response.status, '200 OK')
         response.mustcontain('This pull request has already been merged to default.')
     def test_create_with_invalid_reviewer(self):
         invalid_user_name = 'invalid_user'
         self.log_user()
         response = self.app.post(url(controller='pullrequests', action='create',
                                      repo_name=HG_REPO),
+                                 {
                                   'org_repo': HG_REPO,
                                   'org_ref': 'branch:default:default',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'branch:default:default',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   '_authentication_token': self.authentication_token(),
                                   'review_members': invalid_user_name,
                                  },
                                  status=400)
         response.mustcontain('Invalid reviewer &#34;%s&#34; specified' % invalid_user_name)
     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': 'branch:default:default',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'branch:default:default',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   '_authentication_token': self.authentication_token(),
+                                 }
+                                )
         self.assertEqual(response.status, '302 Found')
         # location is of the form:
         # http://localhost/vcs_test_hg/pull-request/54/_/title
         m = re.search('/pull-request/(\d+)/', response.location)
         self.assertNotEqual(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': 'default',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   'owner': TEST_USER_ADMIN_LOGIN,
                                   '_authentication_token': self.authentication_token(),
                                   'review_members': 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:default:default',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'branch:default:default',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   '_authentication_token': self.authentication_token(),
+                                 }
+                                )
         self.assertEqual(response.status, '302 Found')
         # location is of the form:
         # http://localhost/vcs_test_hg/pull-request/54/_/title
         m = re.search('/pull-request/(\d+)/', response.location)
         self.assertNotEqual(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': invalid_user_id,
                                  },
                                  status=400)
         response.mustcontain('Invalid reviewer &#34;%s&#34; specified' % invalid_user_id)
-class TestPullrequestsGetRepoRefs(TestController):
+class TestPullrequestsGetRepoRefs(TestControllerPytest):
-    def setUp(self):
+    def setup_method(self, method):
         self.main = fixture.create_repo(u'main', repo_type='hg')
         Session.add(self.main)
         Session.commit()
         self.c = PullrequestsController()
-    def tearDown(self):
+    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.main.scm_instance)
         self.assertEqual(default, 'tag:null:0000000000000000000000000000000000000000')
     def test_repo_refs_one_commit_no_hints(self):
         cs0 = fixture.commit_change(self.main.repo_name, filename='file1',
                 content='line1\n', message='commit1', vcs_type='hg',
                 parent=None, newfile=True)
         refs, default = self.c._get_repo_refs(self.main.scm_instance)
         self.assertEqual(default, 'branch:default:%s' % cs0.raw_id)
         self.assertIn(([('branch:default:%s' % cs0.raw_id, 'default (current tip)')],
                 'Branches'), refs)
     def test_repo_refs_one_commit_rev_hint(self):
         cs0 = fixture.commit_change(self.main.repo_name, filename='file1',
                 content='line1\n', message='commit1', vcs_type='hg',
                 parent=None, newfile=True)
         refs, default = self.c._get_repo_refs(self.main.scm_instance, rev=cs0.raw_id)
         expected = 'branch:default:%s' % cs0.raw_id
         self.assertEqual(default, expected)
         self.assertIn(([(expected, 'default (current tip)')], 'Branches'), refs)
     def test_repo_refs_two_commits_no_hints(self):
         cs0 = fixture.commit_change(self.main.repo_name, filename='file1',
                 content='line1\n', message='commit1', vcs_type='hg',
                 parent=None, newfile=True)
         cs1 = fixture.commit_change(self.main.repo_name, filename='file2',
                 content='line2\n', message='commit2', vcs_type='hg',
                 parent=None, newfile=True)
         refs, default = self.c._get_repo_refs(self.main.scm_instance)
         expected = 'branch:default:%s' % cs1.raw_id
         self.assertEqual(default, expected)
         self.assertIn(([(expected, 'default (current tip)')], 'Branches'), refs)
     def test_repo_refs_two_commits_rev_hints(self):
         cs0 = fixture.commit_change(self.main.repo_name, filename='file1',
                 content='line1\n', message='commit1', vcs_type='hg',
                 parent=None, newfile=True)
         cs1 = fixture.commit_change(self.main.repo_name, filename='file2',
                 content='line2\n', message='commit2', vcs_type='hg',
                 parent=None, newfile=True)
         refs, default = self.c._get_repo_refs(self.main.scm_instance, rev=cs0.raw_id)
         expected = 'rev:%s:%s' % (cs0.raw_id, cs0.raw_id)
         self.assertEqual(default, expected)
         self.assertIn(([(expected, 'Changeset: %s' % cs0.raw_id[0:12])], 'Special'), refs)
         self.assertIn(([('branch:default:%s' % cs1.raw_id, 'default (current tip)')], 'Branches'), refs)
         refs, default = self.c._get_repo_refs(self.main.scm_instance, rev=cs1.raw_id)
         expected = 'branch:default:%s' % cs1.raw_id
         self.assertEqual(default, expected)
         self.assertIn(([(expected, 'default (current tip)')], 'Branches'), refs)
     def test_repo_refs_two_commits_branch_hint(self):
         cs0 = fixture.commit_change(self.main.repo_name, filename='file1',
                 content='line1\n', message='commit1', vcs_type='hg',
                 parent=None, newfile=True)
         cs1 = fixture.commit_change(self.main.repo_name, filename='file2',
                 content='line2\n', message='commit2', vcs_type='hg',
                 parent=None, newfile=True)
         refs, default = self.c._get_repo_refs(self.main.scm_instance, branch='default')
         expected = 'branch:default:%s' % cs1.raw_id
         self.assertEqual(default, expected)
         self.assertIn(([(expected, 'default (current tip)')], 'Branches'), refs)
     def test_repo_refs_one_branch_no_hints(self):
         cs0 = fixture.commit_change(self.main.repo_name, filename='file1',
                 content='line1\n', message='commit1', vcs_type='hg',
                 parent=None, newfile=True)
         # TODO

0 comments (0 inline, 0 general)