kallithea Changeset - be1d366f461c

Changeset - be1d366f461c

Parent rev.

Child rev.

[Not reviewed]

default

0 16 0

Thomas De Schampheleire - 10 years ago 2016-05-14 21:04:26
thomas.de.schampheleire@gmail.com

pytest migration: functional: switch to standard assert statements

Use unittest2pytest to replace unittest-style assert statements (e.g.
assertEqual) with standard Python assert statements to benefit from pytest's
improved reporting on assert failures.

The conversion by unittest2pytest was correct, except for line wrapping
problems.

15 files changed:

kallithea/tests/functional/test_admin_auth_settings.py

kallithea/tests/functional/test_admin_defaults.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_admin_settings.py

kallithea/tests/functional/test_admin_user_groups.py

kallithea/tests/functional/test_admin_users.py

kallithea/tests/functional/test_changelog.py

kallithea/tests/functional/test_changeset_comments.py

kallithea/tests/functional/test_compare.py

kallithea/tests/functional/test_files.py

kallithea/tests/functional/test_forks.py

kallithea/tests/functional/test_login.py

kallithea/tests/functional/test_my_account.py

Changeset was too big and was cut off... Show full diff anyway

0 comments (0 inline, 0 general)

kallithea/tests/functional/test_admin_auth_settings.py

➞

Show inline comments

 from kallithea.tests import *
 from kallithea.model.db import Setting
 class TestAuthSettingsController(TestControllerPytest):
     def _enable_plugins(self, plugins_list):
         test_url = url(controller='admin/auth_settings',
                        action='auth_settings')
         params={'auth_plugins': plugins_list, '_authentication_token': self.authentication_token()}
         for plugin in plugins_list.split(','):
             enable = plugin.partition('kallithea.lib.auth_modules.')[-1]
             params.update({'%s_enabled' % enable: True})
         response = self.app.post(url=test_url, params=params)
         return params
         #self.checkSessionFlash(response, 'Auth settings updated successfully')
     def test_index(self):
         self.log_user()
         response = self.app.get(url(controller='admin/auth_settings',
                                     action='index'))
         response.mustcontain('Authentication Plugins')
     @skipif(not ldap_lib_installed, reason='skipping due to missing ldap lib')
     def test_ldap_save_settings(self):
         self.log_user()
         params = self._enable_plugins('kallithea.lib.auth_modules.auth_internal,kallithea.lib.auth_modules.auth_ldap')
         params.update({'auth_ldap_host': u'dc.example.com',
                        'auth_ldap_port': '999',
                        'auth_ldap_tls_kind': 'PLAIN',
                        'auth_ldap_tls_reqcert': 'NEVER',
                        'auth_ldap_dn_user': 'test_user',
                        'auth_ldap_dn_pass': 'test_pass',
                        'auth_ldap_base_dn': 'test_base_dn',
                        'auth_ldap_filter': 'test_filter',
                        'auth_ldap_search_scope': 'BASE',
                        'auth_ldap_attr_login': 'test_attr_login',
                        'auth_ldap_attr_firstname': 'ima',
                        'auth_ldap_attr_lastname': 'tester',
                        'auth_ldap_attr_email': 'test@example.com'})
         test_url = url(controller='admin/auth_settings',
                        action='auth_settings')
         response = self.app.post(url=test_url, params=params)
         self.checkSessionFlash(response, 'Auth settings updated successfully')
         new_settings = Setting.get_auth_settings()
         self.assertEqual(new_settings['auth_ldap_host'], u'dc.example.com',
                          'fail db write compare')
         assert new_settings['auth_ldap_host'] == u'dc.example.com', 'fail db write compare'
     @skipif(not ldap_lib_installed, reason='skipping due to missing ldap lib')
     def test_ldap_error_form_wrong_port_number(self):
         self.log_user()
         params = self._enable_plugins('kallithea.lib.auth_modules.auth_internal,kallithea.lib.auth_modules.auth_ldap')
         params.update({'auth_ldap_host': '',
                        'auth_ldap_port': 'i-should-be-number',  # bad port num
                        'auth_ldap_tls_kind': 'PLAIN',
                        'auth_ldap_tls_reqcert': 'NEVER',
                        'auth_ldap_dn_user': '',
                        'auth_ldap_dn_pass': '',
                        'auth_ldap_base_dn': '',
                        'auth_ldap_filter': '',
                        'auth_ldap_search_scope': 'BASE',
                        'auth_ldap_attr_login': '',
                        'auth_ldap_attr_firstname': '',
                        'auth_ldap_attr_lastname': '',
                        'auth_ldap_attr_email': ''})
         test_url = url(controller='admin/auth_settings',
                        action='auth_settings')
         response = self.app.post(url=test_url, params=params)
         response.mustcontain("""<span class="error-message">"""
                              """Please enter a number</span>""")
     @skipif(not ldap_lib_installed, reason='skipping due to missing ldap lib')
     def test_ldap_error_form(self):
         self.log_user()
         params = self._enable_plugins('kallithea.lib.auth_modules.auth_internal,kallithea.lib.auth_modules.auth_ldap')
         params.update({'auth_ldap_host': 'Host',
                        'auth_ldap_port': '123',
                        'auth_ldap_tls_kind': 'PLAIN',
                        'auth_ldap_tls_reqcert': 'NEVER',
                        'auth_ldap_dn_user': '',
                        'auth_ldap_dn_pass': '',
                        'auth_ldap_base_dn': '',
                        'auth_ldap_filter': '',
                        'auth_ldap_search_scope': 'BASE',
                        'auth_ldap_attr_login': '',  # <----- missing required input
                        'auth_ldap_attr_firstname': '',
                        'auth_ldap_attr_lastname': '',
                        'auth_ldap_attr_email': ''})
         test_url = url(controller='admin/auth_settings',
                        action='auth_settings')
         response = self.app.post(url=test_url, params=params)
         response.mustcontain("""<span class="error-message">The LDAP Login"""
                              """ attribute of the CN must be specified""")
     def test_ldap_login(self):
         pass
     def test_ldap_login_incorrect(self):
         pass
     def _container_auth_setup(self, **settings):
         self.log_user()
         params = self._enable_plugins('kallithea.lib.auth_modules.auth_internal,kallithea.lib.auth_modules.auth_container')
         params.update(settings)
         test_url = url(controller='admin/auth_settings',
                        action='auth_settings')
         response = self.app.post(url=test_url, params=params)
         response = response.follow()
         response.click('Log Out') # end admin login session
     def _container_auth_verify_login(self, resulting_username, **get_kwargs):
         response = self.app.get(
             url=url(controller='admin/my_account', action='my_account'),
             **get_kwargs
+        )
         response.mustcontain('My Account %s' % resulting_username)
     def test_container_auth_login_header(self):
         self._container_auth_setup(
             auth_container_header='THE_USER_NAME',
             auth_container_email_header='',
             auth_container_firstname_header='',
             auth_container_lastname_header='',
             auth_container_fallback_header='',
             auth_container_clean_username='False',
+        )
         self._container_auth_verify_login(
             extra_environ={'THE_USER_NAME': 'john@example.org'},
             resulting_username='john@example.org',
+        )
     def test_container_auth_login_header_attr(self):
         self._container_auth_setup(
             auth_container_header='THE_USER_NAME',
             auth_container_email_header='THE_USER_EMAIL',
             auth_container_firstname_header='THE_USER_FIRSTNAME',
             auth_container_lastname_header='THE_USER_LASTNAME',
             auth_container_fallback_header='',
             auth_container_clean_username='False',
+        )
         response = self.app.get(
             url=url(controller='admin/my_account', action='my_account'),
             extra_environ={'THE_USER_NAME': 'johnd',
                            'THE_USER_EMAIL': 'john@example.org',
                            'THE_USER_FIRSTNAME': 'John',
                            'THE_USER_LASTNAME': 'Doe',
+                           }
+        )
         self.assertEqual(response.form['email'].value, 'john@example.org')
         self.assertEqual(response.form['firstname'].value, 'John')
         self.assertEqual(response.form['lastname'].value, 'Doe')
         assert response.form['email'].value == 'john@example.org'
         assert response.form['firstname'].value == 'John'
         assert response.form['lastname'].value == 'Doe'
     def test_container_auth_login_fallback_header(self):
         self._container_auth_setup(
             auth_container_header='THE_USER_NAME',
             auth_container_email_header='',
             auth_container_firstname_header='',
             auth_container_lastname_header='',
             auth_container_fallback_header='HTTP_X_YZZY',
             auth_container_clean_username='False',
+        )
         self._container_auth_verify_login(
             headers={'X-Yzzy': r'foo\bar'},
             resulting_username=r'foo\bar',
+        )
     def test_container_auth_clean_username_at(self):
         self._container_auth_setup(
             auth_container_header='REMOTE_USER',
             auth_container_email_header='',
             auth_container_firstname_header='',
             auth_container_lastname_header='',
             auth_container_fallback_header='',
             auth_container_clean_username='True',
+        )
         self._container_auth_verify_login(
             extra_environ={'REMOTE_USER': 'john@example.org'},
             resulting_username='john',
+        )
     def test_container_auth_clean_username_backslash(self):
         self._container_auth_setup(
             auth_container_header='REMOTE_USER',
             auth_container_email_header='',
             auth_container_firstname_header='',
             auth_container_lastname_header='',
             auth_container_fallback_header='',
             auth_container_clean_username='True',
+        )
         self._container_auth_verify_login(
             extra_environ={'REMOTE_USER': r'example\jane'},
             resulting_username=r'jane',
+        )
     def test_container_auth_no_logout(self):
         self._container_auth_setup(
             auth_container_header='REMOTE_USER',
             auth_container_email_header='',
             auth_container_firstname_header='',
             auth_container_lastname_header='',
             auth_container_fallback_header='',
             auth_container_clean_username='True',
+        )
         response = self.app.get(
             url=url(controller='admin/my_account', action='my_account'),
             extra_environ={'REMOTE_USER': 'john'},
+        )
-        self.assertNotIn('Log Out', response.normal_body)
+        assert 'Log Out' not in response.normal_body
     def test_crowd_save_settings(self):
         self.log_user()
         params = self._enable_plugins('kallithea.lib.auth_modules.auth_internal,kallithea.lib.auth_modules.auth_crowd')
         params.update({'auth_crowd_host': ' hostname ',
                        'auth_crowd_app_password': 'secret',
                        'auth_crowd_admin_groups': 'mygroup',
                        'auth_crowd_port': '123',
                        'auth_crowd_app_name': 'xyzzy'})
         test_url = url(controller='admin/auth_settings',
                        action='auth_settings')
         response = self.app.post(url=test_url, params=params)
         self.checkSessionFlash(response, 'Auth settings updated successfully')
         new_settings = Setting.get_auth_settings()
         self.assertEqual(new_settings['auth_crowd_host'], u'hostname',
                          'fail db write compare')
         assert new_settings['auth_crowd_host'] == u'hostname', 'fail db write compare'
     @skipif(not pam_lib_installed, reason='skipping due to missing pam lib')
     def test_pam_save_settings(self):
         self.log_user()
         params = self._enable_plugins('kallithea.lib.auth_modules.auth_internal,kallithea.lib.auth_modules.auth_pam')
         params.update({'auth_pam_service': 'kallithea',
                        'auth_pam_gecos': '^foo-.*'})
         test_url = url(controller='admin/auth_settings',
                        action='auth_settings')
         response = self.app.post(url=test_url, params=params)
         self.checkSessionFlash(response, 'Auth settings updated successfully')
         new_settings = Setting.get_auth_settings()
         self.assertEqual(new_settings['auth_pam_service'], u'kallithea',
                          'fail db write compare')
         assert new_settings['auth_pam_service'] == u'kallithea', 'fail db write compare'

kallithea/tests/functional/test_admin_defaults.py

➞

Show inline comments

 from kallithea.tests import *
 from kallithea.model.db import Setting
 class TestDefaultsController(TestControllerPytest):
     def test_index(self):
         self.log_user()
         response = self.app.get(url('defaults'))
         response.mustcontain('default_repo_private')
         response.mustcontain('default_repo_enable_statistics')
         response.mustcontain('default_repo_enable_downloads')
         response.mustcontain('default_repo_enable_locking')
     def test_index_as_xml(self):
         response = self.app.get(url('formatted_defaults', format='xml'))
     def test_create(self):
         response = self.app.post(url('defaults'),
             {'_authentication_token': self.authentication_token()})
     def test_new(self):
         response = self.app.get(url('new_default'))
     def test_new_as_xml(self):
         response = self.app.get(url('formatted_new_default', format='xml'))
     def test_update_params_true_hg(self):
         self.log_user()
         params = {
             'default_repo_enable_locking': True,
             'default_repo_enable_downloads': True,
             'default_repo_enable_statistics': True,
             'default_repo_private': True,
             'default_repo_type': 'hg',
             '_authentication_token': self.authentication_token(),
+        }
         response = self.app.put(url('default', id='default'), params=params)
         self.checkSessionFlash(response, 'Default settings updated successfully')
         params.pop('_authentication_token')
         defs = Setting.get_default_repo_settings()
-        self.assertEqual(params, defs)
+        assert params == defs
     def test_update_params_false_git(self):
         self.log_user()
         params = {
             'default_repo_enable_locking': False,
             'default_repo_enable_downloads': False,
             'default_repo_enable_statistics': False,
             'default_repo_private': False,
             'default_repo_type': 'git',
             '_authentication_token': self.authentication_token(),
+        }
         response = self.app.put(url('default', id='default'), params=params)
         self.checkSessionFlash(response, 'Default settings updated successfully')
         params.pop('_authentication_token')
         defs = Setting.get_default_repo_settings()
-        self.assertEqual(params, defs)
+        assert params == defs
     def test_update_browser_fakeout(self):
         response = self.app.post(url('default', id=1), params=dict(_method='put', _authentication_token=self.authentication_token()))
     def test_delete(self):
         # Not possible due to CSRF protection.
         response = self.app.delete(url('default', id=1), status=405)
     def test_delete_browser_fakeout(self):
         response = self.app.post(url('default', id=1), params=dict(_method='delete', _authentication_token=self.authentication_token()))
     def test_show(self):
         response = self.app.get(url('default', id=1))
     def test_show_as_xml(self):
         response = self.app.get(url('formatted_default', id=1, format='xml'))
     def test_edit(self):
         response = self.app.get(url('edit_default', id=1))
     def test_edit_as_xml(self):
         response = self.app.get(url('formatted_edit_default', id=1, format='xml'))

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(TestControllerPytest):
     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')
+        assert 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')
+        assert 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(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])
         assert get_notif(cur_user.notifications) == [notification]
         assert get_notif(u1.notifications) == [notification]
         assert 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')
+        assert response.body == 'ok'
         cur_user = User.get(cur_usr_id)
-        self.assertEqual(cur_user.notifications, [])
+        assert 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(
         assert 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(
         assert 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

 # -*- coding: utf-8 -*-
 import os
 import mock
 import urllib
 from kallithea.lib import vcs
 from kallithea.lib.utils2 import safe_str, safe_unicode
 from kallithea.model.db import Repository, RepoGroup, UserRepoToPerm, User, \
     Permission, Ui
 from kallithea.model.user import UserModel
 from kallithea.tests import *
 from kallithea.model.repo_group import RepoGroupModel
 from kallithea.model.repo import RepoModel
 from kallithea.model.meta import Session
 from kallithea.tests.fixture import Fixture, error_function
 fixture = Fixture()
 def _get_permission_for_user(user, repo):
     perm = UserRepoToPerm.query() \
                 .filter(UserRepoToPerm.repository ==
                         Repository.get_by_repo_name(repo)) \
                 .filter(UserRepoToPerm.user == User.get_by_username(user)) \
                 .all()
     return perm
 class _BaseTestCase(TestControllerPytest):
     """
     Write all tests here
     """
     REPO = None
     REPO_TYPE = None
     NEW_REPO = None
     OTHER_TYPE_REPO = None
     OTHER_TYPE = None
     def test_index(self):
         self.log_user()
         response = self.app.get(url('repos'))
     def test_create(self):
         self.log_user()
         repo_name = self.NEW_REPO
         description = u'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()))
         ## run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=repo_name))
-        self.assertEqual(response.json, {u'result': True})
+        assert response.json == {u'result': True}
         self.checkSessionFlash(response,
                                'Created repository <a href="/%s">%s</a>'
                                % (repo_name, repo_name))
         # test if the repo was created in the database
         new_repo = Session().query(Repository) \
             .filter(Repository.repo_name == repo_name).one()
         self.assertEqual(new_repo.repo_name, repo_name)
         self.assertEqual(new_repo.description, description)
         assert new_repo.repo_name == repo_name
         assert new_repo.description == description
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=repo_name))
         response.mustcontain(repo_name)
         response.mustcontain(self.REPO_TYPE)
         # test if the repository was created on filesystem
         try:
             vcs.get_repo(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name)))
         except vcs.exceptions.VCSError:
             pytest.fail('no repo %s in filesystem' % repo_name)
         RepoModel().delete(repo_name)
         Session().commit()
     def test_create_in_group(self):
         self.log_user()
         ## create GROUP
         group_name = u'sometest_%s' % self.REPO_TYPE
         gr = RepoGroupModel().create(group_name=group_name,
                                      group_description=u'test',
                                      owner=TEST_USER_ADMIN_LOGIN)
         Session().commit()
         repo_name = u'ingroup'
         repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])
         description = u'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,
                                                 repo_group=gr.group_id,
                                                 _authentication_token=self.authentication_token()))
         ## run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=repo_name_full))
-        self.assertEqual(response.json, {u'result': True})
+        assert response.json == {u'result': True}
         self.checkSessionFlash(response,
                                'Created repository <a href="/%s">%s</a>'
                                % (repo_name_full, repo_name_full))
         # test if the repo was created in the database
         new_repo = Session().query(Repository) \
             .filter(Repository.repo_name == repo_name_full).one()
         new_repo_id = new_repo.repo_id
         self.assertEqual(new_repo.repo_name, repo_name_full)
         self.assertEqual(new_repo.description, description)
         assert new_repo.repo_name == repo_name_full
         assert new_repo.description == description
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=repo_name_full))
         response.mustcontain(repo_name_full)
         response.mustcontain(self.REPO_TYPE)
         inherited_perms = UserRepoToPerm.query() \
             .filter(UserRepoToPerm.repository_id == new_repo_id).all()
