kallithea Changeset - 0efca3ad8467

Changeset - 0efca3ad8467

Parent rev.

Child rev.

[Not reviewed]

default

0 15 0

Mads Kiilerich - 11 years ago 2015-04-07 03:30:05
madski@unity3d.com

tests: provide _authentication_token when POSTing

So far not used, just preparing for the the time when the actual checking is
introduced ...

This change is very verbose. self.app.post should perhaps just at this value
automagically ...

15 files changed with 170 insertions and 86 deletions:

kallithea/config/routing.py

kallithea/controllers/login.py

kallithea/tests/__init__.py

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_permissions.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_changeset_comments.py

kallithea/tests/functional/test_files.py

kallithea/tests/functional/test_forks.py

kallithea/tests/functional/test_my_account.py

0 comments (0 inline, 0 general)

kallithea/config/routing.py

➞

Show inline comments

@@ @@ -478,48 +478,49 @@ def make_map(config): @@
     rmap.connect('public_journal_atom',
                  '%s/public_journal/atom' % ADMIN_PREFIX, controller='journal',
                  action="public_journal_atom")
     rmap.connect('public_journal_atom_old',
                  '%s/public_journal_atom' % ADMIN_PREFIX, controller='journal',
                  action="public_journal_atom")
     rmap.connect('toggle_following', '%s/toggle_following' % ADMIN_PREFIX,
                  controller='journal', action='toggle_following',
                  conditions=dict(method=["POST"]))
     #SEARCH
     rmap.connect('search', '%s/search' % ADMIN_PREFIX, controller='search',)
     rmap.connect('search_repo_admin', '%s/search/{repo_name:.*}' % ADMIN_PREFIX,
                  controller='search',
                  conditions=dict(function=check_repo))
     rmap.connect('search_repo', '/{repo_name:.*?}/search',
                  controller='search',
                  conditions=dict(function=check_repo),
+                 )
     #LOGIN/LOGOUT/REGISTER/SIGN IN
     rmap.connect('authentication_token', '%s/authentication_token' % ADMIN_PREFIX, controller='login', action='authentication_token')
     rmap.connect('login_home', '%s/login' % ADMIN_PREFIX, controller='login')
     rmap.connect('logout_home', '%s/logout' % ADMIN_PREFIX, controller='login',
                  action='logout')
     rmap.connect('register', '%s/register' % ADMIN_PREFIX, controller='login',
                  action='register')
     rmap.connect('reset_password', '%s/password_reset' % ADMIN_PREFIX,
                  controller='login', action='password_reset')
     rmap.connect('reset_password_confirmation',
                  '%s/password_reset_confirmation' % ADMIN_PREFIX,
                  controller='login', action='password_reset_confirmation')
     #FEEDS
     rmap.connect('rss_feed_home', '/{repo_name:.*?}/feed/rss',
                 controller='feed', action='rss',
                 conditions=dict(function=check_repo))
     rmap.connect('atom_feed_home', '/{repo_name:.*?}/feed/atom',
                 controller='feed', action='atom',
                 conditions=dict(function=check_repo))
     #==========================================================================

kallithea/controllers/login.py

➞

Show inline comments

@@ @@ -249,24 +249,32 @@ class LoginController(BaseController): @@
                     encoding="UTF-8",
                     force_defaults=False)
         return render('/password_reset.html')
     def password_reset_confirmation(self):
         if request.GET and request.GET.get('key'):
             try:
                 user = User.get_by_api_key(request.GET.get('key'))
                 data = dict(email=user.email)
                 UserModel().reset_password(data)
                 h.flash(_('Your password reset was successful, '
                           'new password has been sent to your email'),
                             category='success')
             except Exception, e:
                 log.error(e)
                 return redirect(url('reset_password'))
         return redirect(url('login_home'))
     def logout(self):
         session.delete()
         log.info('Logging out and deleting session for user')
         redirect(url('home'))
     def authentication_token(self):
         """Return the CSRF protection token for the session - just like it
         could have been screen scrabed from a page with a form.
         Only intended for testing but might also be useful for other kinds
         of automation.
         """
         return h.authentication_token()

kallithea/tests/__init__.py

➞

Show inline comments