-        self.assertEqual(len(inherited_perms), 1)
+        assert len(inherited_perms) == 1
         # test if the repository was created on filesystem
         try:
             vcs.get_repo(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name_full)))
         except vcs.exceptions.VCSError:
             RepoGroupModel().delete(group_name)
             Session().commit()
             pytest.fail('no repo %s in filesystem' % repo_name)
         RepoModel().delete(repo_name_full)
         RepoGroupModel().delete(group_name)
         Session().commit()
     def test_create_in_group_without_needed_permissions(self):
         usr = self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
         # avoid spurious RepoGroup DetachedInstanceError ...
         authentication_token = self.authentication_token()
         # 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()
         ## create GROUP
         group_name = u'reg_sometest_%s' % self.REPO_TYPE
         gr = RepoGroupModel().create(group_name=group_name,
                                      group_description=u'test',
                                      owner=TEST_USER_ADMIN_LOGIN)
         Session().commit()
         group_name_allowed = u'reg_sometest_allowed_%s' % self.REPO_TYPE
         gr_allowed = RepoGroupModel().create(group_name=group_name_allowed,
                                      group_description=u'test',
                                      owner=TEST_USER_REGULAR_LOGIN)
         Session().commit()
         repo_name = u'ingroup'
         repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])
         description = u'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,
                                                 repo_group=gr.group_id,
                                                 _authentication_token=authentication_token))
         response.mustcontain('Invalid value')
         # user is allowed to create in this group
         repo_name = u'ingroup'
         repo_name_full = RepoGroup.url_sep().join([group_name_allowed, repo_name])
         description = u'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,
                                                 repo_group=gr_allowed.group_id,
                                                 _authentication_token=authentication_token))
         ## run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=repo_name_full))
-        self.assertEqual(response.json, {u'result': True})
+        assert response.json == {u'result': True}
         self.checkSessionFlash(response,
                                'Created repository <a href="/%s">%s</a>'
                                % (repo_name_full, repo_name_full))
         # test if the repo was created in the database
         new_repo = Session().query(Repository) \
             .filter(Repository.repo_name == repo_name_full).one()
         new_repo_id = new_repo.repo_id
         self.assertEqual(new_repo.repo_name, repo_name_full)
         self.assertEqual(new_repo.description, description)
         assert new_repo.repo_name == repo_name_full
         assert new_repo.description == description
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=repo_name_full))
         response.mustcontain(repo_name_full)
         response.mustcontain(self.REPO_TYPE)
         inherited_perms = UserRepoToPerm.query() \
             .filter(UserRepoToPerm.repository_id == new_repo_id).all()
-        self.assertEqual(len(inherited_perms), 1)
+        assert len(inherited_perms) == 1
         # test if the repository was created on filesystem
         try:
             vcs.get_repo(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name_full)))
         except vcs.exceptions.VCSError:
             RepoGroupModel().delete(group_name)
             Session().commit()
             pytest.fail('no repo %s in filesystem' % repo_name)
         RepoModel().delete(repo_name_full)
         RepoGroupModel().delete(group_name)
         RepoGroupModel().delete(group_name_allowed)
         Session().commit()
     def test_create_in_group_inherit_permissions(self):
         self.log_user()
         ## create GROUP
         group_name = u'sometest_%s' % self.REPO_TYPE
         gr = RepoGroupModel().create(group_name=group_name,
                                      group_description=u'test',
                                      owner=TEST_USER_ADMIN_LOGIN)
         perm = Permission.get_by_key('repository.write')
         RepoGroupModel().grant_user_permission(gr, TEST_USER_REGULAR_LOGIN, perm)
         ## add repo permissions
         Session().commit()
         repo_name = u'ingroup_inherited_%s' % self.REPO_TYPE
         repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])
         description = u'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,
                                                 repo_group=gr.group_id,
                                                 repo_copy_permissions=True,
                                                 _authentication_token=self.authentication_token()))
         ## run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=repo_name_full))
         self.checkSessionFlash(response,
                                'Created repository <a href="/%s">%s</a>'
                                % (repo_name_full, repo_name_full))
         # test if the repo was created in the database
         new_repo = Session().query(Repository) \
             .filter(Repository.repo_name == repo_name_full).one()
         new_repo_id = new_repo.repo_id
         self.assertEqual(new_repo.repo_name, repo_name_full)
         self.assertEqual(new_repo.description, description)
         assert new_repo.repo_name == repo_name_full
         assert new_repo.description == description
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=repo_name_full))
         response.mustcontain(repo_name_full)
         response.mustcontain(self.REPO_TYPE)
         # test if the repository was created on filesystem
         try:
             vcs.get_repo(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name_full)))
         except vcs.exceptions.VCSError:
             RepoGroupModel().delete(group_name)
             Session().commit()
             pytest.fail('no repo %s in filesystem' % repo_name)
         #check if inherited permissiona are applied
         inherited_perms = UserRepoToPerm.query() \
             .filter(UserRepoToPerm.repository_id == new_repo_id).all()
-        self.assertEqual(len(inherited_perms), 2)
+        assert len(inherited_perms) == 2
         self.assertTrue(TEST_USER_REGULAR_LOGIN in [x.user.username
                                                     for x in inherited_perms])
         self.assertTrue('repository.write' in [x.permission.permission_name
                                                for x in inherited_perms])
         assert TEST_USER_REGULAR_LOGIN in [x.user.username
                                                     for x in inherited_perms]
         assert 'repository.write' in [x.permission.permission_name
                                                for x in inherited_perms]
         RepoModel().delete(repo_name_full)
         RepoGroupModel().delete(group_name)
         Session().commit()
     def test_create_remote_repo_wrong_clone_uri(self):
         self.log_user()
         repo_name = self.NEW_REPO
         description = u'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,
                                                 clone_uri='http://127.0.0.1/repo',
                                                 _authentication_token=self.authentication_token()))
         response.mustcontain('Invalid repository URL')
     def test_create_remote_repo_wrong_clone_uri_hg_svn(self):
         self.log_user()
         repo_name = self.NEW_REPO
         description = u'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,
                                                 clone_uri='svn+http://127.0.0.1/repo',
                                                 _authentication_token=self.authentication_token()))
         response.mustcontain('Invalid repository URL')
     def test_delete(self):
         self.log_user()
         repo_name = u'vcs_test_new_to_delete_%s' % self.REPO_TYPE
         description = u'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_name=repo_name,
                                                 repo_description=description,
                                                 _authentication_token=self.authentication_token()))
         ## run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=repo_name))
         self.checkSessionFlash(response,
                                'Created repository <a href="/%s">%s</a>'
                                % (repo_name, repo_name))
         # test if the repo was created in the database
         new_repo = Session().query(Repository) \
             .filter(Repository.repo_name == repo_name).one()
         self.assertEqual(new_repo.repo_name, repo_name)
         self.assertEqual(new_repo.description, description)
         assert new_repo.repo_name == repo_name
         assert new_repo.description == description
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=repo_name))
         response.mustcontain(repo_name)
         response.mustcontain(self.REPO_TYPE)
         # test if the repository was created on filesystem
         try:
             vcs.get_repo(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name)))
         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))
         response.follow()
         #check if repo was deleted from db
         deleted_repo = Session().query(Repository) \
             .filter(Repository.repo_name == repo_name).scalar()
-        self.assertEqual(deleted_repo, None)
+        assert deleted_repo == None
         self.assertEqual(os.path.isdir(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name)),
                                   False)
         assert os.path.isdir(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name)) == False
     def test_delete_non_ascii(self):
         self.log_user()
         non_ascii = "ąęł"
         repo_name = "%s%s" % (safe_str(self.NEW_REPO), non_ascii)
         repo_name_unicode = safe_unicode(repo_name)
         description = 'description for newly created repo' + non_ascii
         description_unicode = safe_unicode(description)
         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()))
         ## run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=repo_name))
-        self.assertEqual(response.json, {u'result': True})
+        assert response.json == {u'result': True}
         self.checkSessionFlash(response,
                                u'Created repository <a href="/%s">%s</a>'
                                % (urllib.quote(repo_name), repo_name_unicode))
         # test if the repo was created in the database
         new_repo = Session().query(Repository) \
             .filter(Repository.repo_name == repo_name_unicode).one()
         self.assertEqual(new_repo.repo_name, repo_name_unicode)
         self.assertEqual(new_repo.description, description_unicode)
         assert new_repo.repo_name == repo_name_unicode
         assert new_repo.description == description_unicode
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=repo_name))
         response.mustcontain(repo_name)
         response.mustcontain(self.REPO_TYPE)
         # test if the repository was created on filesystem
         try:
             vcs.get_repo(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name_unicode)))
         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)
+        assert deleted_repo == None
         self.assertEqual(os.path.isdir(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name_unicode)),
                                   False)
         assert 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)
         assert len(perm), 1
         assert perm[0].permission.permission_name == 'repository.read'
         assert 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)
+        assert 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')
         assert len(perm), 1
         assert 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)
+        assert 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')
         assert len(perm), 1
         assert 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)
+        assert 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)))
+        assert not os.path.isdir(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name))
 class TestAdminReposControllerGIT(_BaseTestCase):
     REPO = GIT_REPO
     REPO_TYPE = 'git'
     NEW_REPO = NEW_GIT_REPO
     OTHER_TYPE_REPO = HG_REPO
     OTHER_TYPE = 'hg'
 class TestAdminReposControllerHG(_BaseTestCase):
     REPO = HG_REPO
     REPO_TYPE = 'hg'
     NEW_REPO = NEW_HG_REPO
     OTHER_TYPE_REPO = GIT_REPO
     OTHER_TYPE = 'git'

kallithea/tests/functional/test_admin_settings.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 import tempfile
 from kallithea.model.db import Setting, Ui
 from kallithea.tests import *
 from kallithea.tests.fixture import Fixture
 fixture = Fixture()
 class TestAdminSettingsController(TestControllerPytest):
     def test_index_main(self):
         self.log_user()
         response = self.app.get(url('admin_settings'))
     def test_index_mapping(self):
         self.log_user()
         response = self.app.get(url('admin_settings_mapping'))
     def test_index_global(self):
         self.log_user()
         response = self.app.get(url('admin_settings_global'))
     def test_index_visual(self):
         self.log_user()
         response = self.app.get(url('admin_settings_visual'))
     def test_index_email(self):
         self.log_user()
         response = self.app.get(url('admin_settings_email'))
     def test_index_hooks(self):
         self.log_user()
         response = self.app.get(url('admin_settings_hooks'))
     def test_create_custom_hook(self):
         self.log_user()
         response = self.app.post(url('admin_settings_hooks'),
                                 params=dict(new_hook_ui_key='test_hooks_1',
                                             new_hook_ui_value='cd %s' % tempfile.gettempdir(),
                                             _authentication_token=self.authentication_token()))
         response = response.follow()
         response.mustcontain('test_hooks_1')
         response.mustcontain('cd %s' % tempfile.gettempdir())
     def test_create_custom_hook_delete(self):
         self.log_user()
         response = self.app.post(url('admin_settings_hooks'),
                                 params=dict(new_hook_ui_key='test_hooks_2',
                                             new_hook_ui_value='cd %s2' % tempfile.gettempdir(),
                                             _authentication_token=self.authentication_token()))
         response = response.follow()
         response.mustcontain('test_hooks_2')
         response.mustcontain('cd %s2' % tempfile.gettempdir())
         hook_id = Ui.get_by_key('hooks', 'test_hooks_2').ui_id
         ## delete
         self.app.post(url('admin_settings_hooks'),
                         params=dict(hook_id=hook_id, _authentication_token=self.authentication_token()))
         response = self.app.get(url('admin_settings_hooks'))
         response.mustcontain(no=['test_hooks_2'])
         response.mustcontain(no=['cd %s2' % tempfile.gettempdir()])
     def test_index_search(self):
         self.log_user()
         response = self.app.get(url('admin_settings_search'))
     def test_index_system(self):
         self.log_user()
         response = self.app.get(url('admin_settings_system'))
     def test_ga_code_active(self):
         self.log_user()
         old_title = 'Kallithea'
         old_realm = 'Kallithea authentication'
         new_ga_code = 'ga-test-123456789'
         response = self.app.post(url('admin_settings_global'),
                         params=dict(title=old_title,
                                  realm=old_realm,
                                  ga_code=new_ga_code,
                                  captcha_private_key='',
                                  captcha_public_key='',
                                  _authentication_token=self.authentication_token(),
                                  ))
         self.checkSessionFlash(response, 'Updated application settings')
         self.assertEqual(Setting
                          .get_app_settings()['ga_code'], new_ga_code)
         assert Setting.get_app_settings()['ga_code'] == new_ga_code
         response = response.follow()
         response.mustcontain("""_gaq.push(['_setAccount', '%s']);""" % new_ga_code)
     def test_ga_code_inactive(self):
         self.log_user()
         old_title = 'Kallithea'
         old_realm = 'Kallithea authentication'
         new_ga_code = ''
         response = self.app.post(url('admin_settings_global'),
                         params=dict(title=old_title,
                                  realm=old_realm,
                                  ga_code=new_ga_code,
                                  captcha_private_key='',
                                  captcha_public_key='',
                                  _authentication_token=self.authentication_token(),
                                  ))
         self.checkSessionFlash(response, 'Updated application settings')
         self.assertEqual(Setting
                         .get_app_settings()['ga_code'], new_ga_code)
         assert Setting.get_app_settings()['ga_code'] == new_ga_code
         response = response.follow()
         response.mustcontain(no=["_gaq.push(['_setAccount', '%s']);" % new_ga_code])
     def test_captcha_activate(self):
         self.log_user()
         old_title = 'Kallithea'
         old_realm = 'Kallithea authentication'
         new_ga_code = ''
         response = self.app.post(url('admin_settings_global'),
                         params=dict(title=old_title,
                                  realm=old_realm,
                                  ga_code=new_ga_code,
                                  captcha_private_key='1234567890',
                                  captcha_public_key='1234567890',
                                  _authentication_token=self.authentication_token(),
                                  ))
         self.checkSessionFlash(response, 'Updated application settings')
         self.assertEqual(Setting
                         .get_app_settings()['captcha_private_key'], '1234567890')
         assert Setting.get_app_settings()['captcha_private_key'] == '1234567890'
         response = self.app.get(url('register'))
         response.mustcontain('captcha')
     def test_captcha_deactivate(self):
         self.log_user()
         old_title = 'Kallithea'
         old_realm = 'Kallithea authentication'
         new_ga_code = ''
         response = self.app.post(url('admin_settings_global'),
                         params=dict(title=old_title,
                                  realm=old_realm,
                                  ga_code=new_ga_code,
                                  captcha_private_key='',
                                  captcha_public_key='1234567890',
                                  _authentication_token=self.authentication_token(),
                                  ))
         self.checkSessionFlash(response, 'Updated application settings')
         self.assertEqual(Setting
                         .get_app_settings()['captcha_private_key'], '')
         assert Setting.get_app_settings()['captcha_private_key'] == ''
         response = self.app.get(url('register'))
         response.mustcontain(no=['captcha'])
     def test_title_change(self):
         self.log_user()
         old_title = 'Kallithea'
         new_title = old_title + '_changed'
         old_realm = 'Kallithea authentication'
         for new_title in ['Changed', 'Żółwik', old_title]:
             response = self.app.post(url('admin_settings_global'),
                         params=dict(title=new_title,
                                  realm=old_realm,
                                  ga_code='',
                                  captcha_private_key='',
                                  captcha_public_key='',
                                  _authentication_token=self.authentication_token(),
                                 ))
             self.checkSessionFlash(response, 'Updated application settings')
             self.assertEqual(Setting
                              .get_app_settings()['title'],
                              new_title.decode('utf-8'))
             assert Setting.get_app_settings()['title'] == new_title.decode('utf-8')
             response = response.follow()
             response.mustcontain("""<div class="branding">%s</div>""" % new_title)