@@ @@ -192,36 +192,39 @@ class TestController(BaseTestCase): @@
         self.maxDiff = None
         self.index_location = config['app_conf']['index_dir']
     def log_user(self, username=TEST_USER_ADMIN_LOGIN,
                  password=TEST_USER_ADMIN_PASS):
         self._logged_username = username
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': username,
                                   'password': password})
         if 'invalid user name' in response.body:
             self.fail('could not login using %s %s' % (username, password))
         self.assertEqual(response.status, '302 Found')
         ses = response.session['authuser']
         self.assertEqual(ses.get('username'), username)
         response = response.follow()
         self.assertEqual(ses.get('is_authenticated'), True)
         return response.session['authuser']
     def _get_logged_user(self):
         return User.get_by_username(self._logged_username)
     def authentication_token(self):
         return self.app.get(url('authentication_token')).body
     def checkSessionFlash(self, response, msg, skip=0):
         if 'flash' not in response.session:
             self.fail(safe_str(u'msg `%s` not found - session has no flash ' % msg))
         try:
             level, m = response.session['flash'][-1 - skip]
             if msg in m:
                 return
         except IndexError:
             pass
         self.fail(safe_str(u'msg `%s` not found in session flash (skipping %s): %s' %
                            (msg, skip,
                             ', '.join('`%s`' % m for level, m in response.session['flash']))))

kallithea/tests/functional/test_admin_auth_settings.py

➞

Show inline comments

 from kallithea.tests import *
 from kallithea.model.db import Setting
 class TestAuthSettingsController(TestController):
     def _enable_plugins(self, plugins_list):
         test_url = url(controller='admin/auth_settings',
                        action='auth_settings')
         params={'auth_plugins': plugins_list,}
+        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')
     def test_ldap_save_settings(self):
         self.log_user()
         if not ldap_lib_installed:
             raise SkipTest('skipping due to missing ldap lib')
         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',

kallithea/tests/functional/test_admin_defaults.py

➞

Show inline comments

@@ @@ -11,65 +11,70 @@ class TestDefaultsController(TestControl @@
         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'))
     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)
     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)
     def test_update_browser_fakeout(self):
         response = self.app.post(url('default', id=1), params=dict(_method='put'))
+        response = self.app.post(url('default', id=1), params=dict(_method='put', _authentication_token=self.authentication_token()))
     def test_delete(self):
         response = self.app.delete(url('default', id=1))
     def test_delete_browser_fakeout(self):
         response = self.app.post(url('default', id=1), params=dict(_method='delete'))
+        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

@@ @@ -35,105 +35,110 @@ class TestGistsController(TestController @@
         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}, status=200)
                                  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'},
                                          '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'},
                                          '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'},
                                          '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'},
                                          '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'))
     def test_update(self):
         self.skipTest('not implemented')
         response = self.app.put(url('gist', gist_id=1))
     def test_delete(self):
         self.log_user()
         gist = _create_gist('delete-me')
         response = self.app.delete(url('gist', gist_id=gist.gist_id))
         self.checkSessionFlash(response, 'Deleted gist %s' % gist.gist_id)
     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)

kallithea/tests/functional/test_admin_permissions.py

➞

Show inline comments

 from kallithea.model.db import User, UserIpMap
 from kallithea.tests import *
 class TestAdminPermissionsController(TestController):
     def test_index(self):
         self.log_user()
         response = self.app.get(url('admin_permissions'))
         # Test response...
     def test_index_ips(self):
         self.log_user()
         response = self.app.get(url('admin_permissions_ips'))
         # Test response...
         response.mustcontain('All IP addresses are allowed')
     def test_add_ips(self):
         self.log_user()
         default_user_id = User.get_default_user().user_id
         response = self.app.put(url('edit_user_ips', id=default_user_id),
                                  params=dict(new_ip='127.0.0.0/24'))
                                  params=dict(new_ip='127.0.0.0/24',
                                  _authentication_token=self.authentication_token()))
         response = self.app.get(url('admin_permissions_ips'))
         response.mustcontain('127.0.0.0/24')
         response.mustcontain('127.0.0.0 - 127.0.0.255')
         ## delete
         default_user_id = User.get_default_user().user_id
         del_ip_id = UserIpMap.query().filter(UserIpMap.user_id ==
                                              default_user_id).first().ip_id
         response = self.app.post(url('edit_user_ips', id=default_user_id),
                                  params=dict(_method='delete',
                                              del_ip_id=del_ip_id))
                                              del_ip_id=del_ip_id,
                                              _authentication_token=self.authentication_token()))
         response = self.app.get(url('admin_permissions_ips'))
         response.mustcontain('All IP addresses are allowed')
         response.mustcontain(no=['127.0.0.0/24'])
         response.mustcontain(no=['127.0.0.0 - 127.0.0.255'])
     def test_index_overview(self):
         self.log_user()
         response = self.app.get(url('admin_permissions_perms'))
         # Test response...

kallithea/tests/functional/test_admin_repos.py

➞

Show inline comments

@@ @@ -35,219 +35,226 @@ class _BaseTest(TestController): @@
     NEW_REPO = None
     OTHER_TYPE_REPO = None
     OTHER_TYPE = None
     @classmethod
     def setup_class(cls):
         pass
     @classmethod
     def teardown_class(cls):
         pass
     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 = '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_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})
         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)
         # 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(os.path.join(TESTS_TMP_PATH, repo_name))
         except vcs.exceptions.VCSError:
             self.fail('no repo %s in filesystem' % repo_name)
         RepoModel().delete(repo_name)
         Session().commit()
     def test_create_non_ascii(self):
         self.log_user()
         non_ascii = "ąęł"
         repo_name = "%s%s" % (self.NEW_REPO, non_ascii)
         repo_name_unicode = repo_name.decode('utf8')
         description = 'description for newly created repo' + non_ascii
         description_unicode = description.decode('utf8')
         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_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})
         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)
         # 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(os.path.join(TESTS_TMP_PATH, repo_name))
         except vcs.exceptions.VCSError:
             self.fail('no repo %s in filesystem' % repo_name)
     def test_create_in_group(self):
         self.log_user()
         ## create GROUP
         group_name = 'sometest_%s' % self.REPO_TYPE
         gr = RepoGroupModel().create(group_name=group_name,
                                      group_description='test',
                                      owner=TEST_USER_ADMIN_LOGIN)
         Session().commit()
         repo_name = 'ingroup'
         repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])
         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,
                                                 repo_group=gr.group_id,))
                                                 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})
         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)
         # 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)
         # test if the repository was created on filesystem
         try:
             vcs.get_repo(os.path.join(TESTS_TMP_PATH, repo_name_full))
         except vcs.exceptions.VCSError:
             RepoGroupModel().delete(group_name)
             Session().commit()
             self.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 = 'reg_sometest_%s' % self.REPO_TYPE
         gr = RepoGroupModel().create(group_name=group_name,
                                      group_description='test',
                                      owner=TEST_USER_ADMIN_LOGIN)
         Session().commit()
         group_name_allowed = 'reg_sometest_allowed_%s' % self.REPO_TYPE
         gr_allowed = RepoGroupModel().create(group_name=group_name_allowed,
                                      group_description='test',
                                      owner=TEST_USER_REGULAR_LOGIN)
         Session().commit()
         repo_name = 'ingroup'
         repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])
         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,
                                                 repo_group=gr.group_id,))
                                                 repo_group=gr.group_id,
                                                 _authentication_token=authentication_token))
         response.mustcontain('Invalid value')
         # user is allowed to create in this group
         repo_name = 'ingroup'
         repo_name_full = RepoGroup.url_sep().join([group_name_allowed, repo_name])
         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,
                                                 repo_group=gr_allowed.group_id,))
                                                 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})
         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)
         # 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)
@@ @@ -266,49 +273,50 @@ class _BaseTest(TestController): @@
     def test_create_in_group_inherit_permissions(self):
         self.log_user()
         ## create GROUP
         group_name = 'sometest_%s' % self.REPO_TYPE
         gr = RepoGroupModel().create(group_name=group_name,
                                      group_description='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 = 'ingroup_inherited_%s' % self.REPO_TYPE
         repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])
         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,
                                                 repo_group=gr.group_id,
                                                 repo_copy_permissions=True))
                                                 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)
         # 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(os.path.join(TESTS_TMP_PATH, repo_name_full))
         except vcs.exceptions.VCSError:
             RepoGroupModel().delete(group_name)
@@ @@ -317,74 +325,77 @@ class _BaseTest(TestController): @@
         #check if inherited permissiona are applied
         inherited_perms = UserRepoToPerm.query()\
             .filter(UserRepoToPerm.repository_id == new_repo_id).all()
         self.assertEqual(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])
         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 = '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'))
                                                 clone_uri='http://127.0.0.1/repo',
                                                 _authentication_token=self.authentication_token()))
         response.mustcontain('invalid clone URL')
     def test_create_remote_repo_wrong_clone_uri_hg_svn(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,
                                                 clone_uri='svn+http://127.0.0.1/repo'))
                                                 clone_uri='svn+http://127.0.0.1/repo',
                                                 _authentication_token=self.authentication_token()))
         response.mustcontain('invalid clone URL')
     def test_delete(self):
         self.log_user()
         repo_name = 'vcs_test_new_to_delete_%s' % self.REPO_TYPE
         description = '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))
                                                 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)
         # 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(os.path.join(TESTS_TMP_PATH, repo_name))
         except vcs.exceptions.VCSError:
             self.fail('no repo %s in filesystem' % repo_name)
         response = self.app.delete(url('repo', repo_name=repo_name))