kallithea/tests/functional/test_admin_user_groups.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 from kallithea.tests import *
 from kallithea.model.db import UserGroup, UserGroupToPerm, Permission
 from kallithea.model.meta import Session
 TEST_USER_GROUP = u'admins_test'
 class TestAdminUsersGroupsController(TestControllerPytest):
     def test_index(self):
         self.log_user()
         response = self.app.get(url('users_groups'))
         # Test response...
     def test_create(self):
         self.log_user()
         users_group_name = TEST_USER_GROUP
         response = self.app.post(url('users_groups'),
                                  {'users_group_name': users_group_name,
                                   'user_group_description': u'DESC',
                                   'active': True,
                                   '_authentication_token': self.authentication_token()})
         response.follow()
         self.checkSessionFlash(response,
                                'Created user group <a href="/_admin/user_groups/')
         self.checkSessionFlash(response,
                                '/edit">%s</a>' % TEST_USER_GROUP)
     def test_new(self):
         response = self.app.get(url('new_users_group'))
     def test_update(self):
         response = self.app.put(url('users_group', id=1), status=403)
     def test_update_browser_fakeout(self):
         response = self.app.post(url('users_group', id=1),
                                  params=dict(_method='put', _authentication_token=self.authentication_token()))
     def test_delete(self):
         self.log_user()
         users_group_name = TEST_USER_GROUP + 'another'
         response = self.app.post(url('users_groups'),
                                  {'users_group_name':users_group_name,
                                   'user_group_description': u'DESC',
                                   'active': True,
                                   '_authentication_token': self.authentication_token()})
         response.follow()
         self.checkSessionFlash(response,
                                'Created user group ')
         gr = Session().query(UserGroup) \
             .filter(UserGroup.users_group_name == users_group_name).one()
         response = self.app.post(url('users_group', id=gr.users_group_id),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         gr = Session().query(UserGroup) \
             .filter(UserGroup.users_group_name == users_group_name).scalar()
-        self.assertEqual(gr, None)
+        assert gr == None
     def test_default_perms_enable_repository_read_on_group(self):
         self.log_user()
         users_group_name = TEST_USER_GROUP + 'another2'
         response = self.app.post(url('users_groups'),
                                  {'users_group_name': users_group_name,
                                   'user_group_description': u'DESC',
                                   'active': True,
                                   '_authentication_token': self.authentication_token()})
         response.follow()
         ug = UserGroup.get_by_group_name(users_group_name)
         self.checkSessionFlash(response,
                                'Created user group ')
         ## ENABLE REPO CREATE ON A GROUP
         response = self.app.put(url('edit_user_group_default_perms',
                                     id=ug.users_group_id),
                                  {'create_repo_perm': True,
                                   '_authentication_token': self.authentication_token()})
         response.follow()
         ug = UserGroup.get_by_group_name(users_group_name)
         p = Permission.get_by_key('hg.create.repository')
         p2 = Permission.get_by_key('hg.usergroup.create.false')
         p3 = Permission.get_by_key('hg.fork.none')
         # check if user has this perms, they should be here since
         # defaults are on
         perms = UserGroupToPerm.query() \
             .filter(UserGroupToPerm.users_group == ug).all()
         self.assertEqual(
             sorted([[x.users_group_id, x.permission_id, ] for x in perms]),
             sorted([[ug.users_group_id, p.permission_id],
         assert sorted([[x.users_group_id, x.permission_id, ] for x in perms]) == sorted([[ug.users_group_id, p.permission_id],
                     [ug.users_group_id, p2.permission_id],
-                    [ug.users_group_id, p3.permission_id]]))
                     [ug.users_group_id, p3.permission_id]])
         ## DISABLE REPO CREATE ON A GROUP
         response = self.app.put(
             url('edit_user_group_default_perms', id=ug.users_group_id),
             params={'_authentication_token': self.authentication_token()})
         response.follow()
         ug = UserGroup.get_by_group_name(users_group_name)
         p = Permission.get_by_key('hg.create.none')
         p2 = Permission.get_by_key('hg.usergroup.create.false')
         p3 = Permission.get_by_key('hg.fork.none')
         # check if user has this perms, they should be here since
         # defaults are on
         perms = UserGroupToPerm.query() \
             .filter(UserGroupToPerm.users_group == ug).all()
         self.assertEqual(
             sorted([[x.users_group_id, x.permission_id, ] for x in perms]),
             sorted([[ug.users_group_id, p.permission_id],
         assert sorted([[x.users_group_id, x.permission_id, ] for x in perms]) == sorted([[ug.users_group_id, p.permission_id],
                     [ug.users_group_id, p2.permission_id],
-                    [ug.users_group_id, p3.permission_id]]))
                     [ug.users_group_id, p3.permission_id]])
         # DELETE !
         ug = UserGroup.get_by_group_name(users_group_name)
         ugid = ug.users_group_id
         response = self.app.post(url('users_group', id=ug.users_group_id),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         response = response.follow()
         gr = Session().query(UserGroup) \
             .filter(UserGroup.users_group_name == users_group_name).scalar()
-        self.assertEqual(gr, None)
+        assert gr == None
         p = Permission.get_by_key('hg.create.repository')
         perms = UserGroupToPerm.query() \
             .filter(UserGroupToPerm.users_group_id == ugid).all()
         perms = [[x.users_group_id,
                   x.permission_id, ] for x in perms]
-        self.assertEqual(perms, [])
+        assert perms == []
     def test_default_perms_enable_repository_fork_on_group(self):
         self.log_user()
         users_group_name = TEST_USER_GROUP + 'another2'
         response = self.app.post(url('users_groups'),
                                  {'users_group_name': users_group_name,
                                   'user_group_description': u'DESC',
                                   'active': True,
                                   '_authentication_token': self.authentication_token()})
         response.follow()
         ug = UserGroup.get_by_group_name(users_group_name)
         self.checkSessionFlash(response,
                                'Created user group ')
         ## ENABLE REPO CREATE ON A GROUP
         response = self.app.put(url('edit_user_group_default_perms',
                                     id=ug.users_group_id),
                                 {'fork_repo_perm': True, '_authentication_token': self.authentication_token()})
         response.follow()
         ug = UserGroup.get_by_group_name(users_group_name)
         p = Permission.get_by_key('hg.create.none')
         p2 = Permission.get_by_key('hg.usergroup.create.false')
         p3 = Permission.get_by_key('hg.fork.repository')
         # check if user has this perms, they should be here since
         # defaults are on
         perms = UserGroupToPerm.query() \
             .filter(UserGroupToPerm.users_group == ug).all()
         self.assertEqual(
             sorted([[x.users_group_id, x.permission_id, ] for x in perms]),
             sorted([[ug.users_group_id, p.permission_id],
         assert sorted([[x.users_group_id, x.permission_id, ] for x in perms]) == sorted([[ug.users_group_id, p.permission_id],
                     [ug.users_group_id, p2.permission_id],
-                    [ug.users_group_id, p3.permission_id]]))
                     [ug.users_group_id, p3.permission_id]])
         ## DISABLE REPO CREATE ON A GROUP
         response = self.app.put(url('edit_user_group_default_perms', id=ug.users_group_id),
             params={'_authentication_token': self.authentication_token()})
         response.follow()
         ug = UserGroup.get_by_group_name(users_group_name)
         p = Permission.get_by_key('hg.create.none')
         p2 = Permission.get_by_key('hg.usergroup.create.false')
         p3 = Permission.get_by_key('hg.fork.none')
         # check if user has this perms, they should be here since
         # defaults are on
         perms = UserGroupToPerm.query() \
             .filter(UserGroupToPerm.users_group == ug).all()
         self.assertEqual(
             sorted([[x.users_group_id, x.permission_id, ] for x in perms]),
             sorted([[ug.users_group_id, p.permission_id],
         assert sorted([[x.users_group_id, x.permission_id, ] for x in perms]) == sorted([[ug.users_group_id, p.permission_id],
                     [ug.users_group_id, p2.permission_id],
-                    [ug.users_group_id, p3.permission_id]]))
                     [ug.users_group_id, p3.permission_id]])
         # DELETE !
         ug = UserGroup.get_by_group_name(users_group_name)
         ugid = ug.users_group_id
         response = self.app.post(url('users_group', id=ug.users_group_id),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         response = response.follow()
         gr = Session().query(UserGroup) \
                            .filter(UserGroup.users_group_name ==
                                    users_group_name).scalar()
-        self.assertEqual(gr, None)
+        assert gr == None
         p = Permission.get_by_key('hg.fork.repository')
         perms = UserGroupToPerm.query() \
             .filter(UserGroupToPerm.users_group_id == ugid).all()
         perms = [[x.users_group_id,
                   x.permission_id, ] for x in perms]
         self.assertEqual(
             perms,
             []
+        )
         assert perms == []
     def test_delete_browser_fakeout(self):
         response = self.app.post(url('users_group', id=1),
                                  params=dict(_method='delete', _authentication_token=self.authentication_token()))
     def test_show(self):
         response = self.app.get(url('users_group', id=1))
     def test_edit(self):
         response = self.app.get(url('edit_users_group', id=1))
     def test_assign_members(self):
         pass
     def test_add_create_permission(self):
         pass
     def test_revoke_members(self):
         pass

kallithea/tests/functional/test_admin_users.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 from sqlalchemy.orm.exc import NoResultFound
 import pytest
 from kallithea.tests import *
 from kallithea.tests.fixture import Fixture
 from kallithea.controllers.admin.users import UsersController
 from kallithea.model.db import User, Permission, UserIpMap, UserApiKeys
 from kallithea.lib.auth import check_password
 from kallithea.model.user import UserModel
 from kallithea.model import validators
 from kallithea.lib import helpers as h
 from kallithea.model.meta import Session
 from webob.exc import HTTPNotFound
 fixture = Fixture()
 class TestAdminUsersController(TestControllerPytest):
     test_user_1 = 'testme'
     @classmethod
     def teardown_class(cls):
         if User.get_by_username(cls.test_user_1):
             UserModel().delete(cls.test_user_1)
             Session().commit()
     def test_index(self):
         self.log_user()
         response = self.app.get(url('users'))
         # Test response...
     def test_create(self):
         self.log_user()
         username = 'newtestuser'
         password = 'test12'
         password_confirmation = password
         name = u'name'
         lastname = u'lastname'
         email = 'mail@example.com'
         response = self.app.post(url('users'),
             {'username': username,
              'password': password,
              'password_confirmation': password_confirmation,
              'firstname': name,
              'active': True,
              'lastname': lastname,
              'extern_name': 'internal',
              'extern_type': 'internal',
              'email': email,
              '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, '''Created user <a href="/_admin/users/''')
         self.checkSessionFlash(response, '''/edit">%s</a>''' % (username))
         new_user = Session().query(User). \
             filter(User.username == username).one()
         self.assertEqual(new_user.username, username)
         self.assertEqual(check_password(password, new_user.password), True)
         self.assertEqual(new_user.name, name)
         self.assertEqual(new_user.lastname, lastname)
         self.assertEqual(new_user.email, email)
         assert new_user.username == username
         assert check_password(password, new_user.password) == True
         assert new_user.name == name
         assert new_user.lastname == lastname
         assert new_user.email == email
         response.follow()
         response = response.follow()
         response.mustcontain("""newtestuser""")
     def test_create_err(self):
         self.log_user()
         username = 'new_user'
         password = ''
         name = u'name'
         lastname = u'lastname'
         email = 'errmail.example.com'
         response = self.app.post(url('users'), {'username': username,
                                                'password': password,
                                                'name': name,
                                                'active': False,
                                                'lastname': lastname,
                                                'email': email,
                                                '_authentication_token': self.authentication_token()})
         msg = validators.ValidUsername(False, {})._messages['system_invalid_username']
         msg = h.html_escape(msg % {'username': 'new_user'})
         response.mustcontain("""<span class="error-message">%s</span>""" % msg)
         response.mustcontain("""<span class="error-message">Please enter a value</span>""")
         response.mustcontain("""<span class="error-message">An email address must contain a single @</span>""")
         def get_user():
             Session().query(User).filter(User.username == username).one()
         self.assertRaises(NoResultFound, get_user), 'found user in database'
         with pytest.raises(NoResultFound):
             get_user(), 'found user in database'
     def test_new(self):
         self.log_user()
         response = self.app.get(url('new_user'))
     @parametrize('name,attrs',
         [('firstname', {'firstname': 'new_username'}),
          ('lastname', {'lastname': 'new_username'}),
          ('admin', {'admin': True}),
          ('admin', {'admin': False}),
          ('extern_type', {'extern_type': 'ldap'}),
          ('extern_type', {'extern_type': None}),
          ('extern_name', {'extern_name': 'test'}),
          ('extern_name', {'extern_name': None}),
          ('active', {'active': False}),
          ('active', {'active': True}),
          ('email', {'email': 'someemail@example.com'}),
         # ('new_password', {'new_password': 'foobar123',
         #                   'password_confirmation': 'foobar123'})
         ])
     def test_update(self, name, attrs):
         self.log_user()
         usr = fixture.create_user(self.test_user_1, password='qweqwe',
                                   email='testme@example.com',
                                   extern_type='internal',
                                   extern_name=self.test_user_1,
                                   skip_if_exists=True)
         Session().commit()
         params = usr.get_api_data(True)
         params.update({'password_confirmation': ''})
         params.update({'new_password': ''})
         params.update(attrs)
         if name == 'email':
             params['emails'] = [attrs['email']]
         if name == 'extern_type':
             #cannot update this via form, expected value is original one
             params['extern_type'] = "internal"
         if name == 'extern_name':
             #cannot update this via form, expected value is original one
             params['extern_name'] = self.test_user_1
             # special case since this user is not
                                           # logged in yet his data is not filled
                                           # so we use creation data
         params.update({'_authentication_token': self.authentication_token()})
         response = self.app.put(url('user', id=usr.user_id), params)
         self.checkSessionFlash(response, 'User updated successfully')
         params.pop('_authentication_token')
         updated_user = User.get_by_username(self.test_user_1)
         updated_params = updated_user.get_api_data(True)
         updated_params.update({'password_confirmation': ''})
         updated_params.update({'new_password': ''})
-        self.assertEqual(params, updated_params)
+        assert params == updated_params
     def test_delete(self):
         self.log_user()
         username = 'newtestuserdeleteme'
         fixture.create_user(name=username)
         new_user = Session().query(User) \
             .filter(User.username == username).one()
         response = self.app.post(url('user', id=new_user.user_id),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'Successfully deleted user')
     def test_delete_repo_err(self):
         self.log_user()
         username = 'repoerr'
         reponame = u'repoerr_fail'
         fixture.create_user(name=username)
         fixture.create_repo(name=reponame, cur_user=username)
         new_user = Session().query(User) \
             .filter(User.username == username).one()
         response = self.app.post(url('user', id=new_user.user_id),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'User "%s" still '
                                'owns 1 repositories and cannot be removed. '
                                'Switch owners or remove those repositories: '
                                '%s' % (username, reponame))
         response = self.app.post(url('delete_repo', repo_name=reponame),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'Deleted repository %s' % reponame)
         response = self.app.post(url('user', id=new_user.user_id),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'Successfully deleted user')
     def test_delete_repo_group_err(self):
         self.log_user()
         username = 'repogrouperr'
         groupname = u'repogroup_fail'
         fixture.create_user(name=username)
         fixture.create_repo_group(name=groupname, cur_user=username)
         new_user = Session().query(User) \
             .filter(User.username == username).one()
         response = self.app.post(url('user', id=new_user.user_id),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'User "%s" still '
                                'owns 1 repository groups and cannot be removed. '
                                'Switch owners or remove those repository groups: '
                                '%s' % (username, groupname))
         # Relevant _if_ the user deletion succeeded to make sure we can render groups without owner
         # rg = RepoGroup.get_by_group_name(group_name=groupname)
         # response = self.app.get(url('repos_groups', id=rg.group_id))
         response = self.app.post(url('delete_repo_group', group_name=groupname),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'Removed repository group %s' % groupname)
         response = self.app.post(url('user', id=new_user.user_id),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'Successfully deleted user')
     def test_delete_user_group_err(self):
         self.log_user()
         username = 'usergrouperr'
         groupname = u'usergroup_fail'
         fixture.create_user(name=username)
         ug = fixture.create_user_group(name=groupname, cur_user=username)
         new_user = Session().query(User) \
             .filter(User.username == username).one()
         response = self.app.post(url('user', id=new_user.user_id),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'User "%s" still '
                                'owns 1 user groups and cannot be removed. '
                                'Switch owners or remove those user groups: '
                                '%s' % (username, groupname))
         # TODO: why do this fail?
         #response = self.app.delete(url('delete_users_group', id=groupname))
         #self.checkSessionFlash(response, 'Removed user group %s' % groupname)
         fixture.destroy_user_group(ug.users_group_id)
         response = self.app.post(url('user', id=new_user.user_id),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'Successfully deleted user')
     def test_show(self):
         response = self.app.get(url('user', id=1))
     def test_edit(self):
         self.log_user()
         user = User.get_by_username(TEST_USER_ADMIN_LOGIN)
         response = self.app.get(url('edit_user', id=user.user_id))
     def test_add_perm_create_repo(self):
         self.log_user()
         perm_none = Permission.get_by_key('hg.create.none')
         perm_create = Permission.get_by_key('hg.create.repository')
         user = UserModel().create_or_update(username='dummy', password='qwe',
                                             email='dummy', firstname=u'a',
                                             lastname=u'b')
         Session().commit()
         uid = user.user_id
         try:
             #User should have None permission on creation repository
             self.assertEqual(UserModel().has_perm(user, perm_none), False)
             self.assertEqual(UserModel().has_perm(user, perm_create), False)
             assert UserModel().has_perm(user, perm_none) == False
             assert UserModel().has_perm(user, perm_create) == False
             response = self.app.post(url('edit_user_perms', id=uid),
                                      params=dict(_method='put',
                                                  create_repo_perm=True,
                                                  _authentication_token=self.authentication_token()))
             perm_none = Permission.get_by_key('hg.create.none')
             perm_create = Permission.get_by_key('hg.create.repository')
             #User should have None permission on creation repository
             self.assertEqual(UserModel().has_perm(uid, perm_none), False)
             self.assertEqual(UserModel().has_perm(uid, perm_create), True)
             assert UserModel().has_perm(uid, perm_none) == False
             assert UserModel().has_perm(uid, perm_create) == True
         finally:
             UserModel().delete(uid)
             Session().commit()
     def test_revoke_perm_create_repo(self):
         self.log_user()
         perm_none = Permission.get_by_key('hg.create.none')
         perm_create = Permission.get_by_key('hg.create.repository')
         user = UserModel().create_or_update(username='dummy', password='qwe',
                                             email='dummy', firstname=u'a',
                                             lastname=u'b')
         Session().commit()
         uid = user.user_id
         try:
             #User should have None permission on creation repository
             self.assertEqual(UserModel().has_perm(user, perm_none), False)
             self.assertEqual(UserModel().has_perm(user, perm_create), False)
             assert UserModel().has_perm(user, perm_none) == False
             assert UserModel().has_perm(user, perm_create) == False
             response = self.app.post(url('edit_user_perms', id=uid),
                                      params=dict(_method='put', _authentication_token=self.authentication_token()))
             perm_none = Permission.get_by_key('hg.create.none')
             perm_create = Permission.get_by_key('hg.create.repository')
             #User should have None permission on creation repository
             self.assertEqual(UserModel().has_perm(uid, perm_none), True)
             self.assertEqual(UserModel().has_perm(uid, perm_create), False)
             assert UserModel().has_perm(uid, perm_none) == True
             assert UserModel().has_perm(uid, perm_create) == False
         finally:
             UserModel().delete(uid)
             Session().commit()
     def test_add_perm_fork_repo(self):
         self.log_user()
         perm_none = Permission.get_by_key('hg.fork.none')
         perm_fork = Permission.get_by_key('hg.fork.repository')
         user = UserModel().create_or_update(username='dummy', password='qwe',
                                             email='dummy', firstname=u'a',
                                             lastname=u'b')
         Session().commit()
         uid = user.user_id
         try:
             #User should have None permission on creation repository
             self.assertEqual(UserModel().has_perm(user, perm_none), False)
             self.assertEqual(UserModel().has_perm(user, perm_fork), False)
             assert UserModel().has_perm(user, perm_none) == False
             assert UserModel().has_perm(user, perm_fork) == False
             response = self.app.post(url('edit_user_perms', id=uid),
                                      params=dict(_method='put',
                                                  create_repo_perm=True,
                                                  _authentication_token=self.authentication_token()))
             perm_none = Permission.get_by_key('hg.create.none')
             perm_create = Permission.get_by_key('hg.create.repository')
             #User should have None permission on creation repository
             self.assertEqual(UserModel().has_perm(uid, perm_none), False)
             self.assertEqual(UserModel().has_perm(uid, perm_create), True)
             assert UserModel().has_perm(uid, perm_none) == False
             assert UserModel().has_perm(uid, perm_create) == True
         finally:
             UserModel().delete(uid)
             Session().commit()
     def test_revoke_perm_fork_repo(self):
         self.log_user()
         perm_none = Permission.get_by_key('hg.fork.none')
         perm_fork = Permission.get_by_key('hg.fork.repository')
         user = UserModel().create_or_update(username='dummy', password='qwe',
                                             email='dummy', firstname=u'a',
                                             lastname=u'b')
         Session().commit()
         uid = user.user_id
         try:
             #User should have None permission on creation repository
             self.assertEqual(UserModel().has_perm(user, perm_none), False)
             self.assertEqual(UserModel().has_perm(user, perm_fork), False)
             assert UserModel().has_perm(user, perm_none) == False
             assert UserModel().has_perm(user, perm_fork) == False
             response = self.app.post(url('edit_user_perms', id=uid),
                                      params=dict(_method='put', _authentication_token=self.authentication_token()))
             perm_none = Permission.get_by_key('hg.create.none')
             perm_create = Permission.get_by_key('hg.create.repository')
             #User should have None permission on creation repository
             self.assertEqual(UserModel().has_perm(uid, perm_none), True)
             self.assertEqual(UserModel().has_perm(uid, perm_create), False)
             assert UserModel().has_perm(uid, perm_none) == True
             assert UserModel().has_perm(uid, perm_create) == False
         finally:
             UserModel().delete(uid)
             Session().commit()
     def test_ips(self):
         self.log_user()
         user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
         response = self.app.get(url('edit_user_ips', id=user.user_id))
         response.mustcontain('All IP addresses are allowed')
     @parametrize('test_name,ip,ip_range,failure', [
         ('127/24', '127.0.0.1/24', '127.0.0.0 - 127.0.0.255', False),
         ('10/32', '10.0.0.10/32', '10.0.0.10 - 10.0.0.10', False),
         ('0/16', '0.0.0.0/16', '0.0.0.0 - 0.0.255.255', False),
         ('0/8', '0.0.0.0/8', '0.0.0.0 - 0.255.255.255', False),
         ('127_bad_mask', '127.0.0.1/99', '127.0.0.1 - 127.0.0.1', True),
         ('127_bad_ip', 'foobar', 'foobar', True),
     ])
     def test_add_ip(self, test_name, ip, ip_range, failure):
         self.log_user()
         user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
         user_id = user.user_id
         response = self.app.put(url('edit_user_ips', id=user_id),
                                 params=dict(new_ip=ip, _authentication_token=self.authentication_token()))
         if failure:
             self.checkSessionFlash(response, 'Please enter a valid IPv4 or IPv6 address')
             response = self.app.get(url('edit_user_ips', id=user_id))
             response.mustcontain(no=[ip])
             response.mustcontain(no=[ip_range])
         else:
             response = self.app.get(url('edit_user_ips', id=user_id))
             response.mustcontain(ip)
             response.mustcontain(ip_range)
         ## cleanup
         for del_ip in UserIpMap.query().filter(UserIpMap.user_id == user_id).all():
             Session().delete(del_ip)
             Session().commit()
     def test_delete_ip(self):
         self.log_user()
         user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
         user_id = user.user_id
         ip = '127.0.0.1/32'
         ip_range = '127.0.0.1 - 127.0.0.1'
         new_ip = UserModel().add_extra_ip(user_id, ip)
         Session().commit()
         new_ip_id = new_ip.ip_id
         response = self.app.get(url('edit_user_ips', id=user_id))
         response.mustcontain(ip)
         response.mustcontain(ip_range)
         self.app.post(url('edit_user_ips', id=user_id),
                       params=dict(_method='delete', del_ip_id=new_ip_id, _authentication_token=self.authentication_token()))
         response = self.app.get(url('edit_user_ips', id=user_id))
         response.mustcontain('All IP addresses are allowed')
         response.mustcontain(no=[ip])
         response.mustcontain(no=[ip_range])
     def test_api_keys(self):
         self.log_user()
         user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
         response = self.app.get(url('edit_user_api_keys', id=user.user_id))
         response.mustcontain(user.api_key)
         response.mustcontain('Expires: Never')
     @parametrize('desc,lifetime', [
         ('forever', -1),
         ('5mins', 60*5),
         ('30days', 60*60*24*30),
     ])
     def test_add_api_keys(self, desc, lifetime):
         self.log_user()
         user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
         user_id = user.user_id
         response = self.app.post(url('edit_user_api_keys', id=user_id),
                  {'description': desc, 'lifetime': lifetime, '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'API key successfully created')
         try:
             response = response.follow()
             user = User.get(user_id)
             for api_key in user.api_keys:
                 response.mustcontain(api_key)
         finally:
             for api_key in UserApiKeys.query().filter(UserApiKeys.user_id == user_id).all():
                 Session().delete(api_key)
                 Session().commit()
     def test_remove_api_key(self):
         self.log_user()
         user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
         user_id = user.user_id
         response = self.app.post(url('edit_user_api_keys', id=user_id),
                 {'description': 'desc', 'lifetime': -1, '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'API key successfully created')
         response = response.follow()
         #now delete our key
         keys = UserApiKeys.query().filter(UserApiKeys.user_id == user_id).all()
-        self.assertEqual(1, len(keys))
+        assert 1 == len(keys)
         response = self.app.post(url('edit_user_api_keys', id=user_id),
                  {'_method': 'delete', 'del_api_key': keys[0].api_key, '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'API key successfully deleted')
         keys = UserApiKeys.query().filter(UserApiKeys.user_id == user_id).all()
-        self.assertEqual(0, len(keys))
+        assert 0 == len(keys)
     def test_reset_main_api_key(self):
         self.log_user()
         user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
         user_id = user.user_id
         api_key = user.api_key
         response = self.app.get(url('edit_user_api_keys', id=user_id))
         response.mustcontain(api_key)
         response.mustcontain('Expires: Never')
         response = self.app.post(url('edit_user_api_keys', id=user_id),
                  {'_method': 'delete', 'del_api_key_builtin': api_key, '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'API key successfully reset')
         response = response.follow()
         response.mustcontain(no=[api_key])
 class TestAdminUsersController_unittest(object):
     """ Unit tests for the users controller """
     def test_get_user_or_raise_if_default(self, monkeypatch):
         # flash complains about an non-existing session
         def flash_mock(*args, **kwargs):
             pass
         monkeypatch.setattr(h, 'flash', flash_mock)
         u = UsersController()
         # a regular user should work correctly
         user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
         assert u._get_user_or_raise_if_default(user.user_id) == user
         # the default user should raise
         with pytest.raises(HTTPNotFound):
             u._get_user_or_raise_if_default(User.get_default_user().user_id)
 class TestAdminUsersControllerForDefaultUser(TestControllerPytest):
     """
     Edit actions on the default user are not allowed.
     Validate that they throw a 404 exception.
     """
     def test_edit_default_user(self):
         self.log_user()
         user = User.get_default_user()
         response = self.app.get(url('edit_user', id=user.user_id), status=404)
     def test_edit_advanced_default_user(self):
         self.log_user()
         user = User.get_default_user()
         response = self.app.get(url('edit_user_advanced', id=user.user_id), status=404)
     # API keys
     def test_edit_api_keys_default_user(self):
         self.log_user()
         user = User.get_default_user()
         response = self.app.get(url('edit_user_api_keys', id=user.user_id), status=404)
     def test_add_api_keys_default_user(self):
         self.log_user()
         user = User.get_default_user()
         response = self.app.post(url('edit_user_api_keys', id=user.user_id),
                  {'_method': 'put', '_authentication_token': self.authentication_token()}, status=404)
     def test_delete_api_keys_default_user(self):
         self.log_user()
         user = User.get_default_user()
         response = self.app.post(url('edit_user_api_keys', id=user.user_id),
                  {'_method': 'delete', '_authentication_token': self.authentication_token()}, status=404)
     # Permissions
     def test_edit_perms_default_user(self):
         self.log_user()
         user = User.get_default_user()
         response = self.app.get(url('edit_user_perms', id=user.user_id), status=404)
     def test_update_perms_default_user(self):
         self.log_user()
         user = User.get_default_user()
         response = self.app.post(url('edit_user_perms', id=user.user_id),
                  {'_method': 'put', '_authentication_token': self.authentication_token()}, status=404)
     # Emails
     def test_edit_emails_default_user(self):
         self.log_user()
         user = User.get_default_user()
         response = self.app.get(url('edit_user_emails', id=user.user_id), status=404)
     def test_add_emails_default_user(self):
         self.log_user()
         user = User.get_default_user()
         response = self.app.post(url('edit_user_emails', id=user.user_id),
                  {'_method': 'put', '_authentication_token': self.authentication_token()}, status=404)
     def test_delete_emails_default_user(self):
         self.log_user()
         user = User.get_default_user()
         response = self.app.post(url('edit_user_emails', id=user.user_id),
                  {'_method': 'delete', '_authentication_token': self.authentication_token()}, status=404)
     # IP addresses
     # Add/delete of IP addresses for the default user is used to maintain
     # the global IP whitelist and thus allowed. Only 'edit' is forbidden.
     def test_edit_ip_default_user(self):
         self.log_user()
         user = User.get_default_user()
         response = self.app.get(url('edit_user_ips', id=user.user_id), status=404)

kallithea/tests/functional/test_changelog.py

➞

Show inline comments

 from kallithea.tests import *
 class TestChangelogController(TestControllerPytest):
     def test_index_hg(self):
         self.log_user()
         response = self.app.get(url(controller='changelog', action='index',
                                     repo_name=HG_REPO))
         response.mustcontain('''id="chg_20" class="container mergerow"''')
         response.mustcontain(
             """<input class="changeset_range" """
             """id="7b22a518347bb9bc19679f6af07cd0a61bfe16e7" """
             """name="7b22a518347bb9bc19679f6af07cd0a61bfe16e7" """
             """type="checkbox" value="1" />"""
+        )
         #rev 640: code garden
         response.mustcontain(
             """<span class="changeset_hash">r640:0a4e54a44604</span>"""
+        )
         response.mustcontain("""code garden""")
         response.mustcontain("""var jsdata = [[[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 1, 2, 0], [0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0], [1, 1, 2, 0]], 0, 0], [[0, 1], [[0, 0, 2, 0], [1, 0, 2, 0]], 0, 0], [[0, 2], [[0, 1, 3, 0], [0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 3, 0]], 0, 0], [[1, 3], [[0, 0, 2, 0], [1, 1, 3, 0]], 0, 0], [[1, 3], [[0, 0, 2, 0], [1, 0, 2, 0]], 0, 0], [[0, 2], [[0, 1, 4, 0], [0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 1, 4, 0], [1, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 1, 4, 0], [1, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 1, 5, 0], [0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 5, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 5, 0]], 0, 0], [[1, 5], [[0, 0, 2, 0], [1, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 1, 6, 0], [0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 6, 0]], 0, 0], [[1, 6], [[0, 0, 2, 0], [1, 1, 6, 0]], 0, 0], [[1, 6], [[0, 0, 2, 0], [1, 1, 6, 0], [1, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 6, 0]], 0, 0], [[1, 6], [[0, 0, 2, 0], [1, 0, 2, 0]], 0, 0], [[0, 2], [[0, 1, 7, 0], [0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 7, 0]], 0, 0], [[1, 7], [[0, 0, 2, 0], [1, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 1, 8, 0], [0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 8, 0]], 0, 0], [[1, 8], [[0, 0, 2, 0], [1, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 1, 9, 0], [0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 1, 10, 0], [0, 0, 2, 0], [1, 2, 9, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 10, 0], [2, 2, 9, 0]], 0, 0], [[2, 9], [[0, 0, 2, 0], [1, 1, 10, 0], [2, 1, 10, 0]], 0, 0], [[1, 10], [[0, 0, 2, 0], [1, 1, 10, 0]], 0, 0], [[1, 10], [[0, 0, 2, 0], [1, 1, 10, 0]], 0, 0], [[1, 10], [[0, 0, 2, 0], [1, 1, 10, 0]], 0, 0], [[1, 10], [[0, 0, 2, 0], [1, 1, 10, 0]], 0, 0], [[1, 10], [[0, 0, 2, 0], [1, 1, 10, 0]], 0, 0], [[1, 10], [[0, 0, 2, 0], [1, 1, 10, 0]], 0, 0], [[1, 10], [[0, 0, 2, 0], [1, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[1, 11], [[0, 0, 2, 0], [1, 1, 11, 0]], 1, 0], [[2, 12], [[0, 0, 2, 0], [2, 1, 12, 0]], 1, 0], [[0, 2], [[0, 1, 13, 0], [0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 13, 0]], 0, 0], [[1, 13], [[0, 0, 2, 0], [1, 1, 13, 0], [1, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 13, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 13, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 13, 0]], 0, 0], [[1, 13], [[0, 0, 2, 0], [1, 1, 13, 0], [1, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 1, 14, 0], [0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0]];""")
     def test_index_pagination_hg(self):
         self.log_user()
         #pagination
         self.app.get(url(controller='changelog', action='index',
                                     repo_name=HG_REPO), {'page': 1})
         self.app.get(url(controller='changelog', action='index',
                                     repo_name=HG_REPO), {'page': 2})
         self.app.get(url(controller='changelog', action='index',
                                     repo_name=HG_REPO), {'page': 3})
         self.app.get(url(controller='changelog', action='index',
                                     repo_name=HG_REPO), {'page': 4})
         self.app.get(url(controller='changelog', action='index',
                                     repo_name=HG_REPO), {'page': 5})
         response = self.app.get(url(controller='changelog', action='index',
                                     repo_name=HG_REPO), {'page': 6, 'size': 20})
         # Test response after pagination...
         response.mustcontain(
             """<input class="changeset_range" """
             """id="22baf968d547386b9516965ce89d189665003a31" """
             """name="22baf968d547386b9516965ce89d189665003a31" """
             """type="checkbox" value="1" />"""
+        )
         response.mustcontain(
             """<span class="changeset_hash">r539:22baf968d547</span>"""
+        )
     def test_index_git(self):
         self.log_user()
         response = self.app.get(url(controller='changelog', action='index',
                                     repo_name=GIT_REPO))
         response.mustcontain('''id="chg_20" class="container "''') # why no mergerow for git?
         response.mustcontain(
             """<input class="changeset_range" """
             """id="95f9a91d775b0084b2368ae7779e44931c849c0e" """
             """name="95f9a91d775b0084b2368ae7779e44931c849c0e" """
             """type="checkbox" value="1" />"""
+        )
         response.mustcontain(
             """<span class="changeset_hash">r613:95f9a91d775b</span>"""
+        )
         response.mustcontain("""fixing stupid typo in context for mercurial""")
         response.mustcontain("""var jsdata = [[[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 1, 2, 0], [0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0], [1, 1, 2, 0]], 0, 0], [[0, 1], [[0, 0, 2, 0], [1, 0, 2, 0]], 0, 0], [[0, 2], [[0, 1, 3, 0], [0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 3, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 3, 0]], 0, 0], [[1, 3], [[0, 0, 2, 0], [1, 0, 2, 0]], 0, 0], [[0, 2], [[0, 1, 4, 0], [0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 1, 4, 0], [1, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[0, 2], [[0, 0, 4, 0], [1, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 1, 5, 0], [0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0], [1, 1, 5, 0]], 0, 0], [[1, 5], [[0, 0, 4, 0], [1, 1, 5, 0]], 0, 0], [[1, 5], [[0, 0, 4, 0], [1, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 1, 6, 0], [0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0], [1, 1, 6, 0]], 0, 0], [[1, 6], [[0, 0, 4, 0], [1, 1, 6, 0]], 0, 0], [[1, 6], [[0, 0, 4, 0], [1, 1, 6, 0], [1, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0], [1, 1, 6, 0]], 0, 0], [[1, 6], [[0, 0, 4, 0], [1, 0, 4, 0]], 0, 0], [[0, 4], [[0, 1, 7, 0], [0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0], [1, 1, 7, 0]], 0, 0], [[1, 7], [[0, 0, 4, 0], [1, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 1, 8, 0], [0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0], [1, 1, 8, 0]], 0, 0], [[1, 8], [[0, 0, 4, 0], [1, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 1, 9, 0], [0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0], [1, 1, 9, 0]], 0, 0], [[1, 9], [[0, 0, 4, 0], [1, 0, 4, 0], [1, 1, 9, 0]], 0, 0], [[1, 9], [[0, 0, 4, 0], [1, 1, 9, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0], [1, 1, 9, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0], [1, 1, 9, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0], [1, 1, 9, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0], [1, 1, 9, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0], [1, 1, 9, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0], [1, 1, 9, 0]], 0, 0], [[0, 4], [[0, 0, 9, 0], [1, 0, 9, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0]], 0, 0], [[0, 9], [[0, 1, 10, 0], [0, 0, 9, 0]], 0, 0], [[0, 9], [[0, 1, 11, 0], [0, 0, 9, 0], [1, 2, 10, 0]], 0, 0], [[2, 10], [[0, 0, 9, 0], [1, 1, 11, 0], [2, 0, 9, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0], [1, 1, 11, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0], [1, 1, 11, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0], [1, 1, 11, 0]], 0, 0], [[1, 11], [[0, 0, 9, 0], [1, 0, 9, 0], [1, 1, 11, 0]], 0, 0], [[1, 11], [[0, 0, 9, 0], [1, 1, 11, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0], [1, 1, 11, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0], [1, 1, 11, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0], [1, 1, 11, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0], [1, 1, 11, 0]], 0, 0], [[0, 9], [[0, 1, 11, 0], [0, 0, 9, 0], [1, 1, 11, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0], [1, 1, 11, 0]], 0, 0], [[1, 11], [[0, 0, 9, 0], [1, 0, 9, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0]], 0, 0]];""")
 #        response.mustcontain(
 #            """<div id="changed_total_5e204e7583b9c8e7b93a020bd036564b1e731dae" """
 #            """style="float:right;" class="changed_total tooltip" """
 #            """title="Affected number of files, click to show """
 #            """more details">3</div>"""
 #        )
     def test_index_pagination_git(self):
         self.log_user()
         #pagination
         self.app.get(url(controller='changelog', action='index',
                                     repo_name=GIT_REPO), {'page': 1})
         self.app.get(url(controller='changelog', action='index',
                                     repo_name=GIT_REPO), {'page': 2})
         self.app.get(url(controller='changelog', action='index',
                                     repo_name=GIT_REPO), {'page': 3})
         self.app.get(url(controller='changelog', action='index',
                                     repo_name=GIT_REPO), {'page': 4})
         self.app.get(url(controller='changelog', action='index',
                                     repo_name=GIT_REPO), {'page': 5})
         response = self.app.get(url(controller='changelog', action='index',
                                     repo_name=GIT_REPO), {'page': 6, 'size': 20})
         # Test response after pagination...
         response.mustcontain(
             """<input class="changeset_range" """
             """id="636ed213f2f11ef91071b9c24f2d5e6bd01a6ed5" """
             """name="636ed213f2f11ef91071b9c24f2d5e6bd01a6ed5" """
             """type="checkbox" value="1" />"""
+        )
         response.mustcontain(
             """<span class="changeset_hash">r515:636ed213f2f1</span>"""
+        )
     def test_index_hg_with_filenode(self):
         self.log_user()
         response = self.app.get(url(controller='changelog', action='index',
                                     revision='tip', f_path='/vcs/exceptions.py',
                                     repo_name=HG_REPO))
         #history commits messages
         response.mustcontain('Added exceptions module, this time for real')
         response.mustcontain('Added not implemented hg backend test case')
         response.mustcontain('Added BaseChangeset class')
         # Test response...
     def test_index_git_with_filenode(self):
         self.log_user()
         response = self.app.get(url(controller='changelog', action='index',
                                     revision='tip', f_path='/vcs/exceptions.py',
                                     repo_name=GIT_REPO))
         #history commits messages
         response.mustcontain('Added exceptions module, this time for real')
         response.mustcontain('Added not implemented hg backend test case')
         response.mustcontain('Added BaseChangeset class')
     def test_index_hg_with_filenode_that_is_dirnode(self):
         self.log_user()
         response = self.app.get(url(controller='changelog', action='index',
                                     revision='tip', f_path='/tests',
                                     repo_name=HG_REPO))
-        self.assertEqual(response.status, '302 Found')
+        assert response.status == '302 Found'
     def test_index_git_with_filenode_that_is_dirnode(self):
         self.log_user()
         response = self.app.get(url(controller='changelog', action='index',
                                     revision='tip', f_path='/tests',
                                     repo_name=GIT_REPO))
-        self.assertEqual(response.status, '302 Found')
+        assert response.status == '302 Found'
     def test_index_hg_with_filenode_not_existing(self):
         self.log_user()
         response = self.app.get(url(controller='changelog', action='index',
                                     revision='tip', f_path='/wrong_path',
                                     repo_name=HG_REPO))
-        self.assertEqual(response.status, '302 Found')
+        assert response.status == '302 Found'
     def test_index_git_with_filenode_not_existing(self):
         self.log_user()
         response = self.app.get(url(controller='changelog', action='index',
                                     revision='tip', f_path='/wrong_path',
                                     repo_name=GIT_REPO))
-        self.assertEqual(response.status, '302 Found')
+        assert response.status == '302 Found'

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(TestControllerPytest):
     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')
+        assert 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)
+        assert 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)
         assert Notification.query().count() == 1
         assert ChangesetComment.query().count() == 1
         notification = Notification.query().all()[0]
         ID = ChangesetComment.query().first().comment_id
         self.assertEqual(notification.type_,
                          Notification.TYPE_CHANGESET_COMMENT)
         assert 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)
+        assert 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')
+        assert 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)
+        assert 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)
         assert Notification.query().count() == 1
         assert ChangesetComment.query().count() == 1
         notification = Notification.query().all()[0]
         ID = ChangesetComment.query().first().comment_id
         self.assertEqual(notification.type_,
                          Notification.TYPE_CHANGESET_COMMENT)
         assert 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)
+        assert 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')
+        assert 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)
+        assert ChangesetComment.query().count() == 1
         response.mustcontain(
             '''<div class="comments-number">'''
             ''' 1 comment (0 inline, 1 general)'''
+        )
-        self.assertEqual(Notification.query().count(), 2)
+        assert 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])
+        assert 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)
+        assert 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)
+        assert 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(TestControllerPytest):
     def setup_method(self, method):
         self.r1_id = None
         self.r2_id = None
     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_incoming_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_incoming_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
         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_cherry_pick_changesets_from_bottom(self):
 #        repo1:
 #            cs0:
 #            cs1:
 #        repo1-fork- in which we will cherry pick bottom changesets
 #            cs0:
 #            cs1:
 #            cs2: x
 #            cs3: x
 #            cs4: x
 #            cs5:
         #make repo1, and cs1+cs2
         self.log_user()
         repo1 = fixture.create_repo(u'repo1', 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)
         cs1 = fixture.commit_change(repo1.repo_name, filename='file1',
                 content='line1\nline2\n', message='commit2', vcs_type='hg',
                 parent=cs0)
         #fork this repo
         repo2 = fixture.create_fork(u'repo1', u'repo1-fork')
         self.r2_id = repo2.repo_id
         #now make cs3-6
         cs2 = fixture.commit_change(repo1.repo_name, filename='file1',
                 content='line1\nline2\nline3\n', message='commit3',
                 vcs_type='hg', parent=cs1)
         cs3 = fixture.commit_change(repo1.repo_name, filename='file1',
                 content='line1\nline2\nline3\nline4\n', message='commit4',
                 vcs_type='hg', parent=cs2)
         cs4 = fixture.commit_change(repo1.repo_name, filename='file1',
                 content='line1\nline2\nline3\nline4\nline5\n',
                 message='commit5', vcs_type='hg', parent=cs3)
         cs5 = fixture.commit_change(repo1.repo_name, filename='file1',
                 content='line1\nline2\nline3\nline4\nline5\nline6\n',
                 message='commit6', vcs_type='hg', parent=cs4)
         response = self.app.get(url('compare_url',
                                     repo_name=repo2.repo_name,
                                     org_ref_type="rev",
                                     org_ref_name=cs1.short_id,  # parent of cs2, in repo2
                                     other_repo=repo1.repo_name,
                                     other_ref_type="rev",
                                     other_ref_name=cs4.short_id,
                                     merge='True',
                                     ))
         response.mustcontain('%s@%s' % (repo2.repo_name, cs1.short_id))
         response.mustcontain('%s@%s' % (repo1.repo_name, cs4.short_id))
         response.mustcontain("""Showing 3 commits""")
         response.mustcontain("""1 file changed with 3 insertions and 0 deletions""")
         response.mustcontain(_commit_div(cs2.raw_id, 'commit3'))
         response.mustcontain(_commit_div(cs3.raw_id, 'commit4'))
         response.mustcontain(_commit_div(cs4.raw_id, 'commit5'))
         response.mustcontain("""<a href="/%s/changeset/%s">r2:%s</a>""" % (repo1.repo_name, cs2.raw_id, cs2.short_id))
         response.mustcontain("""<a href="/%s/changeset/%s">r3:%s</a>""" % (repo1.repo_name, cs3.raw_id, cs3.short_id))
         response.mustcontain("""<a href="/%s/changeset/%s">r4:%s</a>""" % (repo1.repo_name, cs4.raw_id, cs4.short_id))
         ## files
         response.mustcontain("""#C--826e8142e6ba">file1</a>""")
     def test_compare_cherry_pick_changesets_from_top(self):
 #        repo1:
 #            cs0:
 #            cs1:
 #        repo1-fork- in which we will cherry pick bottom changesets
 #            cs0:
 #            cs1:
 #            cs2:
 #            cs3: x
 #            cs4: x
 #            cs5: x
+#
         #make repo1, and cs1+cs2
         self.log_user()
         repo1 = fixture.create_repo(u'repo1', 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)
         cs1 = fixture.commit_change(repo1.repo_name, filename='file1',
                 content='line1\nline2\n', message='commit2', vcs_type='hg',
                 parent=cs0)
         #fork this repo
         repo2 = fixture.create_fork(u'repo1', u'repo1-fork')
         self.r2_id = repo2.repo_id
         #now make cs3-6
         cs2 = fixture.commit_change(repo1.repo_name, filename='file1',
                 content='line1\nline2\nline3\n', message='commit3',
                 vcs_type='hg', parent=cs1)
         cs3 = fixture.commit_change(repo1.repo_name, filename='file1',
                 content='line1\nline2\nline3\nline4\n', message='commit4',
                 vcs_type='hg', parent=cs2)
         cs4 = fixture.commit_change(repo1.repo_name, filename='file1',
                 content='line1\nline2\nline3\nline4\nline5\n',
                 message='commit5', vcs_type='hg', parent=cs3)
         cs5 = fixture.commit_change(repo1.repo_name, filename='file1',
                 content='line1\nline2\nline3\nline4\nline5\nline6\n',
                 message='commit6', vcs_type='hg', parent=cs4)
         response = self.app.get(url('compare_url',
                                     repo_name=repo1.repo_name,
                                     org_ref_type="rev",
                                     org_ref_name=cs2.short_id, # parent of cs3, not in repo2
                                     other_ref_type="rev",
                                     other_ref_name=cs5.short_id,
                                     merge='1',))
         response.mustcontain('%s@%s' % (repo1.repo_name, cs2.short_id))
         response.mustcontain('%s@%s' % (repo1.repo_name, cs5.short_id))
         response.mustcontain("""Showing 3 commits""")
         response.mustcontain("""1 file changed with 3 insertions and 0 deletions""")
         response.mustcontain(_commit_div(cs3.raw_id, 'commit4'))
         response.mustcontain(_commit_div(cs4.raw_id, 'commit5'))
         response.mustcontain(_commit_div(cs5.raw_id, 'commit6'))
         response.mustcontain("""<a href="/%s/changeset/%s">r3:%s</a>""" % (repo1.repo_name, cs3.raw_id, cs3.short_id))
         response.mustcontain("""<a href="/%s/changeset/%s">r4:%s</a>""" % (repo1.repo_name, cs4.raw_id, cs4.short_id))
         response.mustcontain("""<a href="/%s/changeset/%s">r5:%s</a>""" % (repo1.repo_name, cs5.raw_id, cs5.short_id))
         ## files
         response.mustcontain("""#C--826e8142e6ba">file1</a>""")
     def test_compare_cherry_pick_changeset_mixed_branches(self):
         #TODO: write this
         assert 1
     def test_compare_remote_branches_hg(self):
         self.log_user()
         repo2 = fixture.create_fork(HG_REPO, HG_FORK)
         self.r2_id = repo2.repo_id
         rev1 = '56349e29c2af'
         rev2 = '7d4bc8ec6be5'
         response = self.app.get(url('compare_url',
                                     repo_name=HG_REPO,
                                     org_ref_type="rev",
                                     org_ref_name=rev1,
                                     other_ref_type="rev",
                                     other_ref_name=rev2,
                                     other_repo=HG_FORK,
                                     merge='1',))
         response.mustcontain('%s@%s' % (HG_REPO, rev1))
         response.mustcontain('%s@%s' % (HG_FORK, rev2))
         ## outgoing changesets between those revisions
         response.mustcontain("""<a href="/%s/changeset/2dda4e345facb0ccff1a191052dd1606dba6781d">r4:2dda4e345fac</a>""" % (HG_FORK))
         response.mustcontain("""<a href="/%s/changeset/6fff84722075f1607a30f436523403845f84cd9e">r5:6fff84722075</a>""" % (HG_FORK))
         response.mustcontain("""<a href="/%s/changeset/7d4bc8ec6be56c0f10425afb40b6fc315a4c25e7">r6:%s</a>""" % (HG_FORK, rev2))
         ## files
         response.mustcontain("""<a href="#C--9c390eb52cd6">vcs/backends/hg.py</a>""")
         response.mustcontain("""<a href="#C--41b41c1f2796">vcs/backends/__init__.py</a>""")
         response.mustcontain("""<a href="#C--2f574d260608">vcs/backends/base.py</a>""")
     def test_compare_remote_branches_git(self):
         self.log_user()
         repo2 = fixture.create_fork(GIT_REPO, GIT_FORK)
         self.r2_id = repo2.repo_id
         rev1 = '102607b09cdd60e2793929c4f90478be29f85a17'
         rev2 = 'd7e0d30fbcae12c90680eb095a4f5f02505ce501'
         response = self.app.get(url('compare_url',
                                     repo_name=GIT_REPO,
                                     org_ref_type="rev",
                                     org_ref_name=rev1,
                                     other_ref_type="rev",
                                     other_ref_name=rev2,
                                     other_repo=GIT_FORK,
                                     merge='1',))
         response.mustcontain('%s@%s' % (GIT_REPO, rev1))
         response.mustcontain('%s@%s' % (GIT_FORK, rev2))
         ## outgoing changesets between those revisions
         response.mustcontain("""<a href="/%s/changeset/49d3fd156b6f7db46313fac355dca1a0b94a0017">r4:49d3fd156b6f</a>""" % (GIT_FORK))
         response.mustcontain("""<a href="/%s/changeset/2d1028c054665b962fa3d307adfc923ddd528038">r5:2d1028c05466</a>""" % (GIT_FORK))
         response.mustcontain("""<a href="/%s/changeset/d7e0d30fbcae12c90680eb095a4f5f02505ce501">r6:%s</a>""" % (GIT_FORK, rev2[:12]))
         ## files
         response.mustcontain("""<a href="#C--9c390eb52cd6">vcs/backends/hg.py</a>""")
         response.mustcontain("""<a href="#C--41b41c1f2796">vcs/backends/__init__.py</a>""")
         response.mustcontain("""<a href="#C--2f574d260608">vcs/backends/base.py</a>""")
     def test_org_repo_new_commits_after_forking_simple_diff_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
         r1_name = repo1.repo_name
         cs0 = fixture.commit_change(repo=r1_name, filename='file1',
                 content='line1', message='commit1', vcs_type='hg', newfile=True)
         Session().commit()
-        self.assertEqual(repo1.scm_instance.revisions, [cs0.raw_id])
+        assert repo1.scm_instance.revisions == [cs0.raw_id]
         #fork the repo1
         repo2 = fixture.create_repo(u'one-fork', repo_type='hg',
                                     repo_description='diff-test',
                                     cur_user=TEST_USER_ADMIN_LOGIN,
                                     clone_uri=repo1.repo_full_path,
                                     fork_of='one')
         Session().commit()
-        self.assertEqual(repo2.scm_instance.revisions, [cs0.raw_id])
+        assert repo2.scm_instance.revisions == [cs0.raw_id]
         self.r2_id = repo2.repo_id
         r2_name = repo2.repo_name
         cs1 = fixture.commit_change(repo=r2_name, filename='file1-fork',
                 content='file1-line1-from-fork', message='commit1-fork',
                 vcs_type='hg', parent=repo2.scm_instance[-1], newfile=True)
         cs2 = fixture.commit_change(repo=r2_name, filename='file2-fork',
                 content='file2-line1-from-fork', message='commit2-fork',
                 vcs_type='hg', parent=cs1, newfile=True)
         cs3 = fixture.commit_change(repo=r2_name, filename='file3-fork',
                 content='file3-line1-from-fork', message='commit3-fork',
                 vcs_type='hg', parent=cs2, newfile=True)
         #compare !
         rev1 = 'default'
         rev2 = 'default'
         response = self.app.get(url('compare_url',
                                     repo_name=r2_name,
                                     org_ref_type="branch",
                                     org_ref_name=rev1,
                                     other_ref_type="branch",
                                     other_ref_name=rev2,
                                     other_repo=r1_name,
                                     merge='1',))
         response.mustcontain('%s@%s' % (r2_name, rev1))
         response.mustcontain('%s@%s' % (r1_name, rev2))
         response.mustcontain('No files')
         response.mustcontain('No changesets')
         cs0 = fixture.commit_change(repo=r1_name, filename='file2',
                 content='line1-added-after-fork', message='commit2-parent',
                 vcs_type='hg', parent=None, newfile=True)
         #compare !
         rev1 = 'default'
         rev2 = 'default'
         response = self.app.get(url('compare_url',
                                     repo_name=r2_name,
                                     org_ref_type="branch",
                                     org_ref_name=rev1,
                                     other_ref_type="branch",
                                     other_ref_name=rev2,
                                     other_repo=r1_name,
                                     merge='1',
                                     ))
         response.mustcontain('%s@%s' % (r2_name, rev1))
         response.mustcontain('%s@%s' % (r1_name, rev2))
         response.mustcontain("""commit2-parent""")
         response.mustcontain("""1 file changed with 1 insertions and 0 deletions""")
         response.mustcontain("""line1-added-after-fork""")
     def test_org_repo_new_commits_after_forking_simple_diff_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
         r1_name = repo1.repo_name
         cs0 = fixture.commit_change(repo=r1_name, filename='file1',
                 content='line1', message='commit1', vcs_type='git',
                 newfile=True)
         Session().commit()
-        self.assertEqual(repo1.scm_instance.revisions, [cs0.raw_id])
+        assert repo1.scm_instance.revisions == [cs0.raw_id]
         #fork the repo1
         repo2 = fixture.create_repo(u'one-git-fork', repo_type='git',
                                     repo_description='diff-test',
                                     cur_user=TEST_USER_ADMIN_LOGIN,
                                     clone_uri=repo1.repo_full_path,
                                     fork_of='one-git')
         Session().commit()
-        self.assertEqual(repo2.scm_instance.revisions, [cs0.raw_id])
+        assert repo2.scm_instance.revisions == [cs0.raw_id]
         self.r2_id = repo2.repo_id
         r2_name = repo2.repo_name
         cs1 = fixture.commit_change(repo=r2_name, filename='file1-fork',
                 content='file1-line1-from-fork', message='commit1-fork',
                 vcs_type='git', parent=repo2.scm_instance[-1], newfile=True)
         cs2 = fixture.commit_change(repo=r2_name, filename='file2-fork',
                 content='file2-line1-from-fork', message='commit2-fork',
                 vcs_type='git', parent=cs1, newfile=True)
         cs3 = fixture.commit_change(repo=r2_name, filename='file3-fork',
                 content='file3-line1-from-fork', message='commit3-fork',
                 vcs_type='git', parent=cs2, newfile=True)
         #compare !
         rev1 = 'master'
         rev2 = 'master'
         response = self.app.get(url('compare_url',
                                     repo_name=r2_name,
                                     org_ref_type="branch",
                                     org_ref_name=rev1,
                                     other_ref_type="branch",
                                     other_ref_name=rev2,
                                     other_repo=r1_name,
                                     merge='1',))
         response.mustcontain('%s@%s' % (r2_name, rev1))
         response.mustcontain('%s@%s' % (r1_name, rev2))
         response.mustcontain('No files')
         response.mustcontain('No changesets')
         cs0 = fixture.commit_change(repo=r1_name, filename='file2',
                 content='line1-added-after-fork', message='commit2-parent',
                 vcs_type='git', parent=None, newfile=True)
         #compare !
         rev1 = 'master'
         rev2 = 'master'
         response = self.app.get(url('compare_url',
                                     repo_name=r2_name,
                                     org_ref_type="branch",
                                     org_ref_name=rev1,
                                     other_ref_type="branch",
                                     other_ref_name=rev2,
                                     other_repo=r1_name,
                                     merge='1',
                                     ))
         response.mustcontain('%s@%s' % (r2_name, rev1))
         response.mustcontain('%s@%s' % (r1_name, rev2))
         response.mustcontain("""commit2-parent""")
         response.mustcontain("""1 file changed with 1 insertions and 0 deletions""")
         response.mustcontain("""line1-added-after-fork""")

kallithea/tests/functional/test_files.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 import os
 import posixpath
 import mimetypes
 from kallithea.tests import *
 from kallithea.model.db import Repository
 from kallithea.model.meta import Session
 from kallithea.tests.fixture import Fixture
 fixture = Fixture()
 ARCHIVE_SPECS = {
     '.tar.bz2': ('application/x-bzip2', 'tbz2', ''),
     '.tar.gz': ('application/x-gzip', 'tgz', ''),
     '.zip': ('application/zip', 'zip', ''),
+}
 HG_NODE_HISTORY = fixture.load_resource('hg_node_history_response.json')
 GIT_NODE_HISTORY = fixture.load_resource('git_node_history_response.json')
 def _set_downloads(repo_name, set_to):
     repo = Repository.get_by_repo_name(repo_name)
     repo.enable_downloads = set_to
     Session().add(repo)
     Session().commit()
 class TestFilesController(TestControllerPytest):
     def test_index(self):
         self.log_user()
         response = self.app.get(url(controller='files', action='index',
                                     repo_name=HG_REPO,
                                     revision='tip',
                                     f_path='/'))
         # Test response...
         response.mustcontain('<a class="browser-dir ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/docs"><i class="icon-folder-open"></i><span>docs</span></a>' % HG_REPO)
         response.mustcontain('<a class="browser-dir ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/vcs"><i class="icon-folder-open"></i><span>vcs</span></a>' % HG_REPO)
         response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/.gitignore"><i class="icon-doc"></i><span>.gitignore</span></a>' % HG_REPO)
         response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/.hgignore"><i class="icon-doc"></i><span>.hgignore</span></a>' % HG_REPO)
         response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/.hgtags"><i class="icon-doc"></i><span>.hgtags</span></a>' % HG_REPO)
         response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/.travis.yml"><i class="icon-doc"></i><span>.travis.yml</span></a>' % HG_REPO)
         response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/MANIFEST.in"><i class="icon-doc"></i><span>MANIFEST.in</span></a>' % HG_REPO)
         response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/README.rst"><i class="icon-doc"></i><span>README.rst</span></a>' % HG_REPO)
         response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/run_test_and_report.sh"><i class="icon-doc"></i><span>run_test_and_report.sh</span></a>' % HG_REPO)
         response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/setup.cfg"><i class="icon-doc"></i><span>setup.cfg</span></a>' % HG_REPO)
         response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/setup.py"><i class="icon-doc"></i><span>setup.py</span></a>' % HG_REPO)
         response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/test_and_report.sh"><i class="icon-doc"></i><span>test_and_report.sh</span></a>' % HG_REPO)
         response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/tox.ini"><i class="icon-doc"></i><span>tox.ini</span></a>' % HG_REPO)
     def test_index_revision(self):
         self.log_user()
         response = self.app.get(
             url(controller='files', action='index',
                 repo_name=HG_REPO,
                 revision='7ba66bec8d6dbba14a2155be32408c435c5f4492',
                 f_path='/')
+        )
         #Test response...
         response.mustcontain('<a class="browser-dir ypjax-link" href="/%s/files/7ba66bec8d6dbba14a2155be32408c435c5f4492/docs"><i class="icon-folder-open"></i><span>docs</span></a>' % HG_REPO)
         response.mustcontain('<a class="browser-dir ypjax-link" href="/%s/files/7ba66bec8d6dbba14a2155be32408c435c5f4492/tests"><i class="icon-folder-open"></i><span>tests</span></a>' % HG_REPO)
         response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/7ba66bec8d6dbba14a2155be32408c435c5f4492/README.rst"><i class="icon-doc"></i><span>README.rst</span></a>' % HG_REPO)
         response.mustcontain('1.1 KiB')
     def test_index_different_branch(self):
         self.log_user()
         response = self.app.get(url(controller='files', action='index',
                                     repo_name=HG_REPO,
                                     revision='97e8b885c04894463c51898e14387d80c30ed1ee',
                                     f_path='/'))
         response.mustcontain("""<option selected="selected" value="97e8b885c04894463c51898e14387d80c30ed1ee">git at 97e8b885c048</option>""")
     def test_index_paging(self):
         self.log_user()
         for r in [(73, 'a066b25d5df7016b45a41b7e2a78c33b57adc235'),
                   (92, 'cc66b61b8455b264a7a8a2d8ddc80fcfc58c221e'),
                   (109, '75feb4c33e81186c87eac740cee2447330288412'),
                   (1, '3d8f361e72ab303da48d799ff1ac40d5ac37c67e'),
                   (0, 'b986218ba1c9b0d6a259fac9b050b1724ed8e545')]:
             response = self.app.get(url(controller='files', action='index',
                                     repo_name=HG_REPO,
                                     revision=r[1],
                                     f_path='/'))
             response.mustcontain("""@ r%s:%s""" % (r[0], r[1][:12]))
     def test_file_source(self):
         self.log_user()
         response = self.app.get(url(controller='files', action='index',
                                     repo_name=HG_REPO,
                                     revision='8911406ad776fdd3d0b9932a2e89677e57405a48',
                                     f_path='vcs/nodes.py'))
         response.mustcontain("""<div class="commit">Partially implemented <a class="issue-tracker-link" href="https://issues.example.com/vcs_test_hg/issue/16">#16</a>. filecontent/commit message/author/node name are safe_unicode now.
 In addition some other __str__ are unicode as well
 Added test for unicode
 Improved test to clone into uniq repository.
 removed extra unicode conversion in diff.</div>
 """)
         response.mustcontain("""<option selected="selected" value="8911406ad776fdd3d0b9932a2e89677e57405a48">default at 8911406ad776</option>""")
     def test_file_source_history(self):
         self.log_user()
         response = self.app.get(url(controller='files', action='history',
                                     repo_name=HG_REPO,
                                     revision='tip',
                                     f_path='vcs/nodes.py'),
                                 extra_environ={'HTTP_X_PARTIAL_XHR': '1'},)
-        self.assertEqual(response.body, HG_NODE_HISTORY)
+        assert response.body == HG_NODE_HISTORY
     def test_file_source_history_git(self):
         self.log_user()
         response = self.app.get(url(controller='files', action='history',
                                     repo_name=GIT_REPO,
                                     revision='master',
                                     f_path='vcs/nodes.py'),
                                 extra_environ={'HTTP_X_PARTIAL_XHR': '1'},)
-        self.assertEqual(response.body, GIT_NODE_HISTORY)
+        assert response.body == GIT_NODE_HISTORY
     def test_file_annotation(self):
         self.log_user()
         response = self.app.get(url(controller='files', action='index',
                                     repo_name=HG_REPO,
                                     revision='tip',
                                     f_path='vcs/nodes.py',
                                     annotate=True))
         response.mustcontain("""r356:25213a5fbb04""")
     def test_file_annotation_git(self):
         self.log_user()
         response = self.app.get(url(controller='files', action='index',
                                     repo_name=GIT_REPO,
                                     revision='master',
                                     f_path='vcs/nodes.py',
                                     annotate=True))
         response.mustcontain("""r345:c994f0de03b2""")
     def test_file_annotation_history(self):
         self.log_user()
         response = self.app.get(url(controller='files', action='history',
                                     repo_name=HG_REPO,
                                     revision='tip',
                                     f_path='vcs/nodes.py',
                                     annotate=True),
                                 extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
-        self.assertEqual(response.body, HG_NODE_HISTORY)
+        assert response.body == HG_NODE_HISTORY
     def test_file_annotation_history_git(self):
         self.log_user()
         response = self.app.get(url(controller='files', action='history',
                                     repo_name=GIT_REPO,
                                     revision='master',
                                     f_path='vcs/nodes.py',
                                     annotate=True),
                                 extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
-        self.assertEqual(response.body, GIT_NODE_HISTORY)
+        assert response.body == GIT_NODE_HISTORY
     def test_file_authors(self):
         self.log_user()
         response = self.app.get(url(controller='files', action='authors',
                                     repo_name=HG_REPO,
                                     revision='tip',
                                     f_path='vcs/nodes.py',
                                     annotate=True))
         response.mustcontain('Marcin Kuzminski')
         response.mustcontain('Lukasz Balcerzak')
     def test_file_authors_git(self):
         self.log_user()
         response = self.app.get(url(controller='files', action='authors',
                                     repo_name=GIT_REPO,
                                     revision='master',
                                     f_path='vcs/nodes.py',
                                     annotate=True))
         response.mustcontain('Marcin Kuzminski')
         response.mustcontain('Lukasz Balcerzak')
     def test_archival(self):
         self.log_user()
         _set_downloads(HG_REPO, set_to=True)
         for arch_ext, info in ARCHIVE_SPECS.items():
             short = '27cd5cce30c9%s' % arch_ext
             fname = '27cd5cce30c96924232dffcd24178a07ffeb5dfc%s' % arch_ext
             filename = '%s-%s' % (HG_REPO, short)
             response = self.app.get(url(controller='files',
                                         action='archivefile',
                                         repo_name=HG_REPO,
                                         fname=fname))
-            self.assertEqual(response.status, '200 OK')
+            assert response.status == '200 OK'
             heads = [
                 ('Pragma', 'no-cache'),
                 ('Cache-Control', 'no-cache'),
                 ('Content-Disposition', 'attachment; filename=%s' % filename),
                 ('Content-Type', '%s; charset=utf-8' % info[0]),
+            ]
-            self.assertEqual(response.response._headers.items(), heads)
+            assert response.response._headers.items() == heads
     def test_archival_wrong_ext(self):
         self.log_user()
         _set_downloads(HG_REPO, set_to=True)
         for arch_ext in ['tar', 'rar', 'x', '..ax', '.zipz']:
             fname = '27cd5cce30c96924232dffcd24178a07ffeb5dfc%s' % arch_ext
             response = self.app.get(url(controller='files',
                                         action='archivefile',
                                         repo_name=HG_REPO,
                                         fname=fname))
             response.mustcontain('Unknown archive type')
     def test_archival_wrong_revision(self):
         self.log_user()
         _set_downloads(HG_REPO, set_to=True)
         for rev in ['00x000000', 'tar', 'wrong', '@##$@$42413232', '232dffcd']:
             fname = '%s.zip' % rev
             response = self.app.get(url(controller='files',
                                         action='archivefile',
                                         repo_name=HG_REPO,
                                         fname=fname))
             response.mustcontain('Unknown revision')
     #==========================================================================
     # RAW FILE
     #==========================================================================
     def test_raw_file_ok(self):
         self.log_user()
         response = self.app.get(url(controller='files', action='rawfile',
                                     repo_name=HG_REPO,
                                     revision='27cd5cce30c96924232dffcd24178a07ffeb5dfc',
                                     f_path='vcs/nodes.py'))
         self.assertEqual(response.content_disposition, "attachment; filename=nodes.py")
         self.assertEqual(response.content_type, mimetypes.guess_type("nodes.py")[0])
         assert response.content_disposition == "attachment; filename=nodes.py"
         assert response.content_type == mimetypes.guess_type("nodes.py")[0]
     def test_raw_file_wrong_cs(self):
         self.log_user()
         rev = u'ERRORce30c96924232dffcd24178a07ffeb5dfc'
         f_path = 'vcs/nodes.py'
         response = self.app.get(url(controller='files', action='rawfile',
                                     repo_name=HG_REPO,
                                     revision=rev,
                                     f_path=f_path), status=404)
         msg = """Such revision does not exist for this repository"""
         response.mustcontain(msg)
     def test_raw_file_wrong_f_path(self):
         self.log_user()
         rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
         f_path = 'vcs/ERRORnodes.py'
         response = self.app.get(url(controller='files', action='rawfile',
                                     repo_name=HG_REPO,
                                     revision=rev,
                                     f_path=f_path), status=404)
         msg = "There is no file nor directory at the given path: &#39;%s&#39; at revision %s" % (f_path, rev[:12])
         response.mustcontain(msg)
     #==========================================================================
     # RAW RESPONSE - PLAIN
     #==========================================================================
     def test_raw_ok(self):
         self.log_user()
         response = self.app.get(url(controller='files', action='raw',
                                     repo_name=HG_REPO,
                                     revision='27cd5cce30c96924232dffcd24178a07ffeb5dfc',
                                     f_path='vcs/nodes.py'))
-        self.assertEqual(response.content_type, "text/plain")
+        assert response.content_type == "text/plain"
     def test_raw_wrong_cs(self):
         self.log_user()
         rev = u'ERRORcce30c96924232dffcd24178a07ffeb5dfc'
         f_path = 'vcs/nodes.py'
         response = self.app.get(url(controller='files', action='raw',
                                     repo_name=HG_REPO,
                                     revision=rev,
                                     f_path=f_path), status=404)
         msg = """Such revision does not exist for this repository"""
         response.mustcontain(msg)
     def test_raw_wrong_f_path(self):
         self.log_user()
         rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
         f_path = 'vcs/ERRORnodes.py'
         response = self.app.get(url(controller='files', action='raw',
                                     repo_name=HG_REPO,
                                     revision=rev,
                                     f_path=f_path), status=404)
         msg = "There is no file nor directory at the given path: &#39;%s&#39; at revision %s" % (f_path, rev[:12])
         response.mustcontain(msg)
     def test_ajaxed_files_list(self):
         self.log_user()
         rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
         response = self.app.get(
             url('files_nodelist_home', repo_name=HG_REPO, f_path='/',
                 revision=rev),
             extra_environ={'HTTP_X_PARTIAL_XHR': '1'},
+        )
         response.mustcontain("vcs/web/simplevcs/views/repository.py")
     # Hg - ADD FILE
     def test_add_file_view_hg(self):
         self.log_user()
         response = self.app.get(url('files_add_home',
                                       repo_name=HG_REPO,
                                       revision='tip', f_path='/'))
     def test_add_file_into_hg_missing_content(self):
         self.log_user()
         response = self.app.post(url('files_add_home',
                                       repo_name=HG_REPO,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': '',
                                     '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         self.checkSessionFlash(response, 'No content')
     def test_add_file_into_hg_missing_filename(self):
         self.log_user()
         response = self.app.post(url('files_add_home',
                                       repo_name=HG_REPO,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "foo",
                                     '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         self.checkSessionFlash(response, 'No filename')
     @parametrize('location,filename', [
         ('/abs', 'foo'),
         ('../rel', 'foo'),
         ('file/../foo', 'foo'),
     ])
     def test_add_file_into_hg_bad_filenames(self, location, filename):
         self.log_user()
         response = self.app.post(url('files_add_home',
                                       repo_name=HG_REPO,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "foo",
                                     'filename': filename,
                                     'location': location,
                                     '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         self.checkSessionFlash(response, 'Location must be relative path and must not contain .. in path')
     @parametrize('cnt,location,filename', [
         (1, '', 'foo.txt'),
         (2, 'dir', 'foo.rst'),
         (3, 'rel/dir', 'foo.bar'),
     ])
     def test_add_file_into_hg(self, cnt, location, filename):
         self.log_user()
         repo = fixture.create_repo(u'commit-test-%s' % cnt, repo_type='hg')
         response = self.app.post(url('files_add_home',
                                       repo_name=repo.repo_name,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "foo",
                                     'filename': filename,
                                     'location': location,
                                     '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
         finally:
             fixture.destroy_repo(repo.repo_name)
     # Git - add file
     def test_add_file_view_git(self):
         self.log_user()
         response = self.app.get(url('files_add_home',
                                       repo_name=GIT_REPO,
                                       revision='tip', f_path='/'))
     def test_add_file_into_git_missing_content(self):
         self.log_user()
         response = self.app.post(url('files_add_home',
                                       repo_name=GIT_REPO,
                                       revision='tip', f_path='/'),
                                  params={
                                      'content': '',
                                      '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         self.checkSessionFlash(response, 'No content')
     def test_add_file_into_git_missing_filename(self):
         self.log_user()
         response = self.app.post(url('files_add_home',
                                       repo_name=GIT_REPO,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "foo",
                                     '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         self.checkSessionFlash(response, 'No filename')
     @parametrize('location,filename', [
         ('/abs', 'foo'),
         ('../rel', 'foo'),
         ('file/../foo', 'foo'),
     ])
     def test_add_file_into_git_bad_filenames(self, location, filename):
         self.log_user()
         response = self.app.post(url('files_add_home',
                                       repo_name=GIT_REPO,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "foo",
                                     'filename': filename,
                                     'location': location,
                                     '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         self.checkSessionFlash(response, 'Location must be relative path and must not contain .. in path')
     @parametrize('cnt,location,filename', [
         (1, '', 'foo.txt'),
         (2, 'dir', 'foo.rst'),
         (3, 'rel/dir', 'foo.bar'),
     ])
     def test_add_file_into_git(self, cnt, location, filename):
         self.log_user()
         repo = fixture.create_repo(u'commit-test-%s' % cnt, repo_type='git')
         response = self.app.post(url('files_add_home',
                                       repo_name=repo.repo_name,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "foo",
                                     'filename': filename,
                                     'location': location,
                                     '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
         finally:
             fixture.destroy_repo(repo.repo_name)
     # Hg - EDIT
     def test_edit_file_view_hg(self):
         self.log_user()
         response = self.app.get(url('files_edit_home',
                                       repo_name=HG_REPO,
                                       revision='tip', f_path='vcs/nodes.py'))
     def test_edit_file_view_not_on_branch_hg(self):
         self.log_user()
         repo = fixture.create_repo(u'test-edit-repo', repo_type='hg')
         ## add file
         location = 'vcs'
         filename = 'nodes.py'
         response = self.app.post(url('files_add_home',
                                       repo_name=repo.repo_name,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
                                     '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
             response = self.app.get(url('files_edit_home',
                                           repo_name=repo.repo_name,
                                           revision='tip', f_path=posixpath.join(location, filename)),
                                     status=302)
             self.checkSessionFlash(response,
                 'You can only edit files with revision being a valid branch')
         finally:
             fixture.destroy_repo(repo.repo_name)
     def test_edit_file_view_commit_changes_hg(self):
         self.log_user()
         repo = fixture.create_repo(u'test-edit-repo', repo_type='hg')
         ## add file
         location = 'vcs'
         filename = 'nodes.py'
         response = self.app.post(url('files_add_home',
                                       repo_name=repo.repo_name,
                                       revision='tip',
                                       f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
                                     '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
             response = self.app.post(url('files_edit_home',
                                           repo_name=repo.repo_name,
                                           revision=repo.scm_instance.DEFAULT_BRANCH_NAME,
                                           f_path=posixpath.join(location, filename)),
                                      params={
                                         'content': "def py():\n print 'hello world'\n",
                                         'message': 'i committed',
                                         '_authentication_token': self.authentication_token(),
                                      },
                                     status=302)
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
         finally:
             fixture.destroy_repo(repo.repo_name)
     # Git - edit
     def test_edit_file_view_git(self):
         self.log_user()
         response = self.app.get(url('files_edit_home',
                                       repo_name=GIT_REPO,
                                       revision='tip', f_path='vcs/nodes.py'))
     def test_edit_file_view_not_on_branch_git(self):
         self.log_user()
         repo = fixture.create_repo(u'test-edit-repo', repo_type='git')
         ## add file
         location = 'vcs'
         filename = 'nodes.py'
         response = self.app.post(url('files_add_home',
                                       repo_name=repo.repo_name,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
                                     '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
             response = self.app.get(url('files_edit_home',
                                           repo_name=repo.repo_name,
                                           revision='tip', f_path=posixpath.join(location, filename)),
                                     status=302)
             self.checkSessionFlash(response,
                 'You can only edit files with revision being a valid branch')
         finally:
             fixture.destroy_repo(repo.repo_name)
     def test_edit_file_view_commit_changes_git(self):
         self.log_user()
         repo = fixture.create_repo(u'test-edit-repo', repo_type='git')
         ## add file
         location = 'vcs'
         filename = 'nodes.py'
         response = self.app.post(url('files_add_home',
                                       repo_name=repo.repo_name,
                                       revision='tip',
                                       f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
                                     '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
             response = self.app.post(url('files_edit_home',
                                           repo_name=repo.repo_name,
                                           revision=repo.scm_instance.DEFAULT_BRANCH_NAME,
                                           f_path=posixpath.join(location, filename)),
                                      params={
                                         'content': "def py():\n print 'hello world'\n",
                                         'message': 'i committed',
                                         '_authentication_token': self.authentication_token(),
                                      },
                                     status=302)
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
         finally:
             fixture.destroy_repo(repo.repo_name)
     # Hg - delete
     def test_delete_file_view_hg(self):
         self.log_user()
         response = self.app.get(url('files_delete_home',
                                      repo_name=HG_REPO,
                                      revision='tip', f_path='vcs/nodes.py'))
     def test_delete_file_view_not_on_branch_hg(self):
         self.log_user()
         repo = fixture.create_repo(u'test-delete-repo', repo_type='hg')
         ## add file
         location = 'vcs'
         filename = 'nodes.py'
         response = self.app.post(url('files_add_home',
                                       repo_name=repo.repo_name,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
                                     '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
             response = self.app.get(url('files_delete_home',
                                           repo_name=repo.repo_name,
                                           revision='tip', f_path=posixpath.join(location, filename)),
                                     status=302)
             self.checkSessionFlash(response,
                 'You can only delete files with revision being a valid branch')
         finally:
             fixture.destroy_repo(repo.repo_name)
     def test_delete_file_view_commit_changes_hg(self):
         self.log_user()
         repo = fixture.create_repo(u'test-delete-repo', repo_type='hg')
         ## add file
         location = 'vcs'
         filename = 'nodes.py'
         response = self.app.post(url('files_add_home',
                                       repo_name=repo.repo_name,
                                       revision='tip',
                                       f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
                                     '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
             response = self.app.post(url('files_delete_home',
                                           repo_name=repo.repo_name,
                                           revision=repo.scm_instance.DEFAULT_BRANCH_NAME,
                                           f_path=posixpath.join(location, filename)),
                                      params={
                                         'message': 'i committed',
                                         '_authentication_token': self.authentication_token(),
                                      },
                                     status=302)
             self.checkSessionFlash(response,
                                    'Successfully deleted file %s' % posixpath.join(location, filename))
         finally:
             fixture.destroy_repo(repo.repo_name)
     # Git - delete
     def test_delete_file_view_git(self):
         self.log_user()
         response = self.app.get(url('files_delete_home',
                                      repo_name=HG_REPO,
                                      revision='tip', f_path='vcs/nodes.py'))
     def test_delete_file_view_not_on_branch_git(self):
         self.log_user()
         repo = fixture.create_repo(u'test-delete-repo', repo_type='git')
         ## add file
         location = 'vcs'
         filename = 'nodes.py'
         response = self.app.post(url('files_add_home',
                                       repo_name=repo.repo_name,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
                                     '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
             response = self.app.get(url('files_delete_home',
                                           repo_name=repo.repo_name,
                                           revision='tip', f_path=posixpath.join(location, filename)),
                                     status=302)
             self.checkSessionFlash(response,
                 'You can only delete files with revision being a valid branch')
         finally:
             fixture.destroy_repo(repo.repo_name)
     def test_delete_file_view_commit_changes_git(self):
         self.log_user()
         repo = fixture.create_repo(u'test-delete-repo', repo_type='git')
         ## add file
         location = 'vcs'
         filename = 'nodes.py'
         response = self.app.post(url('files_add_home',
                                       repo_name=repo.repo_name,
                                       revision='tip',
                                       f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
                                     '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
             response = self.app.post(url('files_delete_home',
                                           repo_name=repo.repo_name,
                                           revision=repo.scm_instance.DEFAULT_BRANCH_NAME,
                                           f_path=posixpath.join(location, filename)),
                                      params={
                                         'message': 'i committed',
                                         '_authentication_token': self.authentication_token(),
                                      },
                                     status=302)
             self.checkSessionFlash(response,
                                    'Successfully deleted file %s' % posixpath.join(location, filename))
         finally:
             fixture.destroy_repo(repo.repo_name)

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()
 class _BaseTestCase(TestControllerPytest):
     """
     Write all tests here
     """
     REPO = None
     REPO_TYPE = None
     NEW_REPO = None
     REPO_FORK = None
     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_method(self, method):
         Session().delete(self.u1)
         Session().commit()
     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)
         assert fork_repo.repo_name == fork_name_full
         assert 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)
         assert fork_repo.repo_name == fork_name
         assert 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))
+        assert 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))
+        assert 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(_BaseTestCase):
     REPO = GIT_REPO
     NEW_REPO = NEW_GIT_REPO
     REPO_TYPE = 'git'
     REPO_FORK = GIT_FORK
 class TestHG(_BaseTestCase):
     REPO = HG_REPO
     NEW_REPO = NEW_HG_REPO
     REPO_TYPE = 'hg'
     REPO_FORK = HG_FORK

kallithea/tests/functional/test_login.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 import re
 import time
 import urlparse
 import mock
 from kallithea.tests import *
 from kallithea.tests.fixture import Fixture
 from kallithea.lib.utils2 import generate_api_key
 from kallithea.lib.auth import check_password
 from kallithea.lib import helpers as h
 from kallithea.model.api_key import ApiKeyModel
 from kallithea.model import validators
 from kallithea.model.db import User, Notification
 from kallithea.model.meta import Session
 from kallithea.model.user import UserModel
 fixture = Fixture()
 class TestLoginController(TestControllerPytest):
     def setup_method(self, method):
         remove_all_notifications()
-        self.assertEqual(Notification.query().all(), [])
+        assert Notification.query().all() == []
     def test_index(self):
         response = self.app.get(url(controller='login', action='index'))
-        self.assertEqual(response.status, '200 OK')
+        assert response.status == '200 OK'
         # Test response...
     def test_login_admin_ok(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_ADMIN_LOGIN,
                                   'password': TEST_USER_ADMIN_PASS})
-        self.assertEqual(response.status, '302 Found')
+        assert response.status == '302 Found'
         self.assert_authenticated_user(response, TEST_USER_ADMIN_LOGIN)
         response = response.follow()
         response.mustcontain('/%s' % HG_REPO)
     def test_login_regular_ok(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_REGULAR_LOGIN,
                                   'password': TEST_USER_REGULAR_PASS})
-        self.assertEqual(response.status, '302 Found')
+        assert response.status == '302 Found'
         self.assert_authenticated_user(response, TEST_USER_REGULAR_LOGIN)
         response = response.follow()
         response.mustcontain('/%s' % HG_REPO)
     def test_login_regular_email_ok(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_REGULAR_EMAIL,
                                   'password': TEST_USER_REGULAR_PASS})
-        self.assertEqual(response.status, '302 Found')
+        assert response.status == '302 Found'
         self.assert_authenticated_user(response, TEST_USER_REGULAR_LOGIN)
         response = response.follow()
         response.mustcontain('/%s' % HG_REPO)
     def test_login_ok_came_from(self):
         test_came_from = '/_admin/users'
         response = self.app.post(url(controller='login', action='index',
                                      came_from=test_came_from),
                                  {'username': TEST_USER_ADMIN_LOGIN,
                                   'password': TEST_USER_ADMIN_PASS})
-        self.assertEqual(response.status, '302 Found')
+        assert response.status == '302 Found'
         response = response.follow()
-        self.assertEqual(response.status, '200 OK')
+        assert response.status == '200 OK'
         response.mustcontain('Users Administration')
     def test_login_do_not_remember(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_REGULAR_LOGIN,
                                   'password': TEST_USER_REGULAR_PASS,
                                   'remember': False})
-        self.assertIn('Set-Cookie', response.headers)
+        assert 'Set-Cookie' in response.headers
         for cookie in response.headers.getall('Set-Cookie'):
             self.assertFalse(re.search(r';\s+(Max-Age|Expires)=', cookie, re.IGNORECASE),
                 'Cookie %r has expiration date, but should be a session cookie' % cookie)
             assert not re.search(r';\s+(Max-Age|Expires)=', cookie, re.IGNORECASE), 'Cookie %r has expiration date, but should be a session cookie' % cookie
     def test_login_remember(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_REGULAR_LOGIN,
                                   'password': TEST_USER_REGULAR_PASS,
                                   'remember': True})
-        self.assertIn('Set-Cookie', response.headers)
+        assert 'Set-Cookie' in response.headers
         for cookie in response.headers.getall('Set-Cookie'):
             self.assertTrue(re.search(r';\s+(Max-Age|Expires)=', cookie, re.IGNORECASE),
                 'Cookie %r should have expiration date, but is a session cookie' % cookie)
             assert re.search(r';\s+(Max-Age|Expires)=', cookie, re.IGNORECASE), 'Cookie %r should have expiration date, but is a session cookie' % cookie
     def test_logout(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_REGULAR_LOGIN,
                                   'password': TEST_USER_REGULAR_PASS})
         # Verify that a login session has been established.
         response = self.app.get(url(controller='login', action='index'))
         response = response.follow()
-        self.assertIn('authuser', response.session)
+        assert 'authuser' in response.session
         response.click('Log Out')
         # Verify that the login session has been terminated.
         response = self.app.get(url(controller='login', action='index'))
-        self.assertNotIn('authuser', response.session)
+        assert 'authuser' not in response.session
     @parametrize('url_came_from', [
           ('data:text/html,<script>window.alert("xss")</script>',),
           ('mailto:test@example.com',),
           ('file:///etc/passwd',),
           ('ftp://ftp.example.com',),
           ('http://other.example.com/bl%C3%A5b%C3%A6rgr%C3%B8d',),
           ('//evil.example.com/',),
           ('/\r\nX-Header-Injection: boo',),
           ('/invälid_url_bytes',),
           ('non-absolute-path',),
     ])
     def test_login_bad_came_froms(self, url_came_from):
         response = self.app.post(url(controller='login', action='index',
                                      came_from=url_came_from),
                                  {'username': TEST_USER_ADMIN_LOGIN,
                                   'password': TEST_USER_ADMIN_PASS},
                                  status=400)
     def test_login_short_password(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_ADMIN_LOGIN,
                                   'password': 'as'})
-        self.assertEqual(response.status, '200 OK')
+        assert response.status == '200 OK'
         response.mustcontain('Enter 3 characters or more')
     def test_login_wrong_username_password(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': 'error',
                                   'password': 'test12'})
         response.mustcontain('Invalid username or password')
     # verify that get arguments are correctly passed along login redirection
     @parametrize('args,args_encoded', [
         ({'foo':'one', 'bar':'two'}, (('foo', 'one'), ('bar', 'two'))),
         ({'blue': u'blå'.encode('utf-8'), 'green':u'grøn'},
              (('blue', u'blå'.encode('utf-8')), ('green', u'grøn'.encode('utf-8')))),
     ])
     def test_redirection_to_login_form_preserves_get_args(self, args, args_encoded):
         with fixture.anon_access(False):
             response = self.app.get(url(controller='summary', action='index',
                                         repo_name=HG_REPO,
                                         **args))
-            self.assertEqual(response.status, '302 Found')
+            assert response.status == '302 Found'
             came_from = urlparse.parse_qs(urlparse.urlparse(response.location).query)['came_from'][0]
             came_from_qs = urlparse.parse_qsl(urlparse.urlparse(came_from).query)
             for encoded in args_encoded:
-                self.assertIn(encoded, came_from_qs)
+                assert encoded in came_from_qs
     @parametrize('args,args_encoded', [
         ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),
         ({'blue': u'blå', 'green':u'grøn'},
              ('blue=bl%C3%A5', 'green=gr%C3%B8n')),
     ])
     def test_login_form_preserves_get_args(self, args, args_encoded):
         response = self.app.get(url(controller='login', action='index',
                                     came_from=url('/_admin/users', **args)))
         came_from = urlparse.parse_qs(urlparse.urlparse(response.form.action).query)['came_from'][0]
         for encoded in args_encoded:
-            self.assertIn(encoded, came_from)
+            assert encoded in came_from
     @parametrize('args,args_encoded', [
         ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),
         ({'blue': u'blå', 'green':u'grøn'},
              ('blue=bl%C3%A5', 'green=gr%C3%B8n')),
     ])
     def test_redirection_after_successful_login_preserves_get_args(self, args, args_encoded):
         response = self.app.post(url(controller='login', action='index',
                                      came_from = url('/_admin/users', **args)),
                                  {'username': TEST_USER_ADMIN_LOGIN,
                                   'password': TEST_USER_ADMIN_PASS})
-        self.assertEqual(response.status, '302 Found')
+        assert response.status == '302 Found'
         for encoded in args_encoded:
-            self.assertIn(encoded, response.location)
+            assert encoded in response.location
     @parametrize('args,args_encoded', [
         ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),
         ({'blue': u'blå', 'green':u'grøn'},
              ('blue=bl%C3%A5', 'green=gr%C3%B8n')),
     ])
     def test_login_form_after_incorrect_login_preserves_get_args(self, args, args_encoded):
         response = self.app.post(url(controller='login', action='index',
                                      came_from=url('/_admin/users', **args)),
                                  {'username': 'error',
                                   'password': 'test12'})
         response.mustcontain('Invalid username or password')
         came_from = urlparse.parse_qs(urlparse.urlparse(response.form.action).query)['came_from'][0]
         for encoded in args_encoded:
-            self.assertIn(encoded, came_from)
+            assert encoded in came_from
     #==========================================================================
     # REGISTRATIONS
     #==========================================================================
     def test_register(self):
         response = self.app.get(url(controller='login', action='register'))
         response.mustcontain('Sign Up')
     def test_register_err_same_username(self):
         uname = TEST_USER_ADMIN_LOGIN
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': uname,
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': 'goodmail@example.com',
                                              'firstname': 'test',
                                              'lastname': 'test'})
         msg = validators.ValidUsername()._messages['username_exists']
         msg = h.html_escape(msg % {'username': uname})
         response.mustcontain(msg)
     def test_register_err_same_email(self):
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': 'test_admin_0',
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': TEST_USER_ADMIN_EMAIL,
                                              'firstname': 'test',
                                              'lastname': 'test'})
         msg = validators.UniqSystemEmail()()._messages['email_taken']
         response.mustcontain(msg)
     def test_register_err_same_email_case_sensitive(self):
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': 'test_admin_1',
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': TEST_USER_ADMIN_EMAIL.title(),
                                              'firstname': 'test',
                                              'lastname': 'test'})
         msg = validators.UniqSystemEmail()()._messages['email_taken']
         response.mustcontain(msg)
     def test_register_err_wrong_data(self):
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': 'xs',
                                              'password': 'test',
                                              'password_confirmation': 'test',
                                              'email': 'goodmailm',
                                              'firstname': 'test',
                                              'lastname': 'test'})
-        self.assertEqual(response.status, '200 OK')
+        assert response.status == '200 OK'
         response.mustcontain('An email address must contain a single @')
         response.mustcontain('Enter a value 6 characters long or more')
     def test_register_err_username(self):
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': 'error user',
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': 'goodmailm',
                                              'firstname': 'test',
                                              'lastname': 'test'})
         response.mustcontain('An email address must contain a single @')
         response.mustcontain('Username may only contain '
                 'alphanumeric characters underscores, '
                 'periods or dashes and must begin with an '
                 'alphanumeric character')
     def test_register_err_case_sensitive(self):
         usr = TEST_USER_ADMIN_LOGIN.title()
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': usr,
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': 'goodmailm',
                                              'firstname': 'test',
                                              'lastname': 'test'})
         response.mustcontain('An email address must contain a single @')
         msg = validators.ValidUsername()._messages['username_exists']
         msg = h.html_escape(msg % {'username': usr})
         response.mustcontain(msg)
     def test_register_special_chars(self):
         response = self.app.post(url(controller='login', action='register'),
                                         {'username': 'xxxaxn',
                                          'password': 'ąćźżąśśśś',
                                          'password_confirmation': 'ąćźżąśśśś',
                                          'email': 'goodmailm@test.plx',
                                          'firstname': 'test',
                                          'lastname': 'test'})
         msg = validators.ValidPassword()._messages['invalid_password']
         response.mustcontain(msg)
     def test_register_password_mismatch(self):
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': 'xs',
                                              'password': '123qwe',
                                              'password_confirmation': 'qwe123',
                                              'email': 'goodmailm@test.plxa',
                                              'firstname': 'test',
                                              'lastname': 'test'})
         msg = validators.ValidPasswordsMatch('password', 'password_confirmation')._messages['password_mismatch']
         response.mustcontain(msg)
     def test_register_ok(self):
         username = 'test_regular4'
         password = 'qweqwe'
         email = 'user4@example.com'
         name = 'testname'
         lastname = 'testlastname'
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': username,
                                              'password': password,
                                              'password_confirmation': password,
                                              'email': email,
                                              'firstname': name,
                                              'lastname': lastname,
                                              'admin': True})  # This should be overridden
-        self.assertEqual(response.status, '302 Found')
+        assert response.status == '302 Found'
         self.checkSessionFlash(response, 'You have successfully registered into Kallithea')
         ret = Session().query(User).filter(User.username == 'test_regular4').one()
         self.assertEqual(ret.username, username)
         self.assertEqual(check_password(password, ret.password), True)
         self.assertEqual(ret.email, email)
         self.assertEqual(ret.name, name)
         self.assertEqual(ret.lastname, lastname)
         self.assertNotEqual(ret.api_key, None)
         self.assertEqual(ret.admin, False)
         assert ret.username == username
         assert check_password(password, ret.password) == True
         assert ret.email == email
         assert ret.name == name
         assert ret.lastname == lastname
         assert ret.api_key != None
         assert ret.admin == False
     #==========================================================================
     # PASSWORD RESET
     #==========================================================================
     def test_forgot_password_wrong_mail(self):
         bad_email = 'username%wrongmail.org'
         response = self.app.post(
                         url(controller='login', action='password_reset'),
                             {'email': bad_email, }
+        )
         response.mustcontain('An email address must contain a single @')
     def test_forgot_password(self):
         response = self.app.get(url(controller='login',
                                     action='password_reset'))
-        self.assertEqual(response.status, '200 OK')
+        assert response.status == '200 OK'
         username = 'test_password_reset_1'
         password = 'qweqwe'
         email = 'username@example.com'
         name = u'passwd'
         lastname = u'reset'
         timestamp = int(time.time())
         new = User()
         new.username = username
         new.password = password
         new.email = email
         new.name = name
         new.lastname = lastname
         new.api_key = generate_api_key()
         Session().add(new)
         Session().commit()
         response = self.app.post(url(controller='login',
                                      action='password_reset'),
                                  {'email': email, })
         self.checkSessionFlash(response, 'A password reset confirmation code has been sent')
         response = response.follow()
         # BAD TOKEN
         token = "bad"
         response = self.app.post(url(controller='login',
                                      action='password_reset_confirmation'),
                                  {'email': email,
                                   'timestamp': timestamp,
                                   'password': "p@ssw0rd",
                                   'password_confirm': "p@ssw0rd",
                                   'token': token,
                                  })
-        self.assertEqual(response.status, '200 OK')
+        assert response.status == '200 OK'
         response.mustcontain('Invalid password reset token')
         # GOOD TOKEN
         # TODO: The token should ideally be taken from the mail sent
         # above, instead of being recalculated.
         token = UserModel().get_reset_password_token(
             User.get_by_username(username), timestamp, self.authentication_token())
         response = self.app.get(url(controller='login',
                                     action='password_reset_confirmation',
                                     email=email,
                                     timestamp=timestamp,
                                     token=token))
-        self.assertEqual(response.status, '200 OK')
+        assert response.status == '200 OK'
         response.mustcontain("You are about to set a new password for the email address %s" % email)
         response = self.app.post(url(controller='login',
                                      action='password_reset_confirmation'),
                                  {'email': email,
                                   'timestamp': timestamp,
                                   'password': "p@ssw0rd",
                                   'password_confirm': "p@ssw0rd",
                                   'token': token,
                                  })
-        self.assertEqual(response.status, '302 Found')
+        assert response.status == '302 Found'
         self.checkSessionFlash(response, 'Successfully updated password')
         response = response.follow()
     #==========================================================================
     # API
     #==========================================================================
     def _get_api_whitelist(self, values=None):
         config = {'api_access_controllers_whitelist': values or []}
         return config
     @parametrize('test_name,api_key', [
         ('none', None),
         ('empty_string', ''),
         ('fake_number', '123456'),
         ('proper_api_key', None)
     ])
     def test_access_not_whitelisted_page_via_api_key(self, test_name, api_key):
         whitelist = self._get_api_whitelist([])
         with mock.patch('kallithea.CONFIG', whitelist):
             self.assertEqual([],
                              whitelist['api_access_controllers_whitelist'])
             assert [] == whitelist['api_access_controllers_whitelist']
             if test_name == 'proper_api_key':
                 #use builtin if api_key is None
                 api_key = User.get_first_admin().api_key
             with fixture.anon_access(False):
                 self.app.get(url(controller='changeset',
                                  action='changeset_raw',
                                  repo_name=HG_REPO, revision='tip', api_key=api_key),
                              status=403)
     @parametrize('test_name,api_key,code', [
         ('none', None, 302),
         ('empty_string', '', 302),
         ('fake_number', '123456', 302),
         ('fake_not_alnum', 'a-z', 302),
         ('fake_api_key', '0123456789abcdef0123456789ABCDEF01234567', 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'])
             assert ['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'])
             assert ['ChangesetController:changeset_raw'] == whitelist['api_access_controllers_whitelist']
             new_api_key = ApiKeyModel().create(TEST_USER_ADMIN_LOGIN, u'test')
             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'])
             assert ['ChangesetController:changeset_raw'] == whitelist['api_access_controllers_whitelist']
             new_api_key = ApiKeyModel().create(TEST_USER_ADMIN_LOGIN, u'test')
             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)

kallithea/tests/functional/test_my_account.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 from kallithea.model.db import User, UserFollowing, Repository, UserApiKeys
 from kallithea.tests import *
 from kallithea.tests.fixture import Fixture
 from kallithea.lib import helpers as h
 from kallithea.model.user import UserModel
 from kallithea.model.meta import Session
 fixture = Fixture()
 class TestMyAccountController(TestControllerPytest):
     test_user_1 = 'testme'
     @classmethod
     def teardown_class(cls):
         if User.get_by_username(cls.test_user_1):
             UserModel().delete(cls.test_user_1)
             Session().commit()
     def test_my_account(self):
         self.log_user()
         response = self.app.get(url('my_account'))
         response.mustcontain('value="%s' % TEST_USER_ADMIN_LOGIN)
     def test_my_account_my_repos(self):
         self.log_user()
         response = self.app.get(url('my_account_repos'))
         cnt = Repository.query().filter(Repository.user ==
                            User.get_by_username(TEST_USER_ADMIN_LOGIN)).count()
         response.mustcontain('"totalRecords": %s' % cnt)
     def test_my_account_my_watched(self):
         self.log_user()
         response = self.app.get(url('my_account_watched'))
         cnt = UserFollowing.query().filter(UserFollowing.user ==
                             User.get_by_username(TEST_USER_ADMIN_LOGIN)).count()
         response.mustcontain('"totalRecords": %s' % cnt)
     def test_my_account_my_emails(self):
         self.log_user()
         response = self.app.get(url('my_account_emails'))
         response.mustcontain('No additional emails specified')
     def test_my_account_my_emails_add_existing_email(self):
         self.log_user()
         response = self.app.get(url('my_account_emails'))
         response.mustcontain('No additional emails specified')
         response = self.app.post(url('my_account_emails'),
                                  {'new_email': TEST_USER_REGULAR_EMAIL, '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'This email address is already in use')
     def test_my_account_my_emails_add_missing_email_in_form(self):
         self.log_user()
         response = self.app.get(url('my_account_emails'))
         response.mustcontain('No additional emails specified')
         response = self.app.post(url('my_account_emails'),
             {'_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'Please enter an email address')
     def test_my_account_my_emails_add_remove(self):
         self.log_user()
         response = self.app.get(url('my_account_emails'))
         response.mustcontain('No additional emails specified')
         response = self.app.post(url('my_account_emails'),
                                  {'new_email': 'barz@example.com', '_authentication_token': self.authentication_token()})
         response = self.app.get(url('my_account_emails'))
         from kallithea.model.db import UserEmailMap
         email_id = UserEmailMap.query() \
             .filter(UserEmailMap.user == User.get_by_username(TEST_USER_ADMIN_LOGIN)) \
             .filter(UserEmailMap.email == 'barz@example.com').one().email_id
         response.mustcontain('barz@example.com')
         response.mustcontain('<input id="del_email_id" name="del_email_id" type="hidden" value="%s" />' % email_id)
         response = self.app.post(url('my_account_emails'),
                                  {'del_email_id': email_id, '_method': 'delete', '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'Removed email from user')
         response = self.app.get(url('my_account_emails'))
         response.mustcontain('No additional emails specified')
     @parametrize('name,attrs',
         [('firstname', {'firstname': 'new_username'}),
          ('lastname', {'lastname': 'new_username'}),
          ('admin', {'admin': True}),
          ('admin', {'admin': False}),
          ('extern_type', {'extern_type': 'ldap'}),
          ('extern_type', {'extern_type': None}),
          #('extern_name', {'extern_name': 'test'}),
          #('extern_name', {'extern_name': None}),
          ('active', {'active': False}),
          ('active', {'active': True}),
          ('email', {'email': 'someemail@example.com'}),
         # ('new_password', {'new_password': 'foobar123',
         #                   'password_confirmation': 'foobar123'})
         ])
     def test_my_account_update(self, name, attrs):
         usr = fixture.create_user(self.test_user_1, password='qweqwe',
                                   email='testme@example.com',
                                   extern_type='internal',
                                   extern_name=self.test_user_1,
                                   skip_if_exists=True)
         params = usr.get_api_data(True)  # current user data
         user_id = usr.user_id
         self.log_user(username=self.test_user_1, password='qweqwe')
         params.update({'password_confirmation': ''})
         params.update({'new_password': ''})
         params.update({'extern_type': 'internal'})
         params.update({'extern_name': self.test_user_1})
         params.update({'_authentication_token': self.authentication_token()})
         params.update(attrs)
         response = self.app.post(url('my_account'), params)
         self.checkSessionFlash(response,
                                'Your account was updated successfully')
         updated_user = User.get_by_username(self.test_user_1)
         updated_params = updated_user.get_api_data(True)
         updated_params.update({'password_confirmation': ''})
         updated_params.update({'new_password': ''})
         params['last_login'] = updated_params['last_login']
         if name == 'email':
             params['emails'] = [attrs['email']]
         if name == 'extern_type':
             #cannot update this via form, expected value is original one
             params['extern_type'] = "internal"
         if name == 'extern_name':
             #cannot update this via form, expected value is original one
             params['extern_name'] = str(user_id)
         if name == 'active':
             #my account cannot deactivate account
             params['active'] = True
         if name == 'admin':
             #my account cannot make you an admin !
             params['admin'] = False
         params.pop('_authentication_token')
-        self.assertEqual(params, updated_params)
+        assert params == updated_params
     def test_my_account_update_err_email_exists(self):
         self.log_user()
         new_email = TEST_USER_REGULAR_EMAIL  # already existing email
         response = self.app.post(url('my_account'),
                                 params=dict(
                                     username=TEST_USER_ADMIN_LOGIN,
                                     new_password=TEST_USER_ADMIN_PASS,
                                     password_confirmation='test122',
                                     firstname=u'NewName',
                                     lastname=u'NewLastname',
                                     email=new_email,
                                     _authentication_token=self.authentication_token())
+                                )
         response.mustcontain('This email address is already in use')
     def test_my_account_update_err(self):
         self.log_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS)
         new_email = 'newmail.pl'
         response = self.app.post(url('my_account'),
                                  params=dict(
                                             username=TEST_USER_ADMIN_LOGIN,
                                             new_password=TEST_USER_ADMIN_PASS,
                                             password_confirmation='test122',
                                             firstname=u'NewName',
                                             lastname=u'NewLastname',
                                             email=new_email,
                                             _authentication_token=self.authentication_token()))
         response.mustcontain('An email address must contain a single @')
         from kallithea.model import validators
         msg = validators.ValidUsername(edit=False, old_data={}) \
                 ._messages['username_exists']
         msg = h.html_escape(msg % {'username': TEST_USER_ADMIN_LOGIN})
         response.mustcontain(msg)
     def test_my_account_api_keys(self):
         usr = self.log_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS)
         user = User.get(usr['user_id'])
         response = self.app.get(url('my_account_api_keys'))
         response.mustcontain(user.api_key)
         response.mustcontain('Expires: Never')
     @parametrize('desc,lifetime', [
         ('forever', -1),
         ('5mins', 60*5),
         ('30days', 60*60*24*30),
     ])
     def test_my_account_add_api_keys(self, desc, lifetime):
         usr = self.log_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS)
         user = User.get(usr['user_id'])
         response = self.app.post(url('my_account_api_keys'),
                                  {'description': desc, 'lifetime': lifetime, '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'API key successfully created')
         try:
             response = response.follow()
             user = User.get(usr['user_id'])
             for api_key in user.api_keys:
                 response.mustcontain(api_key)
         finally:
             for api_key in UserApiKeys.query().all():
                 Session().delete(api_key)
                 Session().commit()
     def test_my_account_remove_api_key(self):
         usr = self.log_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS)
         user = User.get(usr['user_id'])
         response = self.app.post(url('my_account_api_keys'),
                                  {'description': 'desc', 'lifetime': -1, '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'API key successfully created')
         response = response.follow()
         #now delete our key
         keys = UserApiKeys.query().all()
-        self.assertEqual(1, len(keys))
+        assert 1 == len(keys)
         response = self.app.post(url('my_account_api_keys'),
                  {'_method': 'delete', 'del_api_key': keys[0].api_key, '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'API key successfully deleted')
         keys = UserApiKeys.query().all()
-        self.assertEqual(0, len(keys))
+        assert 0 == len(keys)
     def test_my_account_reset_main_api_key(self):
         usr = self.log_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS)
         user = User.get(usr['user_id'])
         api_key = user.api_key
         response = self.app.get(url('my_account_api_keys'))
         response.mustcontain(api_key)
         response.mustcontain('Expires: Never')
         response = self.app.post(url('my_account_api_keys'),
                  {'_method': 'delete', 'del_api_key_builtin': api_key, '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'API key successfully reset')
         response = response.follow()
         response.mustcontain(no=[api_key])

Changeset was too big and was cut off... Show full diff anyway

0 comments (0 inline, 0 general)