@@ @@ -392,247 +403,252 @@ class _BaseTest(TestController): @@
         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)
         self.assertEqual(os.path.isdir(os.path.join(TESTS_TMP_PATH, repo_name)),
                                   False)
     def test_delete_non_ascii(self):
         self.log_user()
         non_ascii = "ąęł"
         repo_name = "%s%s" % (self.NEW_REPO, non_ascii)
         repo_name_unicode = repo_name.decode('utf8')
         description = 'description for newly created repo' + non_ascii
         description_unicode = description.decode('utf8')
         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_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})
         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)
         # 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(os.path.join(TESTS_TMP_PATH, repo_name))
         except vcs.exceptions.VCSError:
             self.fail('no repo %s in filesystem' % repo_name)
         response = self.app.delete(url('repo', repo_name=repo_name))
         self.checkSessionFlash(response, 'Deleted repository %s' % (repo_name_unicode))
         response.follow()
         #check if repo was deleted from db
         deleted_repo = Session().query(Repository)\
             .filter(Repository.repo_name == repo_name_unicode).scalar()
         self.assertEqual(deleted_repo, None)
         self.assertEqual(os.path.isdir(os.path.join(TESTS_TMP_PATH, repo_name)),
                                   False)
     def test_delete_repo_with_group(self):
         #TODO:
         pass
     def test_delete_browser_fakeout(self):
         response = self.app.post(url('repo', repo_name=self.REPO),
                                  params=dict(_method='delete'))
+                                 params=dict(_method='delete', _authentication_token=self.authentication_token()))
     def test_show(self):
         self.log_user()
         response = self.app.get(url('repo', repo_name=self.REPO))
     def test_edit(self):
         response = self.app.get(url('edit_repo', repo_name=self.REPO))
     def test_set_private_flag_sets_default_to_none(self):
         self.log_user()
         #initially repository perm should be read
         perm = _get_permission_for_user(user='default', repo=self.REPO)
         self.assertTrue(len(perm), 1)
         self.assertEqual(perm[0].permission.permission_name, 'repository.read')
         self.assertEqual(Repository.get_by_repo_name(self.REPO).private, False)
         response = self.app.put(url('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))
                                                 user=TEST_USER_ADMIN_LOGIN,
                                                 _authentication_token=self.authentication_token()))
         self.checkSessionFlash(response,
                                msg='Repository %s updated successfully' % (self.REPO))
         self.assertEqual(Repository.get_by_repo_name(self.REPO).private, True)
         #now the repo default permission should be None
         perm = _get_permission_for_user(user='default', repo=self.REPO)
         self.assertTrue(len(perm), 1)
         self.assertEqual(perm[0].permission.permission_name, 'repository.none')
         response = self.app.put(url('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))
                                                 user=TEST_USER_ADMIN_LOGIN,
                                                 _authentication_token=self.authentication_token()))
         self.checkSessionFlash(response,
                                msg='Repository %s updated successfully' % (self.REPO))
         self.assertEqual(Repository.get_by_repo_name(self.REPO).private, False)
         #we turn off private now the repo default permission should stay None
         perm = _get_permission_for_user(user='default', repo=self.REPO)
         self.assertTrue(len(perm), 1)
         self.assertEqual(perm[0].permission.permission_name, 'repository.none')
         #update this permission back
         perm[0].permission = Permission.get_by_key('repository.read')
         Session().add(perm[0])
         Session().commit()
     def test_set_repo_fork_has_no_self_id(self):
         self.log_user()
         repo = Repository.get_by_repo_name(self.REPO)
         response = self.app.get(url('edit_repo_advanced', repo_name=self.REPO))
         opt = """<option value="%s">vcs_test_git</option>""" % repo.repo_id
         response.mustcontain(no=[opt])
     def test_set_fork_of_other_repo(self):
         self.log_user()
         other_repo = '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))
+                                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 repo %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))
+                                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))
+                                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 repo %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))
+                                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+'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))
                                                 repo_description=description,
                                                 _authentication_token=self.authentication_token()))
         response.mustcontain('no permission to create repository in root location')
         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))
                                                 repo_description=description,
                                                 _authentication_token=self.authentication_token()))
         self.checkSessionFlash(response,
                                'Error creating repository %s' % repo_name)
         # repo must not be in db
         repo = Repository.get_by_repo_name(repo_name)
         self.assertEqual(repo, None)
         # repo must not be in filesystem !
         self.assertFalse(os.path.isdir(os.path.join(TESTS_TMP_PATH, repo_name)))
 class TestAdminReposControllerGIT(_BaseTest):
     REPO = GIT_REPO
     REPO_TYPE = 'git'
     NEW_REPO = NEW_GIT_REPO
     OTHER_TYPE_REPO = HG_REPO
     OTHER_TYPE = 'hg'
 class TestAdminReposControllerHG(_BaseTest):
     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

@@ @@ -16,159 +16,166 @@ class TestAdminSettingsController(TestCo @@
     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 /tmp'))
                                             new_hook_ui_value='cd /tmp',
                                             _authentication_token=self.authentication_token()))
         response = response.follow()
         response.mustcontain('test_hooks_1')
         response.mustcontain('cd /tmp')
     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 /tmp2'))
                                             new_hook_ui_value='cd /tmp2',
                                             _authentication_token=self.authentication_token()))
         response = response.follow()
         response.mustcontain('test_hooks_2')
         response.mustcontain('cd /tmp2')
         hook_id = Ui.get_by_key('test_hooks_2').ui_id
         ## delete
         self.app.post(url('admin_settings_hooks'),
                         params=dict(hook_id=hook_id))
+                        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 /tmp2'])
     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)
         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)
         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')
         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'], '')
         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'))
             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 = 'admins_test'
 class TestAdminUsersGroupsController(TestController):
     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': 'DESC',
                                   'active': True})
                                   '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))
     def test_update_browser_fakeout(self):
         response = self.app.post(url('users_group', id=1),
                                  params=dict(_method='put'))
+                                 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': 'DESC',
                                   'active': True})
                                   '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.delete(url('users_group', id=gr.users_group_id))
         gr = Session().query(UserGroup)\
             .filter(UserGroup.users_group_name == users_group_name).scalar()
         self.assertEqual(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': 'DESC',
                                   'active': True})
                                   '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})
                                  {'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],
                     [ug.users_group_id, p2.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), {})
         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')
@@ @@ -114,58 +117,59 @@ class TestAdminUsersGroupsController(Tes @@
                     [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.delete(url('users_group', id=ug.users_group_id))
         response = response.follow()
         gr = Session().query(UserGroup)\
             .filter(UserGroup.users_group_name == users_group_name).scalar()
         self.assertEqual(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, [])
     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': 'DESC',
                                   'active': True})
                                   '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})
+                                {'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],
                     [ug.users_group_id, p2.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), {})
         response.follow()
         ug = UserGroup.get_by_group_name(users_group_name)
         p = Permission.get_by_key('hg.create.none')
@@ @@ -183,40 +187,40 @@ class TestAdminUsersGroupsController(Tes @@
                     [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.delete(url('users_group', id=ug.users_group_id))
         response = response.follow()
         gr = Session().query(UserGroup)\
                            .filter(UserGroup.users_group_name ==
                                    users_group_name).scalar()
         self.assertEqual(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,
             []
+        )
     def test_delete_browser_fakeout(self):
         response = self.app.post(url('users_group', id=1),
                                  params=dict(_method='delete'))
+                                 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

@@ @@ -37,80 +37,82 @@ class TestAdminUsersController(TestContr @@
     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 = 'name'
         lastname = 'lastname'
         email = 'mail@mail.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})
              '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)
         response.follow()
         response = response.follow()
         response.mustcontain("""newtestuser""")
     def test_create_err(self):
         self.log_user()
         username = 'new_user'
         password = ''
         name = 'name'
         lastname = 'lastname'
         email = 'errmail.com'
         response = self.app.post(url('users'), {'username': username,
                                                'password': password,
                                                'name': name,
                                                'active': False,
                                                'lastname': lastname,
                                                'email': email})
                                                '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'
     def test_new(self):
         self.log_user()
         response = self.app.get(url('new_user'))
     @parameterized.expand(
         [('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'}),
@@ @@ -124,50 +126,52 @@ class TestAdminUsersController(TestContr @@
     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)
     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.delete(url('user', id=new_user.user_id))
         self.checkSessionFlash(response, 'Successfully deleted user')
     def test_delete_repo_err(self):
         self.log_user()
         username = 'repoerr'
         reponame = 'repoerr_fail'
@@ @@ -245,249 +249,251 @@ class TestAdminUsersController(TestContr @@
     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='a',
                                             lastname='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)
             response = self.app.post(url('edit_user_perms', id=uid),
                                      params=dict(_method='put',
                                                  create_repo_perm=True))
                                                  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)
         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='a',
                                             lastname='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)
             response = self.app.post(url('edit_user_perms', id=uid),
                                      params=dict(_method='put'))
+                                     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)
         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='a',
                                             lastname='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)
             response = self.app.post(url('edit_user_perms', id=uid),
                                      params=dict(_method='put',
                                                  create_repo_perm=True))
                                                  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)
         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='a',
                                             lastname='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)
             response = self.app.post(url('edit_user_perms', id=uid),
                                      params=dict(_method='put'))
+                                     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)
         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')
     @parameterized.expand([
         ('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))
+                                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))
+                      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')
     @parameterized.expand([
         ('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),
                  {'_method': 'put', 'description': desc, 'lifetime': lifetime})
+                 {'_method': 'put', '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),
                 {'_method': 'put', 'description': 'desc', 'lifetime': -1})
+                {'_method': 'put', '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))
         response = self.app.post(url('edit_user_api_keys', id=user_id),
                  {'_method': 'delete', 'del_api_key': keys[0].api_key})
+                 {'_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))
     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})
+                 {'_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])

kallithea/tests/functional/test_changeset_comments.py

➞

Show inline comments

@@ @@ -8,152 +8,152 @@ class TestChangeSetCommentsController(Te @@
     def setUp(self):
         for x in ChangesetComment.query().all():
             Session().delete(x)
         Session().commit()
         for x in Notification.query().all():
             Session().delete(x)
         Session().commit()
     def tearDown(self):
         for x in ChangesetComment.query().all():
             Session().delete(x)
         Session().commit()
         for x in Notification.query().all():
             Session().delete(x)
         Session().commit()
     def test_create(self):
         self.log_user()
         rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
         text = u'CommentOnRevision'
         params = {'text': text}
+        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)
         # Test response...
         self.assertEqual(response.status, '302 Found')
         response.follow()
         response = self.app.get(url(controller='changeset', action='index',
                                 repo_name=HG_REPO, revision=rev))
         # test DB
         self.assertEqual(ChangesetComment.query().count(), 1)
         response.mustcontain(
             '''<div class="comments-number">'''
             ''' 1 comment (0 inline, 1 general)'''
+        )
         self.assertEqual(Notification.query().count(), 1)
         self.assertEqual(ChangesetComment.query().count(), 1)
         notification = Notification.query().all()[0]
         ID = ChangesetComment.query().first().comment_id
         self.assertEqual(notification.type_,
                          Notification.TYPE_CHANGESET_COMMENT)
         sbj = (u'/vcs_test_hg/changeset/'
                '27cd5cce30c96924232dffcd24178a07ffeb5dfc#comment-%s' % ID)
         print "%s vs %s" % (sbj, notification.subject)
         self.assertTrue(sbj in notification.subject)
     def test_create_inline(self):
         self.log_user()
         rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
         text = u'CommentOnRevision'
         f_path = 'vcs/web/simplevcs/views/repository.py'
         line = 'n1'
         params = {'text': text, 'f_path': f_path, 'line': line}
+        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)
         # Test response...
         self.assertEqual(response.status, '302 Found')
         response.follow()
         response = self.app.get(url(controller='changeset', action='index',
                                 repo_name=HG_REPO, revision=rev))
         #test DB
         self.assertEqual(ChangesetComment.query().count(), 1)
         response.mustcontain(
             '''<div class="comments-number">'''
             ''' 1 comment (1 inline, 0 general)'''
+        )
         response.mustcontain(
             '''<div style="display:none" class="inline-comment-placeholder" '''
             '''path="vcs/web/simplevcs/views/repository.py" '''
             '''target_id="vcswebsimplevcsviewsrepositorypy">'''
+        )
         self.assertEqual(Notification.query().count(), 1)
         self.assertEqual(ChangesetComment.query().count(), 1)
         notification = Notification.query().all()[0]
         ID = ChangesetComment.query().first().comment_id
         self.assertEqual(notification.type_,
                          Notification.TYPE_CHANGESET_COMMENT)
         sbj = (u'/vcs_test_hg/changeset/'
                '27cd5cce30c96924232dffcd24178a07ffeb5dfc#comment-%s' % ID)
         print "%s vs %s" % (sbj, notification.subject)
         self.assertTrue(sbj in notification.subject)
     def test_create_with_mention(self):
         self.log_user()
         rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
         text = u'@test_regular check CommentOnRevision'
         params = {'text':text}
+        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)
         # Test response...
         self.assertEqual(response.status, '302 Found')
         response.follow()
         response = self.app.get(url(controller='changeset', action='index',
                                 repo_name=HG_REPO, revision=rev))
         # test DB
         self.assertEqual(ChangesetComment.query().count(), 1)
         response.mustcontain(
             '''<div class="comments-number">'''
             ''' 1 comment (0 inline, 1 general)'''
+        )
         self.assertEqual(Notification.query().count(), 2)
         users = [x.user.username for x in UserNotification.query().all()]
         # test_regular gets notification by @mention
         self.assertEqual(sorted(users), [u'test_admin', u'test_regular'])
     def test_delete(self):
         self.log_user()
         rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
         text = u'CommentOnRevision'
         params = {'text': text}
+        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)
         comments = ChangesetComment.query().all()
         self.assertEqual(len(comments), 1)
         comment_id = comments[0].comment_id
         self.app.delete(url(controller='changeset',
                                     action='delete_comment',
                                     repo_name=HG_REPO,
                                     comment_id=comment_id))
         comments = ChangesetComment.query().all()
         self.assertEqual(len(comments), 0)
         response = self.app.get(url(controller='changeset', action='index',
                                 repo_name=HG_REPO, revision=rev))
         response.mustcontain(
             '''<div class="comments-number">'''
             ''' 0 comments (0 inline, 0 general)'''
+        )

kallithea/tests/functional/test_files.py

➞

Show inline comments

@@ @@ -307,437 +307,457 @@ removed extra unicode conversion in diff @@
     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': ''
                                     '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"
                                     'content': "foo",
                                     '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         self.checkSessionFlash(response, 'No filename')
     @parameterized.expand([
         ('/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
                                     'location': location,
                                     '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         self.checkSessionFlash(response, 'Location must be relative path and must not contain .. in path')
     @parameterized.expand([
         (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('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
                                     'location': location,
                                     '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % os.path.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': ''
                                      '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"
                                     'content': "foo",
                                     '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         self.checkSessionFlash(response, 'No filename')
     @parameterized.expand([
         ('/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
                                     'location': location,
                                     '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         self.checkSessionFlash(response, 'Location must be relative path and must not contain .. in path')
     @parameterized.expand([
         (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('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
                                     'location': location,
                                     '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % os.path.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('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
                                     'location': location,
                                     '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % os.path.join(location, filename))
             response = self.app.get(url('files_edit_home',
                                           repo_name=repo.repo_name,
                                           revision='tip', f_path='vcs/nodes.py'),
                                     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('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
                                     'location': location,
                                     '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % os.path.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='vcs/nodes.py'),
                                      params={
                                         'content': "def py():\n print 'hello world'\n",
                                         'message': 'i commited',
                                         '_authentication_token': self.authentication_token(),
                                      },
                                     status=302)
             self.checkSessionFlash(response,
                                    'Successfully committed to vcs/nodes.py')
         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('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
                                     'location': location,
                                     '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % os.path.join(location, filename))
             response = self.app.get(url('files_edit_home',
                                           repo_name=repo.repo_name,
                                           revision='tip', f_path='vcs/nodes.py'),
                                     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('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
                                     'location': location,
                                     '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % os.path.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='vcs/nodes.py'),
                                      params={
                                         'content': "def py():\n print 'hello world'\n",
                                         'message': 'i commited',
                                         '_authentication_token': self.authentication_token(),
                                      },
                                     status=302)
             self.checkSessionFlash(response,
                                    'Successfully committed to vcs/nodes.py')
         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('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
                                     'location': location,
                                     '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % os.path.join(location, filename))
             response = self.app.get(url('files_delete_home',
                                           repo_name=repo.repo_name,
                                           revision='tip', f_path='vcs/nodes.py'),
                                     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('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
                                     'location': location,
                                     '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % os.path.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='vcs/nodes.py'),
                                      params={
                                         'message': 'i commited',
                                         '_authentication_token': self.authentication_token(),
                                      },
                                     status=302)
             self.checkSessionFlash(response,
                                    'Successfully deleted file vcs/nodes.py')
         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('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
                                     'location': location,
                                     '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % os.path.join(location, filename))
             response = self.app.get(url('files_delete_home',
                                           repo_name=repo.repo_name,
                                           revision='tip', f_path='vcs/nodes.py'),
                                     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('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
                                     'location': location,
                                     '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % os.path.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='vcs/nodes.py'),
                                      params={
                                         'message': 'i commited',
                                         '_authentication_token': self.authentication_token(),
                                      },
                                     status=302)
             self.checkSessionFlash(response,
                                    'Successfully deleted file vcs/nodes.py')
         finally:
             fixture.destroy_repo(repo.repo_name)

kallithea/tests/functional/test_forks.py

➞

Show inline comments

@@ @@ -39,139 +39,142 @@ class _BaseTest(TestController): @@
     def tearDown(self):
         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), {}, status=403)
+                          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': '',
             'fork_parent_id': org_repo.repo_id,
             'repo_type': self.REPO_TYPE,
             'description': description,
             'private': 'False',
             'landing_rev': 'rev:tip'}
             '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.delete(url('repo', repo_name=fork_name))
     def test_fork_create_into_group(self):
         self.log_user()
         group = fixture.create_repo_group('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'}
             'landing_rev': 'rev:tip',
             '_authentication_token': self.authentication_token()}
         self.app.post(url(controller='forks', action='fork_create',
                           repo_name=repo_name), creation_args)
         repo = Repository.get_by_repo_name(fork_name_full)
         assert repo.fork.repo_name == self.REPO
         ## run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=fork_name_full))
         #test if we have a message that fork is ok
         self.checkSessionFlash(response,
                 'Forked repository %s as <a href="/%s">%s</a>'
                 % (repo_name, fork_name_full, fork_name_full))
         #test if the fork was created in the database
         fork_repo = Session().query(Repository)\
             .filter(Repository.repo_name == fork_name_full).one()
         self.assertEqual(fork_repo.repo_name, fork_name_full)
         self.assertEqual(fork_repo.fork.repo_name, repo_name)
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=fork_name_full))
         response.mustcontain(fork_name_full)
         response.mustcontain(self.REPO_TYPE)
         response.mustcontain('Fork of "<a href="/%s">%s</a>"' % (repo_name, repo_name))
         fixture.destroy_repo(fork_name_full)
         fixture.destroy_repo_group(group_id)
     def test_z_fork_create(self):
         self.log_user()
         fork_name = self.REPO_FORK
         description = 'fork of vcs test'
         repo_name = self.REPO
         org_repo = Repository.get_by_repo_name(repo_name)
         creation_args = {
             'repo_name': fork_name,
             'repo_group': '',
             'fork_parent_id': org_repo.repo_id,
             'repo_type': self.REPO_TYPE,
             'description': description,
             'private': 'False',
             'landing_rev': 'rev:tip'}
             'landing_rev': 'rev:tip',
             '_authentication_token': self.authentication_token()}
         self.app.post(url(controller='forks', action='fork_create',
                           repo_name=repo_name), creation_args)
         repo = Repository.get_by_repo_name(self.REPO_FORK)
         assert repo.fork.repo_name == self.REPO
         ## run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=fork_name))
         #test if we have a message that fork is ok
         self.checkSessionFlash(response,
                 'Forked repository %s as <a href="/%s">%s</a>'
                 % (repo_name, fork_name, fork_name))
         #test if the fork was created in the database
         fork_repo = Session().query(Repository)\
             .filter(Repository.repo_name == fork_name).one()
         self.assertEqual(fork_repo.repo_name, fork_name)
         self.assertEqual(fork_repo.fork.repo_name, repo_name)
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=fork_name))
         response.mustcontain(fork_name)
         response.mustcontain(self.REPO_TYPE)
         response.mustcontain('Fork of "<a href="/%s">%s</a>"' % (repo_name, repo_name))

kallithea/tests/functional/test_my_account.py

➞

Show inline comments

@@ @@ -29,214 +29,218 @@ class TestMyAccountController(TestContro @@
         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})
+                                 {'new_email': TEST_USER_REGULAR_EMAIL, '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'This e-mail address is already taken')
     def test_my_account_my_emails_add_mising_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'),)
         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': 'foo@barz.com'})
+                                 {'new_email': 'foo@barz.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 == 'foo@barz.com').one().email_id
         response.mustcontain('foo@barz.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'})
+                                 {'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')
     @parameterized.expand(
         [('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': 'some@email.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)
     def test_my_account_update_err_email_exists(self):
         self.log_user()
         new_email = 'test_regular@mail.com'  # already exisitn email
         response = self.app.post(url('my_account'),
                                 params=dict(
                                     username='test_admin',
                                     new_password='test12',
                                     password_confirmation='test122',
                                     firstname='NewName',
                                     lastname='NewLastname',
                                     email=new_email,)
                                     email=new_email,
                                     _authentication_token=self.authentication_token())
+                                )
         response.mustcontain('This e-mail address is already taken')
     def test_my_account_update_err(self):
         self.log_user('test_regular2', 'test12')
         new_email = 'newmail.pl'
         response = self.app.post(url('my_account'),
                                  params=dict(
                                             username='test_admin',
                                             new_password='test12',
                                             password_confirmation='test122',
                                             firstname='NewName',
                                             lastname='NewLastname',
                                             email=new_email,))
                                             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_admin'})
         response.mustcontain(u"%s" % msg)
     def test_my_account_api_keys(self):
         usr = self.log_user('test_regular2', 'test12')
         user = User.get(usr['user_id'])
         response = self.app.get(url('my_account_api_keys'))
         response.mustcontain(user.api_key)
         response.mustcontain('expires: never')
     @parameterized.expand([
         ('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_regular2', 'test12')
         user = User.get(usr['user_id'])
         response = self.app.post(url('my_account_api_keys'),
                                  {'description': desc, 'lifetime': lifetime})
+                                 {'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_regular2', 'test12')
         user = User.get(usr['user_id'])
         response = self.app.post(url('my_account_api_keys'),
                                  {'description': 'desc', 'lifetime': -1})
+                                 {'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))
         response = self.app.post(url('my_account_api_keys'),
                  {'_method': 'delete', 'del_api_key': keys[0].api_key})
+                 {'_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))
     def test_my_account_reset_main_api_key(self):
         usr = self.log_user('test_regular2', 'test12')
         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})
+                 {'_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])

0 comments (0 inline, 0 general)