kallithea Changeset - 09100b3b8f42

Changeset - 09100b3b8f42

Parent rev.

Child rev.

[Not reviewed]

default

0 22 0

Mads Kiilerich - 6 years ago 2019-08-06 22:42:37
mads@kiilerich.com

Grafted from: fe1d525763c3

helpers: change CSRF protection POST parameter name to "_session_csrf_secret_token" and fix up tests to use new names

22 files changed with 217 insertions and 217 deletions:

kallithea/config/routing.py

kallithea/controllers/login.py

kallithea/lib/helpers.py

kallithea/public/js/base.js

kallithea/templates/admin/gists/edit.html

kallithea/tests/base.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_repo_groups.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_pullrequests_comments.py

kallithea/tests/functional/test_files.py

kallithea/tests/functional/test_forks.py

kallithea/tests/functional/test_login.py

kallithea/tests/functional/test_my_account.py

kallithea/tests/functional/test_pullrequests.py

kallithea/tests/functional/test_repo_groups.py

0 comments (0 inline, 0 general)

kallithea/config/routing.py

➞

Show inline comments

@@ @@ -444,13 +444,13 @@ def make_map(config): @@
     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('session_csrf_secret_token', '%s/session_csrf_secret_token' % ADMIN_PREFIX, controller='login', action='session_csrf_secret_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')

kallithea/controllers/login.py

➞

Show inline comments

@@ @@ -246,13 +246,13 @@ class LoginController(BaseController): @@
     def logout(self):
         session.delete()
         log.info('Logging out and deleting session for user')
         raise HTTPFound(location=url('home'))
-    def authentication_token(self):
+    def session_csrf_secret_token(self):
         """Return the CSRF protection token for the session - just like it
         could have been screen scraped from a page with a form.
         Only intended for testing but might also be useful for other kinds
         of automation.
         """
         return h.session_csrf_secret_token()

kallithea/lib/helpers.py

➞

Show inline comments

@@ @@ -1270,13 +1270,13 @@ def not_mapped_error(repo_name): @@
 def ip_range(ip_addr):
     from kallithea.model.db import UserIpMap
     s, e = UserIpMap._get_ip_range(ip_addr)
     return '%s - %s' % (s, e)
-session_csrf_secret_name = "_authentication_token"
+session_csrf_secret_name = "_session_csrf_secret_token"
 def session_csrf_secret_token():
     """Return (and create) the current session's CSRF protection token."""
     from tg import session
     if not session_csrf_secret_name in session:
         session[session_csrf_secret_name] = str(random.getrandbits(128))

kallithea/public/js/base.js

➞

Show inline comments

@@ @@ -405,13 +405,13 @@ var ajaxGET = function(url, success, fai @@
     return $.ajax({url: url, headers: {'X-PARTIAL-XHR': '1'}, cache: false})
         .done(success)
         .fail(failure);
 };
 var ajaxPOST = function(url, postData, success, failure) {
-    postData['_authentication_token'] = _session_csrf_secret_token;
+    postData['_session_csrf_secret_token'] = _session_csrf_secret_token;
     var postData = _toQueryString(postData);
     if(failure === undefined) {
         failure = function(jqXHR, textStatus, errorThrown) {
                 if (textStatus != "abort")
                     alert("Error posting to server: " + textStatus);
             };
@@ @@ -455,21 +455,21 @@ var _onSuccessFollow = function(target){ @@
+        }
+    }
+}
 var toggleFollowingRepo = function(target, follows_repository_id){
     var args = 'follows_repository_id=' + follows_repository_id;
-    args += '&amp;_authentication_token=' + _session_csrf_secret_token;
+    args += '&amp;_session_csrf_secret_token=' + _session_csrf_secret_token;
     $.post(TOGGLE_FOLLOW_URL, args, function(data){
             _onSuccessFollow(target);
         });
     return false;
 };
 var showRepoSize = function(target, repo_name){
-    var args = '_authentication_token=' + _session_csrf_secret_token;
+    var args = '_session_csrf_secret_token=' + _session_csrf_secret_token;
     if(!$("#" + target).hasClass('loaded')){
         $("#" + target).html(_TM['Loading ...']);
         var url = pyroutes.url('repo_size', {"repo_name":repo_name});
         $.post(url, args, function(data) {
             $("#" + target).html(data);

kallithea/templates/admin/gists/edit.html

➞

Show inline comments

@@ @@ -150,13 +150,13 @@ @@
               $('#update').on('click', function(e){
                   e.preventDefault();
                   // check for newer version.
                   $.ajax({
                     url: ${h.js(h.url('edit_gist_check_revision', gist_id=c.gist.gist_access_id))},
-                    data: {'revision': ${h.js(c.file_changeset.raw_id)}, '_authentication_token': _session_csrf_secret_token},
+                    data: {'revision': ${h.js(c.file_changeset.raw_id)}, '_session_csrf_secret_token': _session_csrf_secret_token},
                     dataType: 'json',
                     type: 'POST',
                     success: function(data) {
                       if(data.success == false){
                           $('#edit_error').show();
+                      }

kallithea/tests/base.py

➞

Show inline comments

@@ @@ -154,13 +154,13 @@ class TestController(object): @@
     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,
-                                  '_authentication_token': self.authentication_token()})
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})
         if 'Invalid username or password' in response.body:
             pytest.fail('could not login using %s %s' % (username, password))
         assert response.status == '302 Found'
         self.assert_authenticated_user(response, username)
@@ @@ -175,14 +175,14 @@ class TestController(object): @@
         cookie = response.session.get('authuser')
         user = cookie and cookie.get('user_id')
         user = user and User.get(user)
         user = user and user.username
         assert user == expected_username
     def authentication_token(self):
         return self.app.get(url('authentication_token')).body
     def session_csrf_secret_token(self):
         return self.app.get(url('session_csrf_secret_token')).body
     def checkSessionFlash(self, response, msg=None, skip=0, _matcher=lambda msg, m: msg in m):
         if 'flash' not in response.session:
             pytest.fail(safe_str(u'msg `%s` not found - session has no flash:\n%s' % (msg, response)))
         try:
             level, m = response.session['flash'][-1 - skip]

kallithea/tests/functional/test_admin_auth_settings.py

➞

Show inline comments

@@ @@ -3,13 +3,13 @@ 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, '_authentication_token': self.authentication_token()}
+        params={'auth_plugins': plugins_list, '_session_csrf_secret_token': self.session_csrf_secret_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

kallithea/tests/functional/test_admin_defaults.py

➞

Show inline comments

@@ @@ -15,30 +15,30 @@ class TestDefaultsController(TestControl @@
         self.log_user()
         params = {
             'default_repo_enable_downloads': True,
             'default_repo_enable_statistics': True,
             'default_repo_private': True,
             'default_repo_type': 'hg',
-            '_authentication_token': self.authentication_token(),
+            '_session_csrf_secret_token': self.session_csrf_secret_token(),
+        }
         response = self.app.post(url('defaults_update', id='default'), params=params)
         self.checkSessionFlash(response, 'Default settings updated successfully')
-        params.pop('_authentication_token')
+        params.pop('_session_csrf_secret_token')
         defs = Setting.get_default_repo_settings()
         assert params == defs
     def test_update_params_false_git(self):
         self.log_user()
         params = {
             'default_repo_enable_downloads': False,
             'default_repo_enable_statistics': False,
             'default_repo_private': False,
             'default_repo_type': 'git',
-            '_authentication_token': self.authentication_token(),
+            '_session_csrf_secret_token': self.session_csrf_secret_token(),
+        }
         response = self.app.post(url('defaults_update', id='default'), params=params)
         self.checkSessionFlash(response, 'Default settings updated successfully')
-        params.pop('_authentication_token')
+        params.pop('_session_csrf_secret_token')
         defs = Setting.get_default_repo_settings()
         assert params == defs

kallithea/tests/functional/test_admin_gists.py

➞

Show inline comments

@@ @@ -53,25 +53,25 @@ class TestGistsController(TestController @@
         # and privates
         response.mustcontain('gist: %s' % gist.gist_access_id)
     def test_create_missing_description(self):
         self.log_user()
         response = self.app.post(url('gists'),
-                                 params={'lifetime': -1, '_authentication_token': self.authentication_token()},
+                                 params={'lifetime': -1, '_session_csrf_secret_token': self.session_csrf_secret_token()},
                                  status=200)
         response.mustcontain('Missing value')
     def test_create(self):
         self.log_user()
         response = self.app.post(url('gists'),
                                  params={'lifetime': -1,
                                          'content': 'gist test',
                                          'filename': 'foo',
                                          'public': 'public',
-                                         '_authentication_token': self.authentication_token()},
+                                         '_session_csrf_secret_token': self.session_csrf_secret_token()},
                                  status=302)
         response = response.follow()
         response.mustcontain('added file: foo')
         response.mustcontain('gist test')
         response.mustcontain('<div class="label label-success">Public Gist</div>')
@@ @@ -79,13 +79,13 @@ class TestGistsController(TestController @@
         self.log_user()
         response = self.app.post(url('gists'),
                                  params={'lifetime': -1,
                                          'content': 'gist test',
                                          'filename': '/home/foo',
                                          'public': 'public',
-                                         '_authentication_token': self.authentication_token()},
+                                         '_session_csrf_secret_token': self.session_csrf_secret_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')
@@ @@ -98,13 +98,13 @@ class TestGistsController(TestController @@
         self.log_user()
         response = self.app.post(url('gists'),
                                  params={'lifetime': -1,
                                          'content': 'private gist test',
                                          'filename': 'private-foo',
                                          'private': 'private',
-                                         '_authentication_token': self.authentication_token()},
+                                         '_session_csrf_secret_token': self.session_csrf_secret_token()},
                                  status=302)
         response = response.follow()
         response.mustcontain('added file: private-foo<')
         response.mustcontain('private gist test')
         response.mustcontain('<div class="label label-warning">Private Gist</div>')
@@ @@ -113,13 +113,13 @@ class TestGistsController(TestController @@
         response = self.app.post(url('gists'),
                                  params={'lifetime': -1,
                                          'content': 'gist test',
                                          'filename': 'foo-desc',
                                          'description': 'gist-desc',
                                          'public': 'public',
-                                         '_authentication_token': self.authentication_token()},
+                                         '_session_csrf_secret_token': self.session_csrf_secret_token()},
                                  status=302)
         response = response.follow()
         response.mustcontain('added file: foo-desc')
         response.mustcontain('gist test')
         response.mustcontain('gist-desc')
         response.mustcontain('<div class="label label-success">Public Gist</div>')
@@ @@ -129,25 +129,25 @@ class TestGistsController(TestController @@
         response = self.app.get(url('new_gist'))
     def test_delete(self):
         self.log_user()
         gist = _create_gist('delete-me')
         response = self.app.post(url('gist_delete', gist_id=gist.gist_id),
-            params={'_authentication_token': self.authentication_token()})
+            params={'_session_csrf_secret_token': self.session_csrf_secret_token()})
     def test_delete_normal_user_his_gist(self):
         self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
         gist = _create_gist('delete-me', owner=TEST_USER_REGULAR_LOGIN)
         response = self.app.post(url('gist_delete', gist_id=gist.gist_id),
-            params={'_authentication_token': self.authentication_token()})
+            params={'_session_csrf_secret_token': self.session_csrf_secret_token()})
     def test_delete_normal_user_not_his_own_gist(self):
         self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
         gist = _create_gist('delete-me')
         response = self.app.post(url('gist_delete', gist_id=gist.gist_id), status=403,
-            params={'_authentication_token': self.authentication_token()})
+            params={'_session_csrf_secret_token': self.session_csrf_secret_token()})
     def test_show(self):
         gist = _create_gist('gist-show-me')
         response = self.app.get(url('gist', gist_id=gist.gist_access_id))
         response.mustcontain('added file: gist-show-me<')
         response.mustcontain('%s - created' % TEST_USER_ADMIN_LOGIN)

kallithea/tests/functional/test_admin_permissions.py

➞

Show inline comments

@@ @@ -26,13 +26,13 @@ class TestAdminPermissionsController(Tes @@
         default_user_id = User.get_default_user().user_id
         # Add IP and verify it is shown in UI and both gives access and rejects
         response = self.app.post(url('edit_user_ips_update', id=default_user_id),
                                  params=dict(new_ip='0.0.0.0/24',
-                                 _authentication_token=self.authentication_token()))
+                                 _session_csrf_secret_token=self.session_csrf_secret_token()))
         invalidate_all_caches()
         response = self.app.get(url('admin_permissions_ips'),
                                 extra_environ={'REMOTE_ADDR': '0.0.0.1'})
         response.mustcontain('0.0.0.0/24')
         response.mustcontain('0.0.0.0 - 0.0.0.255')
@@ @@ -40,35 +40,35 @@ class TestAdminPermissionsController(Tes @@
                                 extra_environ={'REMOTE_ADDR': '0.0.1.1'}, status=403)
         # Add another IP and verify previously rejected now works
         response = self.app.post(url('edit_user_ips_update', id=default_user_id),
                                  params=dict(new_ip='0.0.1.0/24',
-                                 _authentication_token=self.authentication_token()))
+                                 _session_csrf_secret_token=self.session_csrf_secret_token()))
         invalidate_all_caches()
         response = self.app.get(url('admin_permissions_ips'),
                                 extra_environ={'REMOTE_ADDR': '0.0.1.1'})
         # Delete latest IP and verify same IP is rejected again
         x = UserIpMap.query().filter_by(ip_addr='0.0.1.0/24').first()
         response = self.app.post(url('edit_user_ips_delete', id=default_user_id),
                                  params=dict(del_ip_id=x.ip_id,
-                                             _authentication_token=self.authentication_token()))
+                                             _session_csrf_secret_token=self.session_csrf_secret_token()))
         invalidate_all_caches()
         response = self.app.get(url('admin_permissions_ips'),
                                 extra_environ={'REMOTE_ADDR': '0.0.1.1'}, status=403)
         # Delete first IP and verify unlimited access again
         x = UserIpMap.query().filter_by(ip_addr='0.0.0.0/24').first()
         response = self.app.post(url('edit_user_ips_delete', id=default_user_id),
                                  params=dict(del_ip_id=x.ip_id,
-                                             _authentication_token=self.authentication_token()))
+                                             _session_csrf_secret_token=self.session_csrf_secret_token()))
         invalidate_all_caches()
         response = self.app.get(url('admin_permissions_ips'),
                                 extra_environ={'REMOTE_ADDR': '0.0.1.1'})
     def test_index_overview(self):
@@ @@ -83,24 +83,24 @@ class TestAdminPermissionsController(Tes @@
         response = self.app.post(
             url('edit_repo_perms_update', repo_name=HG_REPO),
             params=dict(
                 perm_new_member_1='repository.read',
                 perm_new_member_name_1=user.username,
                 perm_new_member_type_1='user',
-                _authentication_token=self.authentication_token()),
+                _session_csrf_secret_token=self.session_csrf_secret_token()),
             status=302)
         assert not response.location.endswith(url('edit_repo_perms_update', repo_name=HG_REPO))
         assert response.location.endswith(url('login_home', came_from=url('edit_repo_perms_update', repo_name=HG_REPO)))
         response = self.app.post(
             url('edit_repo_perms_revoke', repo_name=HG_REPO),
             params=dict(
                 obj_type='user',
                 user_id=user.user_id,
-                _authentication_token=self.authentication_token()),
+                _session_csrf_secret_token=self.session_csrf_secret_token()),
             status=302)
         assert response.location.endswith(url('login_home', came_from=url('edit_repo_perms_revoke', repo_name=HG_REPO)))
         # Test authenticated access
         self.log_user()
@@ @@ -108,19 +108,19 @@ class TestAdminPermissionsController(Tes @@
         response = self.app.post(
             url('edit_repo_perms_update', repo_name=HG_REPO),
             params=dict(
                 perm_new_member_1='repository.read',
                 perm_new_member_name_1=user.username,
                 perm_new_member_type_1='user',
-                _authentication_token=self.authentication_token()),
+                _session_csrf_secret_token=self.session_csrf_secret_token()),
             status=302)
         assert response.location.endswith(url('edit_repo_perms_update', repo_name=HG_REPO))
         response = self.app.post(
             url('edit_repo_perms_revoke', repo_name=HG_REPO),
             params=dict(
                 obj_type='user',
                 user_id=user.user_id,
-                _authentication_token=self.authentication_token()),
+                _session_csrf_secret_token=self.session_csrf_secret_token()),
             status=200)
         assert not response.body

kallithea/tests/functional/test_admin_repo_groups.py

➞

Show inline comments

@@ @@ -12,16 +12,16 @@ class TestRepoGroupsController(TestContr @@
     def test_case_insensitivity(self):
         self.log_user()
         group_name = u'newgroup'
         response = self.app.post(url('repos_groups'),
                                  fixture._get_repo_group_create_params(group_name=group_name,
-                                                                 _authentication_token=self.authentication_token()))
+                                                                 _session_csrf_secret_token=self.session_csrf_secret_token()))
         # try to create repo group with swapped case
         swapped_group_name = group_name.swapcase()
         response = self.app.post(url('repos_groups'),
                                  fixture._get_repo_group_create_params(group_name=swapped_group_name,
-                                                                 _authentication_token=self.authentication_token()))
+                                                                 _session_csrf_secret_token=self.session_csrf_secret_token()))
         response.mustcontain('already exists')
         RepoGroupModel().delete(group_name)
         Session().commit()

kallithea/tests/functional/test_admin_repos.py

➞

Show inline comments

@@ @@ -50,13 +50,13 @@ class _BaseTestCase(TestController): @@
         description = u'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
-                                                _authentication_token=self.authentication_token()))
+                                                _session_csrf_secret_token=self.session_csrf_secret_token()))
         ## run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=repo_name))
         assert response.json == {u'result': True}
         self.checkSessionFlash(response,
                                'Created repository <a href="/%s">%s</a>'
                                % (repo_name, repo_name))
@@ @@ -88,21 +88,21 @@ class _BaseTestCase(TestController): @@
         description = u'description for newly created repo'
         response = self.app.post(url('repos'),
                                  fixture._get_repo_create_params(repo_private=False,
                                                                  repo_name=repo_name,
                                                                  repo_type=self.REPO_TYPE,
                                                                  repo_description=description,
-                                                                 _authentication_token=self.authentication_token()))
+                                                                 _session_csrf_secret_token=self.session_csrf_secret_token()))
         # try to create repo with swapped case
         swapped_repo_name = repo_name.swapcase()
         response = self.app.post(url('repos'),
                                  fixture._get_repo_create_params(repo_private=False,
                                                                  repo_name=swapped_repo_name,
                                                                  repo_type=self.REPO_TYPE,
                                                                  repo_description=description,
-                                                                 _authentication_token=self.authentication_token()))
+                                                                 _session_csrf_secret_token=self.session_csrf_secret_token()))
         response.mustcontain('already exists')
         RepoModel().delete(repo_name)
         Session().commit()
     def test_create_in_group(self):
@@ @@ -121,13 +121,13 @@ class _BaseTestCase(TestController): @@
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 repo_group=gr.group_id,
-                                                _authentication_token=self.authentication_token()))
+                                                _session_csrf_secret_token=self.session_csrf_secret_token()))
         ## run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=repo_name_full))
         assert response.json == {u'result': True}
         self.checkSessionFlash(response,
                                'Created repository <a href="/%s">%s</a>'
                                % (repo_name_full, repo_name_full))
@@ @@ -160,13 +160,13 @@ class _BaseTestCase(TestController): @@
         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()
+        session_csrf_secret_token = self.session_csrf_secret_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')
@@ @@ -198,13 +198,13 @@ class _BaseTestCase(TestController): @@
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 repo_group=gr.group_id,
-                                                _authentication_token=authentication_token))
+                                                _session_csrf_secret_token=session_csrf_secret_token))
         response.mustcontain('Invalid value')
         # user is allowed to create in this group
         repo_name = u'ingroup'
         repo_name_full = RepoGroup.url_sep().join([group_name_allowed, repo_name])
@@ @@ -212,13 +212,13 @@ class _BaseTestCase(TestController): @@
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 repo_group=gr_allowed.group_id,
-                                                _authentication_token=authentication_token))
+                                                _session_csrf_secret_token=session_csrf_secret_token))
         ## run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=repo_name_full))
         assert response.json == {u'result': True}
         self.checkSessionFlash(response,
                                'Created repository <a href="/%s">%s</a>'
@@ @@ -274,13 +274,13 @@ class _BaseTestCase(TestController): @@
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 repo_group=gr.group_id,
                                                 repo_copy_permissions=True,
-                                                _authentication_token=self.authentication_token()))
+                                                _session_csrf_secret_token=self.session_csrf_secret_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))
@@ @@ -326,38 +326,38 @@ class _BaseTestCase(TestController): @@
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 clone_uri='http://127.0.0.1/repo',
-                                                _authentication_token=self.authentication_token()))
+                                                _session_csrf_secret_token=self.session_csrf_secret_token()))
         response.mustcontain('Invalid repository URL')
     def test_create_remote_repo_wrong_clone_uri_hg_svn(self):
         self.log_user()
         repo_name = self.NEW_REPO
         description = u'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 clone_uri='svn+http://127.0.0.1/repo',
-                                                _authentication_token=self.authentication_token()))
+                                                _session_csrf_secret_token=self.session_csrf_secret_token()))
         response.mustcontain('Invalid repository URL')
     def test_delete(self):
         self.log_user()
         repo_name = u'vcs_test_new_to_delete_%s' % self.REPO_TYPE
         description = u'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_name=repo_name,
                                                 repo_description=description,
-                                                _authentication_token=self.authentication_token()))
+                                                _session_csrf_secret_token=self.session_csrf_secret_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
@@ @@ -376,13 +376,13 @@ class _BaseTestCase(TestController): @@
         try:
             vcs.get_repo(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name)))
         except vcs.exceptions.VCSError:
             pytest.fail('no repo %s in filesystem' % repo_name)
         response = self.app.post(url('delete_repo', repo_name=repo_name),
-            params={'_authentication_token': self.authentication_token()})
+            params={'_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'Deleted repository %s' % (repo_name))
         response.follow()
         # check if repo was deleted from db
@@ @@ -402,13 +402,13 @@ class _BaseTestCase(TestController): @@
         description_unicode = safe_unicode(description)
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
-                                                _authentication_token=self.authentication_token()))
+                                                _session_csrf_secret_token=self.session_csrf_secret_token()))
         ## run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=repo_name))
         assert response.json == {u'result': True}
         self.checkSessionFlash(response,
                                u'Created repository <a href="/%s">%s</a>'
                                % (urllib.quote(repo_name), repo_name_unicode))
@@ @@ -428,13 +428,13 @@ class _BaseTestCase(TestController): @@
         try:
             vcs.get_repo(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name_unicode)))
         except vcs.exceptions.VCSError:
             pytest.fail('no repo %s in filesystem' % repo_name)
         response = self.app.post(url('delete_repo', repo_name=repo_name),
-            params={'_authentication_token': self.authentication_token()})
+            params={'_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'Deleted repository %s' % (repo_name_unicode))
         response.follow()
         # check if repo was deleted from db
         deleted_repo = Session().query(Repository) \
             .filter(Repository.repo_name == repo_name_unicode).scalar()
@@ @@ -446,13 +446,13 @@ class _BaseTestCase(TestController): @@
     def test_delete_repo_with_group(self):
         # TODO:
         pass
     def test_delete_browser_fakeout(self):
         response = self.app.post(url('delete_repo', repo_name=self.REPO),
-                                 params=dict(_authentication_token=self.authentication_token()))
+                                 params=dict(_session_csrf_secret_token=self.session_csrf_secret_token()))
     def test_show(self):
         self.log_user()
         response = self.app.get(url('summary_home', repo_name=self.REPO))
     def test_edit(self):
@@ @@ -468,13 +468,13 @@ class _BaseTestCase(TestController): @@
         response = self.app.post(url('update_repo', repo_name=self.REPO),
                         fixture._get_repo_create_params(repo_private=1,
                                                 repo_name=self.REPO,
                                                 repo_type=self.REPO_TYPE,
                                                 owner=TEST_USER_ADMIN_LOGIN,
-                                                _authentication_token=self.authentication_token()))
+                                                _session_csrf_secret_token=self.session_csrf_secret_token()))
         self.checkSessionFlash(response,
                                msg='Repository %s updated successfully' % (self.REPO))
         assert Repository.get_by_repo_name(self.REPO).private == True
         # now the repo default permission should be None
         perm = _get_permission_for_user(user='default', repo=self.REPO)
@@ @@ -483,13 +483,13 @@ class _BaseTestCase(TestController): @@
         response = self.app.post(url('update_repo', repo_name=self.REPO),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=self.REPO,
                                                 repo_type=self.REPO_TYPE,
                                                 owner=TEST_USER_ADMIN_LOGIN,
-                                                _authentication_token=self.authentication_token()))
+                                                _session_csrf_secret_token=self.session_csrf_secret_token()))
         self.checkSessionFlash(response,
                                msg='Repository %s updated successfully' % (self.REPO))
         assert Repository.get_by_repo_name(self.REPO).private == False
         # we turn off private now the repo default permission should stay None
         perm = _get_permission_for_user(user='default', repo=self.REPO)
@@ @@ -511,13 +511,13 @@ class _BaseTestCase(TestController): @@
         self.log_user()
         other_repo = u'other_%s' % self.REPO_TYPE
         fixture.create_repo(other_repo, repo_type=self.REPO_TYPE)
         repo = Repository.get_by_repo_name(self.REPO)
         repo2 = Repository.get_by_repo_name(other_repo)
         response = self.app.post(url('edit_repo_advanced_fork', repo_name=self.REPO),
-                                params=dict(id_fork_of=repo2.repo_id, _authentication_token=self.authentication_token()))
+                                params=dict(id_fork_of=repo2.repo_id, _session_csrf_secret_token=self.session_csrf_secret_token()))
         repo = Repository.get_by_repo_name(self.REPO)
         repo2 = Repository.get_by_repo_name(other_repo)
         self.checkSessionFlash(response,
             'Marked repository %s as fork of %s' % (repo.repo_name, repo2.repo_name))
         assert repo.fork == repo2
@@ @@ -532,35 +532,35 @@ class _BaseTestCase(TestController): @@
     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.post(url('edit_repo_advanced_fork', repo_name=self.REPO),
-                                params=dict(id_fork_of=repo2.repo_id, _authentication_token=self.authentication_token()))
+                                params=dict(id_fork_of=repo2.repo_id, _session_csrf_secret_token=self.session_csrf_secret_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.post(url('edit_repo_advanced_fork', repo_name=self.REPO),
-                                params=dict(id_fork_of=None, _authentication_token=self.authentication_token()))
+                                params=dict(id_fork_of=None, _session_csrf_secret_token=self.session_csrf_secret_token()))
         repo = Repository.get_by_repo_name(self.REPO)
         repo2 = Repository.get_by_repo_name(self.OTHER_TYPE_REPO)
         self.checkSessionFlash(response,
                                'Marked repository %s as fork of %s'
                                % (repo.repo_name, "Nothing"))
         assert repo.fork is None
     def test_set_fork_of_same_repo(self):
         self.log_user()
         repo = Repository.get_by_repo_name(self.REPO)
         response = self.app.post(url('edit_repo_advanced_fork', repo_name=self.REPO),
-                                params=dict(id_fork_of=repo.repo_id, _authentication_token=self.authentication_token()))
+                                params=dict(id_fork_of=repo.repo_id, _session_csrf_secret_token=self.session_csrf_secret_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
@@ @@ -585,13 +585,13 @@ class _BaseTestCase(TestController): @@
         description = 'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
-                                                _authentication_token=self.authentication_token()))
+                                                _session_csrf_secret_token=self.session_csrf_secret_token()))
         response.mustcontain('<span class="error-message">Invalid value</span>')
         RepoModel().delete(repo_name)
         Session().commit()
@@ @@ -603,13 +603,13 @@ class _BaseTestCase(TestController): @@
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
-                                                _authentication_token=self.authentication_token()))
+                                                _session_csrf_secret_token=self.session_csrf_secret_token()))
         self.checkSessionFlash(response,
                                'Error creating repository %s' % repo_name)
         # repo must not be in db
         repo = Repository.get_by_repo_name(repo_name)
         assert repo is None

kallithea/tests/functional/test_admin_settings.py

➞

Show inline comments

@@ @@ -35,69 +35,69 @@ class TestAdminSettingsController(TestCo @@
     def test_create_custom_hook(self):
         self.log_user()
         response = self.app.post(url('admin_settings_hooks'),
                                 params=dict(new_hook_ui_key='test_hooks_1',
                                             new_hook_ui_value='cd %s' % TESTS_TMP_PATH,
-                                            _authentication_token=self.authentication_token()))
+                                            _session_csrf_secret_token=self.session_csrf_secret_token()))
         self.checkSessionFlash(response, 'Added new hook')
         response = response.follow()
         response.mustcontain('test_hooks_1')
         response.mustcontain('cd %s' % TESTS_TMP_PATH)
     def test_edit_custom_hook(self):
         self.log_user()
         response = self.app.post(url('admin_settings_hooks'),
                                 params=dict(hook_ui_key='test_hooks_1',
                                             hook_ui_value='old_value_of_hook_1',
                                             hook_ui_value_new='new_value_of_hook_1',
-                                            _authentication_token=self.authentication_token()))
+                                            _session_csrf_secret_token=self.session_csrf_secret_token()))
         response = response.follow()
         response.mustcontain('test_hooks_1')
         response.mustcontain('new_value_of_hook_1')
     def test_add_existing_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='attempted_new_value',
-                                            _authentication_token=self.authentication_token()))
+                                            _session_csrf_secret_token=self.session_csrf_secret_token()))
         self.checkSessionFlash(response, 'Hook already exists')
         response = response.follow()
         response.mustcontain('test_hooks_1')
         response.mustcontain('new_value_of_hook_1')
     def test_create_custom_hook_delete(self):
         self.log_user()
         response = self.app.post(url('admin_settings_hooks'),
                                 params=dict(new_hook_ui_key='test_hooks_2',
                                             new_hook_ui_value='cd %s2' % TESTS_TMP_PATH,
-                                            _authentication_token=self.authentication_token()))
+                                            _session_csrf_secret_token=self.session_csrf_secret_token()))
         self.checkSessionFlash(response, 'Added new hook')
         response = response.follow()
         response.mustcontain('test_hooks_2')
         response.mustcontain('cd %s2' % TESTS_TMP_PATH)
         hook_id = Ui.get_by_key('hooks', 'test_hooks_2').ui_id
         ## delete
         self.app.post(url('admin_settings_hooks'),
-                        params=dict(hook_id=hook_id, _authentication_token=self.authentication_token()))
+                        params=dict(hook_id=hook_id, _session_csrf_secret_token=self.session_csrf_secret_token()))
         response = self.app.get(url('admin_settings_hooks'))
         response.mustcontain(no=['test_hooks_2'])
         response.mustcontain(no=['cd %s2' % TESTS_TMP_PATH])
     def test_add_existing_builtin_hook(self):
         self.log_user()
         response = self.app.post(url('admin_settings_hooks'),
                                 params=dict(new_hook_ui_key='changegroup.update',
                                             new_hook_ui_value='attempted_new_value',
-                                            _authentication_token=self.authentication_token()))
+                                            _session_csrf_secret_token=self.session_csrf_secret_token()))
         self.checkSessionFlash(response, 'Builtin hooks are read-only')
         response = response.follow()
         response.mustcontain('changegroup.update')
         response.mustcontain('hg update &gt;&amp;2')
@@ @@ -117,13 +117,13 @@ class TestAdminSettingsController(TestCo @@
         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(),
+                                 _session_csrf_secret_token=self.session_csrf_secret_token(),
                                  ))
         self.checkSessionFlash(response, 'Updated application settings')
         assert Setting.get_app_settings()['ga_code'] == new_ga_code
@@ @@ -138,13 +138,13 @@ class TestAdminSettingsController(TestCo @@
         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(),
+                                 _session_csrf_secret_token=self.session_csrf_secret_token(),
                                  ))
         self.checkSessionFlash(response, 'Updated application settings')
         assert Setting.get_app_settings()['ga_code'] == new_ga_code
         response = response.follow()
@@ @@ -158,13 +158,13 @@ class TestAdminSettingsController(TestCo @@
         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(),
+                                 _session_csrf_secret_token=self.session_csrf_secret_token(),
                                  ))
         self.checkSessionFlash(response, 'Updated application settings')
         assert Setting.get_app_settings()['captcha_private_key'] == '1234567890'
         response = self.app.get(url('register'))
@@ @@ -178,13 +178,13 @@ class TestAdminSettingsController(TestCo @@
         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(),
+                                 _session_csrf_secret_token=self.session_csrf_secret_token(),
                                  ))
         self.checkSessionFlash(response, 'Updated application settings')
         assert Setting.get_app_settings()['captcha_private_key'] == ''
         response = self.app.get(url('register'))
@@ @@ -200,13 +200,13 @@ class TestAdminSettingsController(TestCo @@
             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(),
+                                 _session_csrf_secret_token=self.session_csrf_secret_token(),
                                 ))
             self.checkSessionFlash(response, 'Updated application settings')
             assert Setting.get_app_settings()['title'] == new_title.decode('utf-8')
             response = response.follow()

kallithea/tests/functional/test_admin_user_groups.py

➞

Show inline comments

@@ @@ -17,13 +17,13 @@ class TestAdminUsersGroupsController(Tes @@
         self.log_user()
         users_group_name = TEST_USER_GROUP
         response = self.app.post(url('users_groups'),
                                  {'users_group_name': users_group_name,
                                   'user_group_description': u'DESC',
                                   'active': True,
-                                  '_authentication_token': self.authentication_token()})
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})
         response.follow()
         self.checkSessionFlash(response,
                                'Created user group <a href="/_admin/user_groups/')
         self.checkSessionFlash(response,
                                '/edit">%s</a>' % TEST_USER_GROUP)
@@ @@ -33,32 +33,32 @@ class TestAdminUsersGroupsController(Tes @@
     def test_update(self):
         response = self.app.post(url('update_users_group', id=1), status=403)
     def test_update_browser_fakeout(self):
         response = self.app.post(url('update_users_group', id=1),
-                                 params=dict(_authentication_token=self.authentication_token()))
+                                 params=dict(_session_csrf_secret_token=self.session_csrf_secret_token()))
     def test_delete(self):
         self.log_user()
         users_group_name = TEST_USER_GROUP + 'another'
         response = self.app.post(url('users_groups'),
                                  {'users_group_name': users_group_name,
                                   'user_group_description': u'DESC',
                                   'active': True,
-                                  '_authentication_token': self.authentication_token()})
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})
         response.follow()
         self.checkSessionFlash(response,
                                'Created user group ')
         gr = Session().query(UserGroup) \
             .filter(UserGroup.users_group_name == users_group_name).one()
         response = self.app.post(url('delete_users_group', id=gr.users_group_id),
-            params={'_authentication_token': self.authentication_token()})
+            params={'_session_csrf_secret_token': self.session_csrf_secret_token()})
         gr = Session().query(UserGroup) \
             .filter(UserGroup.users_group_name == users_group_name).scalar()
         assert gr is None
@@ @@ -66,23 +66,23 @@ class TestAdminUsersGroupsController(Tes @@
         self.log_user()
         users_group_name = TEST_USER_GROUP + 'another2'
         response = self.app.post(url('users_groups'),
                                  {'users_group_name': users_group_name,
                                   'user_group_description': u'DESC',
                                   'active': True,
-                                  '_authentication_token': self.authentication_token()})
+                                  '_session_csrf_secret_token': self.session_csrf_secret_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.post(url('edit_user_group_default_perms_update',
                                      id=ug.users_group_id),
                                  {'create_repo_perm': True,
-                                  '_authentication_token': self.authentication_token()})
+                                  '_session_csrf_secret_token': self.session_csrf_secret_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
@@ @@ -94,13 +94,13 @@ class TestAdminUsersGroupsController(Tes @@
                     [ug.users_group_id, p2.permission_id],
                     [ug.users_group_id, p3.permission_id]])
         ## DISABLE REPO CREATE ON A GROUP
         response = self.app.post(
             url('edit_user_group_default_perms_update', id=ug.users_group_id),
-            params={'_authentication_token': self.authentication_token()})
+            params={'_session_csrf_secret_token': self.session_csrf_secret_token()})
         response.follow()
         ug = UserGroup.get_by_group_name(users_group_name)
         p = Permission.get_by_key('hg.create.none')
         p2 = Permission.get_by_key('hg.usergroup.create.false')
         p3 = Permission.get_by_key('hg.fork.none')
@@ @@ -115,13 +115,13 @@ 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.post(url('delete_users_group', id=ug.users_group_id),
-            params={'_authentication_token': self.authentication_token()})
+            params={'_session_csrf_secret_token': self.session_csrf_secret_token()})
         response = response.follow()
         gr = Session().query(UserGroup) \
             .filter(UserGroup.users_group_name == users_group_name).scalar()
         assert gr is None
         p = Permission.get_by_key('hg.create.repository')
@@ @@ -135,22 +135,22 @@ class TestAdminUsersGroupsController(Tes @@
         self.log_user()
         users_group_name = TEST_USER_GROUP + 'another2'
         response = self.app.post(url('users_groups'),
                                  {'users_group_name': users_group_name,
                                   'user_group_description': u'DESC',
                                   'active': True,
-                                  '_authentication_token': self.authentication_token()})
+                                  '_session_csrf_secret_token': self.session_csrf_secret_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.post(url('edit_user_group_default_perms_update',
                                      id=ug.users_group_id),
-                                 {'fork_repo_perm': True, '_authentication_token': self.authentication_token()})
+                                 {'fork_repo_perm': True, '_session_csrf_secret_token': self.session_csrf_secret_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')
@@ @@ -162,13 +162,13 @@ class TestAdminUsersGroupsController(Tes @@
         assert sorted([[x.users_group_id, x.permission_id, ] for x in perms]) == sorted([[ug.users_group_id, p.permission_id],
                     [ug.users_group_id, p2.permission_id],
                     [ug.users_group_id, p3.permission_id]])
         ## DISABLE REPO CREATE ON A GROUP
         response = self.app.post(url('edit_user_group_default_perms_update', id=ug.users_group_id),
-            params={'_authentication_token': self.authentication_token()})
+            params={'_session_csrf_secret_token': self.session_csrf_secret_token()})
         response.follow()
         ug = UserGroup.get_by_group_name(users_group_name)
         p = Permission.get_by_key('hg.create.none')
         p2 = Permission.get_by_key('hg.usergroup.create.false')
         p3 = Permission.get_by_key('hg.fork.none')
@@ @@ -182,13 +182,13 @@ 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.post(url('delete_users_group', id=ug.users_group_id),
-            params={'_authentication_token': self.authentication_token()})
+            params={'_session_csrf_secret_token': self.session_csrf_secret_token()})
         response = response.follow()
         gr = Session().query(UserGroup) \
                            .filter(UserGroup.users_group_name ==
                                    users_group_name).scalar()
         assert gr is None
@@ @@ -198,7 +198,7 @@ class TestAdminUsersGroupsController(Tes @@
         perms = [[x.users_group_id,
                   x.permission_id, ] for x in perms]
         assert perms == []
     def test_delete_browser_fakeout(self):
         response = self.app.post(url('delete_users_group', id=1),
-                                 params=dict(_authentication_token=self.authentication_token()))
+                                 params=dict(_session_csrf_secret_token=self.session_csrf_secret_token()))

kallithea/tests/functional/test_admin_users.py

➞

Show inline comments

@@ @@ -73,13 +73,13 @@ class TestAdminUsersController(TestContr @@
              'firstname': name,
              'active': True,
              'lastname': lastname,
              'extern_name': 'internal',
              'extern_type': 'internal',
              'email': email,
-             '_authentication_token': self.authentication_token()})
+             '_session_csrf_secret_token': self.session_csrf_secret_token()})
         # 302 Found
         # The resource was found at http://localhost/_admin/users/5/edit; you should be redirected automatically.
         self.checkSessionFlash(response, '''Created user %s''' % username)
         response = response.follow()
@@ @@ -106,13 +106,13 @@ class TestAdminUsersController(TestContr @@
             {'username': username,
              'password': password,
              'name': name,
              'active': False,
              'lastname': lastname,
              'email': email,
-             '_authentication_token': self.authentication_token()})
+             '_session_csrf_secret_token': self.session_csrf_secret_token()})
         with test_context(self.app):
             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>""")
@@ @@ -163,16 +163,16 @@ class TestAdminUsersController(TestContr @@
         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()})
+        params.update({'_session_csrf_secret_token': self.session_csrf_secret_token()})
         response = self.app.post(url('update_user', id=usr.user_id), params)
         self.checkSessionFlash(response, 'User updated successfully')
-        params.pop('_authentication_token')
+        params.pop('_session_csrf_secret_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': ''})
@@ @@ -184,13 +184,13 @@ class TestAdminUsersController(TestContr @@
         fixture.create_user(name=username)
         new_user = Session().query(User) \
             .filter(User.username == username).one()
         response = self.app.post(url('delete_user', id=new_user.user_id),
-            params={'_authentication_token': self.authentication_token()})
+            params={'_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'Successfully deleted user')
     def test_delete_repo_err(self):
         self.log_user()
         username = 'repoerr'
@@ @@ -199,50 +199,50 @@ class TestAdminUsersController(TestContr @@
         fixture.create_user(name=username)
         fixture.create_repo(name=reponame, cur_user=username)
         new_user = Session().query(User) \
             .filter(User.username == username).one()
         response = self.app.post(url('delete_user', id=new_user.user_id),
-            params={'_authentication_token': self.authentication_token()})
+            params={'_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'User "%s" still '
                                'owns 1 repositories and cannot be removed. '
                                'Switch owners or remove those repositories: '
                                '%s' % (username, reponame))
         response = self.app.post(url('delete_repo', repo_name=reponame),
-            params={'_authentication_token': self.authentication_token()})
+            params={'_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'Deleted repository %s' % reponame)
         response = self.app.post(url('delete_user', id=new_user.user_id),
-            params={'_authentication_token': self.authentication_token()})
+            params={'_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'Successfully deleted user')
     def test_delete_repo_group_err(self, user_and_repo_group_fail):
         new_user, repo_group = user_and_repo_group_fail
         username = new_user.username
         groupname = repo_group.group_name
         self.log_user()
         response = self.app.post(url('delete_user', id=new_user.user_id),
-            params={'_authentication_token': self.authentication_token()})
+            params={'_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'User "%s" still '
                                'owns 1 repository groups and cannot be removed. '
                                'Switch owners or remove those repository groups: '
                                '%s' % (username, groupname))
         # Relevant _if_ the user deletion succeeded to make sure we can render groups without owner
         # rg = RepoGroup.get_by_group_name(group_name=groupname)
         # response = self.app.get(url('repos_groups', id=rg.group_id))
         response = self.app.post(url('delete_repo_group', group_name=groupname),
-            params={'_authentication_token': self.authentication_token()})
+            params={'_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'Removed repository group %s' % groupname)
         response = self.app.post(url('delete_user', id=new_user.user_id),
-            params={'_authentication_token': self.authentication_token()})
+            params={'_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'Successfully deleted user')
     def test_delete_user_group_err(self):
         self.log_user()
         username = 'usergrouperr'
         groupname = u'usergroup_fail'
@@ @@ -250,26 +250,26 @@ class TestAdminUsersController(TestContr @@
         fixture.create_user(name=username)
         ug = fixture.create_user_group(name=groupname, cur_user=username)
         new_user = Session().query(User) \
             .filter(User.username == username).one()
         response = self.app.post(url('delete_user', id=new_user.user_id),
-            params={'_authentication_token': self.authentication_token()})
+            params={'_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'User "%s" still '
                                'owns 1 user groups and cannot be removed. '
                                'Switch owners or remove those user groups: '
                                '%s' % (username, groupname))
         # TODO: why do this fail?
         #response = self.app.delete(url('delete_users_group', id=groupname))
         #self.checkSessionFlash(response, 'Removed user group %s' % groupname)
         fixture.destroy_user_group(ug.users_group_id)
         response = self.app.post(url('delete_user', id=new_user.user_id),
-            params={'_authentication_token': self.authentication_token()})
+            params={'_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'Successfully deleted user')
     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))
@@ @@ -289,13 +289,13 @@ class TestAdminUsersController(TestContr @@
             # User should have None permission on creation repository
             assert UserModel().has_perm(user, perm_none) == False
             assert UserModel().has_perm(user, perm_create) == False
             response = self.app.post(url('edit_user_perms_update', id=uid),
                                      params=dict(create_repo_perm=True,
-                                                 _authentication_token=self.authentication_token()))
+                                                 _session_csrf_secret_token=self.session_csrf_secret_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
             assert UserModel().has_perm(uid, perm_none) == False
@@ @@ -318,13 +318,13 @@ class TestAdminUsersController(TestContr @@
         try:
             # User should have None permission on creation repository
             assert UserModel().has_perm(user, perm_none) == False
             assert UserModel().has_perm(user, perm_create) == False
             response = self.app.post(url('edit_user_perms_update', id=uid),
-                                     params=dict(_authentication_token=self.authentication_token()))
+                                     params=dict(_session_csrf_secret_token=self.session_csrf_secret_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
             assert UserModel().has_perm(uid, perm_none) == True
@@ @@ -348,13 +348,13 @@ class TestAdminUsersController(TestContr @@
             # User should have None permission on creation repository
             assert UserModel().has_perm(user, perm_none) == False
             assert UserModel().has_perm(user, perm_fork) == False
             response = self.app.post(url('edit_user_perms_update', id=uid),
                                      params=dict(create_repo_perm=True,
-                                                 _authentication_token=self.authentication_token()))
+                                                 _session_csrf_secret_token=self.session_csrf_secret_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
             assert UserModel().has_perm(uid, perm_none) == False
@@ @@ -377,13 +377,13 @@ class TestAdminUsersController(TestContr @@
         try:
             # User should have None permission on creation repository
             assert UserModel().has_perm(user, perm_none) == False
             assert UserModel().has_perm(user, perm_fork) == False
             response = self.app.post(url('edit_user_perms_update', id=uid),
-                                     params=dict(_authentication_token=self.authentication_token()))
+                                     params=dict(_session_csrf_secret_token=self.session_csrf_secret_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
             assert UserModel().has_perm(uid, perm_none) == True
@@ @@ -409,13 +409,13 @@ class TestAdminUsersController(TestContr @@
     def test_add_ip(self, test_name, ip, ip_range, failure, auto_clear_ip_permissions):
         self.log_user()
         user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
         user_id = user.user_id
         response = self.app.post(url('edit_user_ips_update', id=user_id),
-                                 params=dict(new_ip=ip, _authentication_token=self.authentication_token()))
+                                 params=dict(new_ip=ip, _session_csrf_secret_token=self.session_csrf_secret_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])
@@ @@ -438,13 +438,13 @@ class TestAdminUsersController(TestContr @@
         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_delete', id=user_id),
-                      params=dict(del_ip_id=new_ip_id, _authentication_token=self.authentication_token()))
+                      params=dict(del_ip_id=new_ip_id, _session_csrf_secret_token=self.session_csrf_secret_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])
@@ @@ -464,13 +464,13 @@ class TestAdminUsersController(TestContr @@
     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_update', id=user_id),
-                 {'description': desc, 'lifetime': lifetime, '_authentication_token': self.authentication_token()})
+                 {'description': desc, 'lifetime': lifetime, '_session_csrf_secret_token': self.session_csrf_secret_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)
@@ @@ -482,22 +482,22 @@ class TestAdminUsersController(TestContr @@
     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_update', id=user_id),
-                {'description': 'desc', 'lifetime': -1, '_authentication_token': self.authentication_token()})
+                {'description': 'desc', 'lifetime': -1, '_session_csrf_secret_token': self.session_csrf_secret_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()
         assert 1 == len(keys)
         response = self.app.post(url('edit_user_api_keys_delete', id=user_id),
-                 {'del_api_key': keys[0].api_key, '_authentication_token': self.authentication_token()})
+                 {'del_api_key': keys[0].api_key, '_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'API key successfully deleted')
         keys = UserApiKeys.query().filter(UserApiKeys.user_id == user_id).all()
         assert 0 == len(keys)
     def test_reset_main_api_key(self):
         self.log_user()
@@ @@ -506,13 +506,13 @@ class TestAdminUsersController(TestContr @@
         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_delete', id=user_id),
-                 {'del_api_key_builtin': api_key, '_authentication_token': self.authentication_token()})
+                 {'del_api_key_builtin': api_key, '_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'API key successfully reset')
         response = response.follow()
         response.mustcontain(no=[api_key])
     def test_add_ssh_key(self):
         description = u'something'
@@ @@ -523,13 +523,13 @@ class TestAdminUsersController(TestContr @@
         user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
         user_id = user.user_id
         response = self.app.post(url('edit_user_ssh_keys', id=user_id),
                                  {'description': description,
                                   'public_key': public_key,
-                                  '_authentication_token': self.authentication_token()})
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'SSH key %s successfully added' % fingerprint)
         response = response.follow()
         response.mustcontain(fingerprint)
         ssh_key = UserSshKeys.query().filter(UserSshKeys.user_id == user_id).one()
         assert ssh_key.fingerprint == fingerprint
@@ @@ -546,21 +546,21 @@ class TestAdminUsersController(TestContr @@
         user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
         user_id = user.user_id
         response = self.app.post(url('edit_user_ssh_keys', id=user_id),
                                  {'description': description,
                                   'public_key': public_key,
-                                  '_authentication_token': self.authentication_token()})
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'SSH key %s successfully added' % fingerprint)
         response.follow()
         ssh_key = UserSshKeys.query().filter(UserSshKeys.user_id == user_id).one()
         assert ssh_key.description == u'me@localhost'
         response = self.app.post(url('edit_user_ssh_keys_delete', id=user_id),
                                  {'del_public_key': ssh_key.public_key,
-                                  '_authentication_token': self.authentication_token()})
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'SSH key successfully deleted')
         keys = UserSshKeys.query().all()
         assert 0 == len(keys)
 class TestAdminUsersController_unittest(TestController):
@@ @@ -603,49 +603,49 @@ class TestAdminUsersControllerForDefault @@
         response = self.app.get(url('edit_user_api_keys', id=user.user_id), status=404)
     def test_add_api_keys_default_user(self):
         self.log_user()
         user = User.get_default_user()
         response = self.app.post(url('edit_user_api_keys_update', id=user.user_id),
-                 {'_authentication_token': self.authentication_token()}, status=404)
+                 {'_session_csrf_secret_token': self.session_csrf_secret_token()}, status=404)
     def test_delete_api_keys_default_user(self):
         self.log_user()
         user = User.get_default_user()
         response = self.app.post(url('edit_user_api_keys_delete', id=user.user_id),
-                 {'_authentication_token': self.authentication_token()}, status=404)
+                 {'_session_csrf_secret_token': self.session_csrf_secret_token()}, status=404)
     # Permissions
     def test_edit_perms_default_user(self):
         self.log_user()
         user = User.get_default_user()
         response = self.app.get(url('edit_user_perms', id=user.user_id), status=404)
     def test_update_perms_default_user(self):
         self.log_user()
         user = User.get_default_user()
         response = self.app.post(url('edit_user_perms_update', id=user.user_id),
-                 {'_authentication_token': self.authentication_token()}, status=404)
+                 {'_session_csrf_secret_token': self.session_csrf_secret_token()}, status=404)
     # Emails
     def test_edit_emails_default_user(self):
         self.log_user()
         user = User.get_default_user()
         response = self.app.get(url('edit_user_emails', id=user.user_id), status=404)
     def test_add_emails_default_user(self):
         self.log_user()
         user = User.get_default_user()
         response = self.app.post(url('edit_user_emails_update', id=user.user_id),
-                 {'_authentication_token': self.authentication_token()}, status=404)
+                 {'_session_csrf_secret_token': self.session_csrf_secret_token()}, status=404)
     def test_delete_emails_default_user(self):
         self.log_user()
         user = User.get_default_user()
         response = self.app.post(url('edit_user_emails_delete', id=user.user_id),
-                 {'_authentication_token': self.authentication_token()}, status=404)
+                 {'_session_csrf_secret_token': self.session_csrf_secret_token()}, status=404)
     # IP addresses
     # Add/delete of IP addresses for the default user is used to maintain
     # the global IP whitelist and thus allowed. Only 'edit' is forbidden.
     def test_edit_ip_default_user(self):
         self.log_user()

kallithea/tests/functional/test_changeset_pullrequests_comments.py

➞

Show inline comments

@@ @@ -15,13 +15,13 @@ class TestChangeSetCommentsController(Te @@
     def test_create(self):
         self.log_user()
         rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
         text = u'general comment on changeset'
-        params = {'text': text, '_authentication_token': self.authentication_token()}
+        params = {'text': text, '_session_csrf_secret_token': self.session_csrf_secret_token()}
         response = self.app.post(url(controller='changeset', action='comment',
                                      repo_name=HG_REPO, revision=rev),
                                      params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         # Test response...
         assert response.status == '200 OK'
@@ @@ -40,13 +40,13 @@ class TestChangeSetCommentsController(Te @@
         self.log_user()
         rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
         text = u'inline comment on changeset'
         f_path = 'vcs/web/simplevcs/views/repository.py'
         line = 'n1'
-        params = {'text': text, 'f_path': f_path, 'line': line, '_authentication_token': self.authentication_token()}
+        params = {'text': text, 'f_path': f_path, 'line': line, '_session_csrf_secret_token': self.session_csrf_secret_token()}
         response = self.app.post(url(controller='changeset', action='comment',
                                      repo_name=HG_REPO, revision=rev),
                                      params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         # Test response...
         assert response.status == '200 OK'
@@ @@ -69,13 +69,13 @@ class TestChangeSetCommentsController(Te @@
     def test_create_with_mention(self):
         self.log_user()
         rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
         text = u'@%s check CommentOnRevision' % TEST_USER_REGULAR_LOGIN
-        params = {'text': text, '_authentication_token': self.authentication_token()}
+        params = {'text': text, '_session_csrf_secret_token': self.session_csrf_secret_token()}
         response = self.app.post(url(controller='changeset', action='comment',
                                      repo_name=HG_REPO, revision=rev),
                                      params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         # Test response...
         assert response.status == '200 OK'
@@ @@ -93,13 +93,13 @@ class TestChangeSetCommentsController(Te @@
     def test_create_status_change(self):
         self.log_user()
         rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
         text = u'general comment on changeset'
         params = {'text': text, 'changeset_status': 'rejected',
-                '_authentication_token': self.authentication_token()}
+                '_session_csrf_secret_token': self.session_csrf_secret_token()}
         response = self.app.post(url(controller='changeset', action='comment',
                                      repo_name=HG_REPO, revision=rev),
                                      params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         # Test response...
         assert response.status == '200 OK'
@@ @@ -120,25 +120,25 @@ class TestChangeSetCommentsController(Te @@
     def test_delete(self):
         self.log_user()
         rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
         text = u'general comment on changeset to be deleted'
-        params = {'text': text, '_authentication_token': self.authentication_token()}
+        params = {'text': text, '_session_csrf_secret_token': self.session_csrf_secret_token()}
         response = self.app.post(url(controller='changeset', action='comment',
                                      repo_name=HG_REPO, revision=rev),
                                      params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         comments = ChangesetComment.query().all()
         assert len(comments) == 1
         comment_id = comments[0].comment_id
         self.app.post(url("changeset_comment_delete",
                                     repo_name=HG_REPO,
                                     comment_id=comment_id),
-            params={'_authentication_token': self.authentication_token()})
+            params={'_session_csrf_secret_token': self.session_csrf_secret_token()})
         comments = ChangesetComment.query().all()
         assert len(comments) == 0
         response = self.app.get(url(controller='changeset', action='index',
                                 repo_name=HG_REPO, revision=rev))
@@ @@ -162,24 +162,24 @@ class TestPullrequestsCommentsController @@
                                  {'org_repo': HG_REPO,
                                   'org_ref': 'branch:stable:4f7e2131323e0749a740c0a56ab68ae9269c562a',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
-                                  '_authentication_token': self.authentication_token(),
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         pr_id = int(re.search('/pull-request/(\d+)/', response.location).group(1))
         return pr_id
     def test_create(self):
         self.log_user()
         pr_id = self._create_pr()
         text = u'general comment on pullrequest'
-        params = {'text': text, '_authentication_token': self.authentication_token()}
+        params = {'text': text, '_session_csrf_secret_token': self.session_csrf_secret_token()}
         response = self.app.post(url(controller='pullrequests', action='comment',
                                      repo_name=HG_REPO, pull_request_id=pr_id),
                                      params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         # Test response...
         assert response.status == '200 OK'
@@ @@ -201,13 +201,13 @@ class TestPullrequestsCommentsController @@
         self.log_user()
         pr_id = self._create_pr()
         text = u'inline comment on changeset'
         f_path = 'vcs/web/simplevcs/views/repository.py'
         line = 'n1'
-        params = {'text': text, 'f_path': f_path, 'line': line, '_authentication_token': self.authentication_token()}
+        params = {'text': text, 'f_path': f_path, 'line': line, '_session_csrf_secret_token': self.session_csrf_secret_token()}
         response = self.app.post(url(controller='pullrequests', action='comment',
                                      repo_name=HG_REPO, pull_request_id=pr_id),
                                      params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         # Test response...
         assert response.status == '200 OK'
@@ @@ -229,13 +229,13 @@ class TestPullrequestsCommentsController @@
     def test_create_with_mention(self):
         self.log_user()
         pr_id = self._create_pr()
         text = u'@%s check CommentOnRevision' % TEST_USER_REGULAR_LOGIN
-        params = {'text': text, '_authentication_token': self.authentication_token()}
+        params = {'text': text, '_session_csrf_secret_token': self.session_csrf_secret_token()}
         response = self.app.post(url(controller='pullrequests', action='comment',
                                      repo_name=HG_REPO, pull_request_id=pr_id),
                                      params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         # Test response...
         assert response.status == '200 OK'
@@ @@ -253,13 +253,13 @@ class TestPullrequestsCommentsController @@
     def test_create_status_change(self):
         self.log_user()
         pr_id = self._create_pr()
         text = u'general comment on pullrequest'
         params = {'text': text, 'changeset_status': 'rejected',
-                '_authentication_token': self.authentication_token()}
+                '_session_csrf_secret_token': self.session_csrf_secret_token()}
         response = self.app.post(url(controller='pullrequests', action='comment',
                                      repo_name=HG_REPO, pull_request_id=pr_id),
                                      params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         # Test response...
         assert response.status == '200 OK'
@@ @@ -283,25 +283,25 @@ class TestPullrequestsCommentsController @@
     def test_delete(self):
         self.log_user()
         pr_id = self._create_pr()
         text = u'general comment on changeset to be deleted'
-        params = {'text': text, '_authentication_token': self.authentication_token()}
+        params = {'text': text, '_session_csrf_secret_token': self.session_csrf_secret_token()}
         response = self.app.post(url(controller='pullrequests', action='comment',
                                      repo_name=HG_REPO, pull_request_id=pr_id),
                                      params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         comments = ChangesetComment.query().all()
         assert len(comments) == 2
         comment_id = comments[-1].comment_id
         self.app.post(url("pullrequest_comment_delete",
                                     repo_name=HG_REPO,
                                     comment_id=comment_id),
-            params={'_authentication_token': self.authentication_token()})
+            params={'_session_csrf_secret_token': self.session_csrf_secret_token()})
         comments = ChangesetComment.query().all()
         assert len(comments) == 1
         response = self.app.get(url(controller='pullrequests', action='show',
                                 repo_name=HG_REPO, pull_request_id=pr_id, extra=''))
@@ @@ -314,13 +314,13 @@ class TestPullrequestsCommentsController @@
     def test_close_pr(self):
         self.log_user()
         pr_id = self._create_pr()
         text = u'general comment on pullrequest'
         params = {'text': text, 'save_close': 'close',
-                '_authentication_token': self.authentication_token()}
+                '_session_csrf_secret_token': self.session_csrf_secret_token()}
         response = self.app.post(url(controller='pullrequests', action='comment',
                                      repo_name=HG_REPO, pull_request_id=pr_id),
                                      params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         # Test response...
         assert response.status == '200 OK'
@@ @@ -337,13 +337,13 @@ class TestPullrequestsCommentsController @@
     def test_delete_pr(self):
         self.log_user()
         pr_id = self._create_pr()
         text = u'general comment on pullrequest'
         params = {'text': text, 'save_delete': 'delete',
-                '_authentication_token': self.authentication_token()}
+                '_session_csrf_secret_token': self.session_csrf_secret_token()}
         response = self.app.post(url(controller='pullrequests', action='comment',
                                      repo_name=HG_REPO, pull_request_id=pr_id),
                                      params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         # Test response...
         assert response.status == '200 OK'
@@ @@ -357,21 +357,21 @@ class TestPullrequestsCommentsController @@
         self.log_user()
         pr_id = self._create_pr()
         # first close
         text = u'general comment on pullrequest'
         params = {'text': text, 'save_close': 'close',
-                '_authentication_token': self.authentication_token()}
+                '_session_csrf_secret_token': self.session_csrf_secret_token()}
         response = self.app.post(url(controller='pullrequests', action='comment',
                                      repo_name=HG_REPO, pull_request_id=pr_id),
                                      params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         assert response.status == '200 OK'
         # attempt delete, should fail
         params = {'text': text, 'save_delete': 'delete',
-                '_authentication_token': self.authentication_token()}
+                '_session_csrf_secret_token': self.session_csrf_secret_token()}
         response = self.app.post(url(controller='pullrequests', action='comment',
                                      repo_name=HG_REPO, pull_request_id=pr_id),
                                      params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'}, status=403)
         # verify that PR still exists, in closed state
         assert PullRequest.get(pr_id).status == PullRequest.STATUS_CLOSED

kallithea/tests/functional/test_files.py

➞

Show inline comments

@@ @@ -330,26 +330,26 @@ class TestFilesController(TestController @@
         self.log_user()
         response = self.app.post(url('files_add_home',
                                       repo_name=HG_REPO,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': '',
-                                    '_authentication_token': self.authentication_token(),
+                                    '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         self.checkSessionFlash(response, 'No content')
     def test_add_file_into_hg_missing_filename(self):
         self.log_user()
         response = self.app.post(url('files_add_home',
                                       repo_name=HG_REPO,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "foo",
-                                    '_authentication_token': self.authentication_token(),
+                                    '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         self.checkSessionFlash(response, 'No filename')
     @parametrize('location,filename', [
@@ @@ -363,13 +363,13 @@ class TestFilesController(TestController @@
                                       repo_name=HG_REPO,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "foo",
                                     'filename': filename,
                                     'location': location,
-                                    '_authentication_token': self.authentication_token(),
+                                    '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         self.checkSessionFlash(response, 'Location must be relative path and must not contain .. in path')
     @parametrize('cnt,location,filename', [
@@ @@ -384,13 +384,13 @@ class TestFilesController(TestController @@
                                       repo_name=repo.repo_name,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "foo",
                                     'filename': filename,
                                     'location': location,
-                                    '_authentication_token': self.authentication_token(),
+                                    '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
         finally:
@@ @@ -407,25 +407,25 @@ class TestFilesController(TestController @@
         self.log_user()
         response = self.app.post(url('files_add_home',
                                       repo_name=GIT_REPO,
                                       revision='tip', f_path='/'),
                                  params={
                                      'content': '',
-                                     '_authentication_token': self.authentication_token(),
+                                     '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         self.checkSessionFlash(response, 'No content')
     def test_add_file_into_git_missing_filename(self):
         self.log_user()
         response = self.app.post(url('files_add_home',
                                       repo_name=GIT_REPO,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "foo",
-                                    '_authentication_token': self.authentication_token(),
+                                    '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         self.checkSessionFlash(response, 'No filename')
     @parametrize('location,filename', [
@@ @@ -439,13 +439,13 @@ class TestFilesController(TestController @@
                                       repo_name=GIT_REPO,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "foo",
                                     'filename': filename,
                                     'location': location,
-                                    '_authentication_token': self.authentication_token(),
+                                    '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         self.checkSessionFlash(response, 'Location must be relative path and must not contain .. in path')
     @parametrize('cnt,location,filename', [
@@ @@ -460,13 +460,13 @@ class TestFilesController(TestController @@
                                       repo_name=repo.repo_name,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "foo",
                                     'filename': filename,
                                     'location': location,
-                                    '_authentication_token': self.authentication_token(),
+                                    '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
         finally:
@@ @@ -490,13 +490,13 @@ class TestFilesController(TestController @@
                                       repo_name=repo.repo_name,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
-                                    '_authentication_token': self.authentication_token(),
+                                    '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
@@ @@ -521,13 +521,13 @@ class TestFilesController(TestController @@
                                       revision='tip',
                                       f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
-                                    '_authentication_token': self.authentication_token(),
+                                    '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
@@ @@ -535,13 +535,13 @@ class TestFilesController(TestController @@
                                           repo_name=repo.repo_name,
                                           revision=repo.scm_instance.DEFAULT_BRANCH_NAME,
                                           f_path=posixpath.join(location, filename)),
                                      params={
                                         'content': "def py():\n print 'hello world'\n",
                                         'message': 'i committed',
-                                        '_authentication_token': self.authentication_token(),
+                                        '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                      },
                                     status=302)
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
         finally:
             fixture.destroy_repo(repo.repo_name)
@@ @@ -564,13 +564,13 @@ class TestFilesController(TestController @@
                                       repo_name=repo.repo_name,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
-                                    '_authentication_token': self.authentication_token(),
+                                    '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
@@ @@ -595,13 +595,13 @@ class TestFilesController(TestController @@
                                       revision='tip',
                                       f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
-                                    '_authentication_token': self.authentication_token(),
+                                    '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
@@ @@ -609,13 +609,13 @@ class TestFilesController(TestController @@
                                           repo_name=repo.repo_name,
                                           revision=repo.scm_instance.DEFAULT_BRANCH_NAME,
                                           f_path=posixpath.join(location, filename)),
                                      params={
                                         'content': "def py():\n print 'hello world'\n",
                                         'message': 'i committed',
-                                        '_authentication_token': self.authentication_token(),
+                                        '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                      },
                                     status=302)
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
         finally:
             fixture.destroy_repo(repo.repo_name)
@@ @@ -638,13 +638,13 @@ class TestFilesController(TestController @@
                                       repo_name=repo.repo_name,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
-                                    '_authentication_token': self.authentication_token(),
+                                    '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
@@ @@ -669,26 +669,26 @@ class TestFilesController(TestController @@
                                       revision='tip',
                                       f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
-                                    '_authentication_token': self.authentication_token(),
+                                    '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
             response = self.app.post(url('files_delete_home',
                                           repo_name=repo.repo_name,
                                           revision=repo.scm_instance.DEFAULT_BRANCH_NAME,
                                           f_path=posixpath.join(location, filename)),
                                      params={
                                         'message': 'i committed',
-                                        '_authentication_token': self.authentication_token(),
+                                        '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                      },
                                     status=302)
             self.checkSessionFlash(response,
                                    'Successfully deleted file %s' % posixpath.join(location, filename))
         finally:
             fixture.destroy_repo(repo.repo_name)
@@ @@ -711,13 +711,13 @@ class TestFilesController(TestController @@
                                       repo_name=repo.repo_name,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
-                                    '_authentication_token': self.authentication_token(),
+                                    '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
@@ @@ -742,26 +742,26 @@ class TestFilesController(TestController @@
                                       revision='tip',
                                       f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
-                                    '_authentication_token': self.authentication_token(),
+                                    '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
             response = self.app.post(url('files_delete_home',
                                           repo_name=repo.repo_name,
                                           revision=repo.scm_instance.DEFAULT_BRANCH_NAME,
                                           f_path=posixpath.join(location, filename)),
                                      params={
                                         'message': 'i committed',
-                                        '_authentication_token': self.authentication_token(),
+                                        '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                      },
                                     status=302)
             self.checkSessionFlash(response,
                                    'Successfully deleted file %s' % posixpath.join(location, filename))
         finally:
             fixture.destroy_repo(repo.repo_name)

kallithea/tests/functional/test_forks.py

➞

Show inline comments

@@ @@ -51,13 +51,13 @@ class _BaseTestCase(TestController): @@
             user_model.revoke_perm(usr, 'hg.fork.repository')
             user_model.grant_perm(usr, 'hg.fork.none')
             Session().commit()
             # try create a fork
             repo_name = self.REPO
             self.app.post(url(controller='forks', action='fork_create',
-                              repo_name=repo_name), {'_authentication_token': self.authentication_token()}, status=403)
+                              repo_name=repo_name), {'_session_csrf_secret_token': self.session_csrf_secret_token()}, status=403)
         finally:
             usr = User.get_default_user()
             user_model.revoke_perm(usr, 'hg.fork.none')
             user_model.grant_perm(usr, 'hg.fork.repository')
             Session().commit()
@@ @@ -74,13 +74,13 @@ class _BaseTestCase(TestController): @@
             'repo_group': u'-1',
             'fork_parent_id': org_repo.repo_id,
             'repo_type': self.REPO_TYPE,
             'description': description,
             'private': 'False',
             'landing_rev': 'rev:tip',
-            '_authentication_token': self.authentication_token()}
+            '_session_csrf_secret_token': self.session_csrf_secret_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))
@@ @@ -88,13 +88,13 @@ class _BaseTestCase(TestController): @@
         response.mustcontain(
             """<a href="/%s">%s</a>""" % (fork_name, fork_name)
+        )
         # remove this fork
         response = self.app.post(url('delete_repo', repo_name=fork_name),
-            params={'_authentication_token': self.authentication_token()})
+            params={'_session_csrf_secret_token': self.session_csrf_secret_token()})
     def test_fork_create_into_group(self):
         self.log_user()
         group = fixture.create_repo_group(u'vc')
         group_id = group.group_id
         fork_name = self.REPO_FORK
@@ @@ -107,13 +107,13 @@ class _BaseTestCase(TestController): @@
             'repo_group': group_id,
             'fork_parent_id': org_repo.repo_id,
             'repo_type': self.REPO_TYPE,
             'description': description,
             'private': 'False',
             'landing_rev': 'rev:tip',
-            '_authentication_token': self.authentication_token()}
+            '_session_csrf_secret_token': self.session_csrf_secret_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
@@ @@ -151,13 +151,13 @@ class _BaseTestCase(TestController): @@
             'repo_group': u'-1',
             'fork_parent_id': org_repo.repo_id,
             'repo_type': self.REPO_TYPE,
             'description': 'unicode repo 1',
             'private': 'False',
             'landing_rev': 'rev:tip',
-            '_authentication_token': self.authentication_token()}
+            '_session_csrf_secret_token': self.session_csrf_secret_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>""" % (urllib.quote(fork_name), fork_name)
@@ @@ -172,26 +172,26 @@ class _BaseTestCase(TestController): @@
             'repo_group': u'-1',
             'fork_parent_id': fork_repo.repo_id,
             'repo_type': self.REPO_TYPE,
             'description': 'unicode repo 2',
             'private': 'False',
             'landing_rev': 'rev:tip',
-            '_authentication_token': self.authentication_token()}
+            '_session_csrf_secret_token': self.session_csrf_secret_token()}
         self.app.post(url(controller='forks', action='fork_create',
                           repo_name=fork_name), creation_args)
         response = self.app.get(url(controller='forks', action='forks',
                                     repo_name=fork_name))
         response.mustcontain(
             """<a href="/%s">%s</a>""" % (urllib.quote(fork_name_2), fork_name_2)
+        )
         # remove these forks
         response = self.app.post(url('delete_repo', repo_name=fork_name_2),
-            params={'_authentication_token': self.authentication_token()})
+            params={'_session_csrf_secret_token': self.session_csrf_secret_token()})
         response = self.app.post(url('delete_repo', repo_name=fork_name),
-            params={'_authentication_token': self.authentication_token()})
+            params={'_session_csrf_secret_token': self.session_csrf_secret_token()})
     def test_fork_create_and_permissions(self):
         self.log_user()
         fork_name = self.REPO_FORK
         description = 'fork of vcs test'
         repo_name = self.REPO
@@ @@ -201,13 +201,13 @@ class _BaseTestCase(TestController): @@
             'repo_group': u'-1',
             'fork_parent_id': org_repo.repo_id,
             'repo_type': self.REPO_TYPE,
             'description': description,
             'private': 'False',
             'landing_rev': 'rev:tip',
-            '_authentication_token': self.authentication_token()}
+            '_session_csrf_secret_token': self.session_csrf_secret_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

kallithea/tests/functional/test_login.py

➞

Show inline comments

@@ @@ -29,36 +29,36 @@ class TestLoginController(TestController @@
         # Test response...
     def test_login_admin_ok(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_ADMIN_LOGIN,
                                   'password': TEST_USER_ADMIN_PASS,
-                                  '_authentication_token': self.authentication_token()})
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})
         assert response.status == '302 Found'
         self.assert_authenticated_user(response, TEST_USER_ADMIN_LOGIN)
         response = response.follow()
         response.mustcontain('/%s' % HG_REPO)
     def test_login_regular_ok(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_REGULAR_LOGIN,
                                   'password': TEST_USER_REGULAR_PASS,
-                                  '_authentication_token': self.authentication_token()})
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})
         assert response.status == '302 Found'
         self.assert_authenticated_user(response, TEST_USER_REGULAR_LOGIN)
         response = response.follow()
         response.mustcontain('/%s' % HG_REPO)
     def test_login_regular_email_ok(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_REGULAR_EMAIL,
                                   'password': TEST_USER_REGULAR_PASS,
-                                  '_authentication_token': self.authentication_token()})
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})
         assert response.status == '302 Found'
         self.assert_authenticated_user(response, TEST_USER_REGULAR_LOGIN)
         response = response.follow()
         response.mustcontain('/%s' % HG_REPO)
@@ @@ -66,46 +66,46 @@ class TestLoginController(TestController @@
     def test_login_ok_came_from(self):
         test_came_from = '/_admin/users'
         response = self.app.post(url(controller='login', action='index',
                                      came_from=test_came_from),
                                  {'username': TEST_USER_ADMIN_LOGIN,
                                   'password': TEST_USER_ADMIN_PASS,
-                                  '_authentication_token': self.authentication_token()})
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})
         assert response.status == '302 Found'
         response = response.follow()
         assert response.status == '200 OK'
         response.mustcontain('Users Administration')
     def test_login_do_not_remember(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_REGULAR_LOGIN,
                                   'password': TEST_USER_REGULAR_PASS,
                                   'remember': False,
-                                  '_authentication_token': self.authentication_token()})
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})
         assert 'Set-Cookie' in response.headers
         for cookie in response.headers.getall('Set-Cookie'):
             assert not re.search(r';\s+(Max-Age|Expires)=', cookie, re.IGNORECASE), 'Cookie %r has expiration date, but should be a session cookie' % cookie
     def test_login_remember(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_REGULAR_LOGIN,
                                   'password': TEST_USER_REGULAR_PASS,
                                   'remember': True,
-                                  '_authentication_token': self.authentication_token()})
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})
         assert 'Set-Cookie' in response.headers
         for cookie in response.headers.getall('Set-Cookie'):
             assert re.search(r';\s+(Max-Age|Expires)=', cookie, re.IGNORECASE), 'Cookie %r should have expiration date, but is a session cookie' % cookie
     def test_logout(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_REGULAR_LOGIN,
                                   'password': TEST_USER_REGULAR_PASS,
-                                  '_authentication_token': self.authentication_token()})
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})
         # Verify that a login session has been established.
         response = self.app.get(url(controller='login', action='index'))
         response = response.follow()
         assert 'authuser' in response.session
@@ @@ -128,37 +128,37 @@ class TestLoginController(TestController @@
     ])
     def test_login_bad_came_froms(self, url_came_from):
         response = self.app.post(url(controller='login', action='index',
                                      came_from=url_came_from),
                                  {'username': TEST_USER_ADMIN_LOGIN,
                                   'password': TEST_USER_ADMIN_PASS,
-                                  '_authentication_token': self.authentication_token()},
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token()},
                                  status=400)
     def test_login_short_password(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_ADMIN_LOGIN,
                                   'password': 'as',
-                                  '_authentication_token': self.authentication_token()})
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})
         assert response.status == '200 OK'
         response.mustcontain('Enter 3 characters or more')
     def test_login_wrong_username_password(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': 'error',
                                   'password': 'test12',
-                                  '_authentication_token': self.authentication_token()})
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})
         response.mustcontain('Invalid username or password')
     def test_login_non_ascii(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_REGULAR_LOGIN,
                                   'password': 'blåbærgrød',
-                                  '_authentication_token': self.authentication_token()})
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})
         response.mustcontain('>Invalid username or password<')
     # verify that get arguments are correctly passed along login redirection
     @parametrize('args,args_encoded', [
@@ @@ -196,13 +196,13 @@ class TestLoginController(TestController @@
     ])
     def test_redirection_after_successful_login_preserves_get_args(self, args, args_encoded):
         response = self.app.post(url(controller='login', action='index',
                                      came_from=url('/_admin/users', **args)),
                                  {'username': TEST_USER_ADMIN_LOGIN,
                                   'password': TEST_USER_ADMIN_PASS,
-                                  '_authentication_token': self.authentication_token()})
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})
         assert response.status == '302 Found'
         for encoded in args_encoded:
             assert encoded in response.location
     @parametrize('args,args_encoded', [
         ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),
@@ @@ -211,13 +211,13 @@ class TestLoginController(TestController @@
     ])
     def test_login_form_after_incorrect_login_preserves_get_args(self, args, args_encoded):
         response = self.app.post(url(controller='login', action='index',
                                      came_from=url('/_admin/users', **args)),
                                  {'username': 'error',
                                   'password': 'test12',
-                                  '_authentication_token': self.authentication_token()})
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})
         response.mustcontain('Invalid username or password')
         came_from = urlparse.parse_qs(urlparse.urlparse(response.form.action).query)['came_from'][0]
         for encoded in args_encoded:
             assert encoded in came_from
@@ @@ -234,13 +234,13 @@ class TestLoginController(TestController @@
                                             {'username': uname,
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': 'goodmail@example.com',
                                              'firstname': 'test',
                                              'lastname': 'test',
-                                             '_authentication_token': self.authentication_token()})
+                                             '_session_csrf_secret_token': self.session_csrf_secret_token()})
         with test_context(self.app):
             msg = validators.ValidUsername()._messages['username_exists']
         msg = h.html_escape(msg % {'username': uname})
         response.mustcontain(msg)
@@ @@ -249,13 +249,13 @@ class TestLoginController(TestController @@
                                             {'username': 'test_admin_0',
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': TEST_USER_ADMIN_EMAIL,
                                              'firstname': 'test',
                                              'lastname': 'test',
-                                             '_authentication_token': self.authentication_token()})
+                                             '_session_csrf_secret_token': self.session_csrf_secret_token()})
         with test_context(self.app):
             msg = validators.UniqSystemEmail()()._messages['email_taken']
         response.mustcontain(msg)
     def test_register_err_same_email_case_sensitive(self):
@@ @@ -263,39 +263,39 @@ class TestLoginController(TestController @@
                                             {'username': 'test_admin_1',
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': TEST_USER_ADMIN_EMAIL.title(),
                                              'firstname': 'test',
                                              'lastname': 'test',
-                                             '_authentication_token': self.authentication_token()})
+                                             '_session_csrf_secret_token': self.session_csrf_secret_token()})
         with test_context(self.app):
             msg = validators.UniqSystemEmail()()._messages['email_taken']
         response.mustcontain(msg)
     def test_register_err_wrong_data(self):
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': 'xs',
                                              'password': 'test',
                                              'password_confirmation': 'test',
                                              'email': 'goodmailm',
                                              'firstname': 'test',
                                              'lastname': 'test',
-                                             '_authentication_token': self.authentication_token()})
+                                             '_session_csrf_secret_token': self.session_csrf_secret_token()})
         assert response.status == '200 OK'
         response.mustcontain('An email address must contain a single @')
         response.mustcontain('Enter a value 6 characters long or more')
     def test_register_err_username(self):
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': 'error user',
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': 'goodmailm',
                                              'firstname': 'test',
                                              'lastname': 'test',
-                                             '_authentication_token': self.authentication_token()})
+                                             '_session_csrf_secret_token': self.session_csrf_secret_token()})
         response.mustcontain('An email address must contain a single @')
         response.mustcontain('Username may only contain '
                 'alphanumeric characters underscores, '
                 'periods or dashes and must begin with an '
                 'alphanumeric character')
@@ @@ -306,13 +306,13 @@ class TestLoginController(TestController @@
                                             {'username': usr,
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': 'goodmailm',
                                              'firstname': 'test',
                                              'lastname': 'test',
-                                             '_authentication_token': self.authentication_token()})
+                                             '_session_csrf_secret_token': self.session_csrf_secret_token()})
         response.mustcontain('An email address must contain a single @')
         with test_context(self.app):
             msg = validators.ValidUsername()._messages['username_exists']
         msg = h.html_escape(msg % {'username': usr})
         response.mustcontain(msg)
@@ @@ -322,13 +322,13 @@ class TestLoginController(TestController @@
                                         {'username': 'xxxaxn',
                                          'password': 'ąćźżąśśśś',
                                          'password_confirmation': 'ąćźżąśśśś',
                                          'email': 'goodmailm@test.plx',
                                          'firstname': 'test',
                                          'lastname': 'test',
-                                         '_authentication_token': self.authentication_token()})
+                                         '_session_csrf_secret_token': self.session_csrf_secret_token()})
         with test_context(self.app):
             msg = validators.ValidPassword()._messages['invalid_password']
         response.mustcontain(msg)
     def test_register_password_mismatch(self):
@@ @@ -336,13 +336,13 @@ class TestLoginController(TestController @@
                                             {'username': 'xs',
                                              'password': '123qwe',
                                              'password_confirmation': 'qwe123',
                                              'email': 'goodmailm@test.plxa',
                                              'firstname': 'test',
                                              'lastname': 'test',
-                                             '_authentication_token': self.authentication_token()})
+                                             '_session_csrf_secret_token': self.session_csrf_secret_token()})
         with test_context(self.app):
             msg = validators.ValidPasswordsMatch('password', 'password_confirmation')._messages['password_mismatch']
         response.mustcontain(msg)
     def test_register_ok(self):
         username = 'test_regular4'
@@ @@ -356,13 +356,13 @@ class TestLoginController(TestController @@
                                              'password': password,
                                              'password_confirmation': password,
                                              'email': email,
                                              'firstname': name,
                                              'lastname': lastname,
                                              'admin': True,
-                                             '_authentication_token': self.authentication_token()})  # This should be overridden
+                                             '_session_csrf_secret_token': self.session_csrf_secret_token()})  # This should be overridden
         assert response.status == '302 Found'
         self.checkSessionFlash(response, 'You have successfully registered with Kallithea')
         ret = Session().query(User).filter(User.username == 'test_regular4').one()
         assert ret.username == username
         assert check_password(password, ret.password) == True
@@ @@ -378,13 +378,13 @@ class TestLoginController(TestController @@
     def test_forgot_password_wrong_mail(self):
         bad_email = 'username%wrongmail.org'
         response = self.app.post(
                         url(controller='login', action='password_reset'),
                             {'email': bad_email,
-                             '_authentication_token': self.authentication_token()})
+                             '_session_csrf_secret_token': self.session_csrf_secret_token()})
         response.mustcontain('An email address must contain a single @')
     def test_forgot_password(self):
         response = self.app.get(url(controller='login',
                                     action='password_reset'))
@@ @@ -407,13 +407,13 @@ class TestLoginController(TestController @@
         Session().add(new)
         Session().commit()
         response = self.app.post(url(controller='login',
                                      action='password_reset'),
                                  {'email': email,
-                                  '_authentication_token': self.authentication_token()})
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'A password reset confirmation code has been sent')
         response = response.follow()
         # BAD TOKEN
@@ @@ -424,24 +424,24 @@ class TestLoginController(TestController @@
                                      action='password_reset_confirmation'),
                                  {'email': email,
                                   'timestamp': timestamp,
                                   'password': "p@ssw0rd",
                                   'password_confirm': "p@ssw0rd",
                                   'token': token,
-                                  '_authentication_token': self.authentication_token(),
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  })
         assert response.status == '200 OK'
         response.mustcontain('Invalid password reset token')
         # GOOD TOKEN
         # TODO: The token should ideally be taken from the mail sent
         # above, instead of being recalculated.
         token = UserModel().get_reset_password_token(
-            User.get_by_username(username), timestamp, self.authentication_token())
+            User.get_by_username(username), timestamp, self.session_csrf_secret_token())
         response = self.app.get(url(controller='login',
                                     action='password_reset_confirmation',
                                     email=email,
                                     timestamp=timestamp,
                                     token=token))
@@ @@ -452,13 +452,13 @@ class TestLoginController(TestController @@
                                      action='password_reset_confirmation'),
                                  {'email': email,
                                   'timestamp': timestamp,
                                   'password': "p@ssw0rd",
                                   'password_confirm': "p@ssw0rd",
                                   'token': token,
-                                  '_authentication_token': self.authentication_token(),
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  })
         assert response.status == '302 Found'
         self.checkSessionFlash(response, 'Successfully updated password')
         response = response.follow()

kallithea/tests/functional/test_my_account.py

➞

Show inline comments

@@ @@ -51,43 +51,43 @@ class TestMyAccountController(TestContro @@
     def test_my_account_my_emails_add_existing_email(self):
         self.log_user()
         response = self.app.get(url('my_account_emails'))
         response.mustcontain('No additional emails specified')
         response = self.app.post(url('my_account_emails'),
-                                 {'new_email': TEST_USER_REGULAR_EMAIL, '_authentication_token': self.authentication_token()})
+                                 {'new_email': TEST_USER_REGULAR_EMAIL, '_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'This email address is already in use')
     def test_my_account_my_emails_add_missing_email_in_form(self):
         self.log_user()
         response = self.app.get(url('my_account_emails'))
         response.mustcontain('No additional emails specified')
         response = self.app.post(url('my_account_emails'),
-            {'_authentication_token': self.authentication_token()})
+            {'_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'Please enter an email address')
     def test_my_account_my_emails_add_remove(self):
         self.log_user()
         response = self.app.get(url('my_account_emails'))
         response.mustcontain('No additional emails specified')
         response = self.app.post(url('my_account_emails'),
-                                 {'new_email': 'barz@example.com', '_authentication_token': self.authentication_token()})
+                                 {'new_email': 'barz@example.com', '_session_csrf_secret_token': self.session_csrf_secret_token()})
         response = self.app.get(url('my_account_emails'))
         from kallithea.model.db import UserEmailMap
         email_id = UserEmailMap.query() \
             .filter(UserEmailMap.user == User.get_by_username(TEST_USER_ADMIN_LOGIN)) \
             .filter(UserEmailMap.email == 'barz@example.com').one().email_id
         response.mustcontain('barz@example.com')
         response.mustcontain('<input id="del_email_id" name="del_email_id" type="hidden" value="%s" />' % email_id)
         response = self.app.post(url('my_account_emails_delete'),
-                                 {'del_email_id': email_id, '_authentication_token': self.authentication_token()})
+                                 {'del_email_id': email_id, '_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'Removed email from user')
         response = self.app.get(url('my_account_emails'))
         response.mustcontain('No additional emails specified')
     @parametrize('name,attrs',
@@ @@ -116,13 +116,13 @@ class TestMyAccountController(TestContro @@
         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({'_session_csrf_secret_token': self.session_csrf_secret_token()})
         params.update(attrs)
         response = self.app.post(url('my_account'), params)
         self.checkSessionFlash(response,
                                'Your account was updated successfully')
@@ @@ -145,13 +145,13 @@ class TestMyAccountController(TestContro @@
             # 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')
+        params.pop('_session_csrf_secret_token')
         assert params == updated_params
     def test_my_account_update_err_email_exists(self):
         self.log_user()
         new_email = TEST_USER_REGULAR_EMAIL  # already existing email
@@ @@ -160,13 +160,13 @@ class TestMyAccountController(TestContro @@
                                     username=TEST_USER_ADMIN_LOGIN,
                                     new_password=TEST_USER_ADMIN_PASS,
                                     password_confirmation='test122',
                                     firstname=u'NewName',
                                     lastname=u'NewLastname',
                                     email=new_email,
-                                    _authentication_token=self.authentication_token())
+                                    _session_csrf_secret_token=self.session_csrf_secret_token())
+                                )
         response.mustcontain('This email address is already in use')
     def test_my_account_update_err(self):
         self.log_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS)
@@ @@ -177,13 +177,13 @@ class TestMyAccountController(TestContro @@
                                             username=TEST_USER_ADMIN_LOGIN,
                                             new_password=TEST_USER_ADMIN_PASS,
                                             password_confirmation='test122',
                                             firstname=u'NewName',
                                             lastname=u'NewLastname',
                                             email=new_email,
-                                            _authentication_token=self.authentication_token()))
+                                            _session_csrf_secret_token=self.session_csrf_secret_token()))
         response.mustcontain('An email address must contain a single @')
         from kallithea.model import validators
         with test_context(self.app):
             msg = validators.ValidUsername(edit=False, old_data={}) \
                     ._messages['username_exists']
@@ @@ -203,13 +203,13 @@ class TestMyAccountController(TestContro @@
         ('30days', 60*60*24*30),
     ])
     def test_my_account_add_api_keys(self, desc, lifetime):
         usr = self.log_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS)
         user = User.get(usr['user_id'])
         response = self.app.post(url('my_account_api_keys'),
-                                 {'description': desc, 'lifetime': lifetime, '_authentication_token': self.authentication_token()})
+                                 {'description': desc, 'lifetime': lifetime, '_session_csrf_secret_token': self.session_csrf_secret_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)
@@ @@ -219,22 +219,22 @@ class TestMyAccountController(TestContro @@
                 Session().commit()
     def test_my_account_remove_api_key(self):
         usr = self.log_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS)
         user = User.get(usr['user_id'])
         response = self.app.post(url('my_account_api_keys'),
-                                 {'description': 'desc', 'lifetime': -1, '_authentication_token': self.authentication_token()})
+                                 {'description': 'desc', 'lifetime': -1, '_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'API key successfully created')
         response = response.follow()
         # now delete our key
         keys = UserApiKeys.query().all()
         assert 1 == len(keys)
         response = self.app.post(url('my_account_api_keys_delete'),
-                 {'del_api_key': keys[0].api_key, '_authentication_token': self.authentication_token()})
+                 {'del_api_key': keys[0].api_key, '_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'API key successfully deleted')
         keys = UserApiKeys.query().all()
         assert 0 == len(keys)
     def test_my_account_reset_main_api_key(self):
         usr = self.log_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS)
@@ @@ -242,13 +242,13 @@ class TestMyAccountController(TestContro @@
         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_delete'),
-                 {'del_api_key_builtin': api_key, '_authentication_token': self.authentication_token()})
+                 {'del_api_key_builtin': api_key, '_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'API key successfully reset')
         response = response.follow()
         response.mustcontain(no=[api_key])
     def test_my_account_add_ssh_key(self):
         description = u'something'
@@ @@ -256,13 +256,13 @@ class TestMyAccountController(TestContro @@
         fingerprint = u'Ke3oUCNJM87P0jJTb3D+e3shjceP2CqMpQKVd75E9I8'
         self.log_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS)
         response = self.app.post(url('my_account_ssh_keys'),
                                  {'description': description,
                                   'public_key': public_key,
-                                  '_authentication_token': self.authentication_token()})
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'SSH key %s successfully added' % fingerprint)
         response = response.follow()
         response.mustcontain(fingerprint)
         user_id = response.session['authuser']['user_id']
         ssh_key = UserSshKeys.query().filter(UserSshKeys.user_id == user_id).one()
@@ @@ -277,19 +277,19 @@ class TestMyAccountController(TestContro @@
         fingerprint = u'Ke3oUCNJM87P0jJTb3D+e3shjceP2CqMpQKVd75E9I8'
         self.log_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS)
         response = self.app.post(url('my_account_ssh_keys'),
                                  {'description': description,
                                   'public_key': public_key,
-                                  '_authentication_token': self.authentication_token()})
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'SSH key %s successfully added' % fingerprint)
         response.follow()
         user_id = response.session['authuser']['user_id']
         ssh_key = UserSshKeys.query().filter(UserSshKeys.user_id == user_id).one()
         assert ssh_key.description == u'me@localhost'
         response = self.app.post(url('my_account_ssh_keys_delete'),
                                  {'del_public_key': ssh_key.public_key,
-                                  '_authentication_token': self.authentication_token()})
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'SSH key successfully deleted')
         keys = UserSshKeys.query().all()
         assert 0 == len(keys)

kallithea/tests/functional/test_pullrequests.py

➞

Show inline comments

@@ @@ -27,13 +27,13 @@ class TestPullrequestsController(TestCon @@
                                  {'org_repo': HG_REPO,
                                   'org_ref': 'branch:stable:4f7e2131323e0749a740c0a56ab68ae9269c562a',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
-                                  '_authentication_token': self.authentication_token(),
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         response = response.follow()
         assert response.status == '200 OK'
         response.mustcontain('Successfully opened new pull request')
         response.mustcontain('No additional changesets found for iterating on this pull request')
@@ @@ -46,13 +46,13 @@ class TestPullrequestsController(TestCon @@
                                  {'org_repo': HG_REPO,
                                   'org_ref': 'rev:94f45ed825a1:94f45ed825a113e61af7e141f44ca578374abef0',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
-                                  '_authentication_token': self.authentication_token(),
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         response = response.follow()
         assert response.status == '200 OK'
         response.mustcontain(no='No additional changesets found for iterating on this pull request')
         response.mustcontain('The following additional changes are available on stable:')
@@ @@ -66,13 +66,13 @@ class TestPullrequestsController(TestCon @@
                                  {'org_repo': HG_REPO,
                                   'org_ref': 'branch:stable:4f7e2131323e0749a740c0a56ab68ae9269c562a',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'rev:94f45ed825a1:94f45ed825a113e61af7e141f44ca578374abef0',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
-                                  '_authentication_token': self.authentication_token(),
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         response = response.follow()
         assert response.status == '200 OK'
         response.mustcontain('No additional changesets found for iterating on this pull request')
         response.mustcontain('href="/vcs_test_hg/changeset/4f7e2131323e0749a740c0a56ab68ae9269c562a"')
@@ @@ -89,13 +89,13 @@ class TestPullrequestsController(TestCon @@
                                  {'org_repo': HG_REPO,
                                   'org_ref': 'rev:94f45ed825a1:94f45ed825a113e61af7e141f44ca578374abef0',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
-                                  '_authentication_token': self.authentication_token(),
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         pull_request1_id = re.search('/pull-request/(\d+)/', response.location).group(1)
         assert response.location == 'http://localhost/%s/pull-request/%s/_/stable' % (HG_REPO, pull_request1_id)
         # create new iteration
@@ @@ -103,13 +103,13 @@ class TestPullrequestsController(TestCon @@
                                      repo_name=HG_REPO, pull_request_id=pull_request1_id),
+                                 {
                                   'updaterev': '4f7e2131323e0749a740c0a56ab68ae9269c562a',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   'owner': TEST_USER_ADMIN_LOGIN,
-                                  '_authentication_token': self.authentication_token(),
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                   'review_members': [regular_user.user_id],
                                  },
                                  status=302)
         pull_request2_id = re.search('/pull-request/(\d+)/', response.location).group(1)
         assert pull_request2_id != pull_request1_id
         assert response.location == 'http://localhost/%s/pull-request/%s/_/stable' % (HG_REPO, pull_request2_id)
@@ @@ -121,13 +121,13 @@ class TestPullrequestsController(TestCon @@
         response = self.app.post(url(controller='pullrequests', action='post',
                                      repo_name=HG_REPO, pull_request_id=pull_request2_id),
+                                 {
                                   'pullrequest_title': 'Title',
                                   'pullrequest_desc': 'description',
                                   'owner': TEST_USER_ADMIN_LOGIN,
-                                  '_authentication_token': self.authentication_token(),
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                   'org_review_members': [admin_user.user_id], # fake - just to get some 'meanwhile' warning ... but it is also added ...
                                   'review_members': [regular_user2.user_id, admin_user.user_id],
                                  },
                                  status=302)
         assert response.location == 'http://localhost/%s/pull-request/%s/_/stable' % (HG_REPO, pull_request2_id)
         response = response.follow()
@@ @@ -148,13 +148,13 @@ class TestPullrequestsController(TestCon @@
                                   'org_repo': HG_REPO,
                                   'org_ref': 'rev:94f45ed825a1:94f45ed825a113e61af7e141f44ca578374abef0',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
-                                  '_authentication_token': self.authentication_token(),
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                 status=302)
         # location is of the form:
         # http://localhost/vcs_test_hg/pull-request/54/_/title
         m = re.search('/pull-request/(\d+)/', response.location)
         assert m is not None
@@ @@ -165,13 +165,13 @@ class TestPullrequestsController(TestCon @@
                                      repo_name=HG_REPO, pull_request_id=pull_request_id),
+                                 {
                                   'updaterev': '4f7e2131323e0749a740c0a56ab68ae9269c562a',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   'owner': TEST_USER_ADMIN_LOGIN,
-                                  '_authentication_token': self.authentication_token(),
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                   'review_members': [str(invalid_user_id)],
                                  },
                                  status=400)
         response.mustcontain('Invalid reviewer &#34;%s&#34; specified' % invalid_user_id)
     def test_edit_with_invalid_reviewer(self):
@@ @@ -184,13 +184,13 @@ class TestPullrequestsController(TestCon @@
                                   'org_repo': HG_REPO,
                                   'org_ref': 'branch:stable:4f7e2131323e0749a740c0a56ab68ae9269c562a',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
-                                  '_authentication_token': self.authentication_token(),
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                 status=302)
         # location is of the form:
         # http://localhost/vcs_test_hg/pull-request/54/_/title
         m = re.search('/pull-request/(\d+)/', response.location)
         assert m is not None
@@ @@ -200,13 +200,13 @@ class TestPullrequestsController(TestCon @@
         response = self.app.post(url(controller='pullrequests', action='post',
                                      repo_name=HG_REPO, pull_request_id=pull_request_id),
+                                 {
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   'owner': TEST_USER_ADMIN_LOGIN,
-                                  '_authentication_token': self.authentication_token(),
+                                  '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                   'review_members': [str(invalid_user_id)],
                                  },
                                  status=400)
         response.mustcontain('Invalid reviewer &#34;%s&#34; specified' % invalid_user_id)
     def test_iteration_refs(self):
@@ @@ -232,13 +232,13 @@ class TestPullrequestsController(TestCon @@
                 'org_repo': HG_REPO,
                 'org_ref': 'rev:9e6119747791:9e6119747791ff886a5abe1193a730b6bf874e1c',
                 'other_repo': HG_REPO,
                 'other_ref': 'branch:default:3d1091ee5a533b1f4577ec7d8a226bb315fb1336',
                 'pullrequest_title': 'title',
                 'pullrequest_desc': 'description',
-                '_authentication_token': self.authentication_token(),
+                '_session_csrf_secret_token': self.session_csrf_secret_token(),
             },
             status=302)
         pr1_id = int(re.search('/pull-request/(\d+)/', response.location).group(1))
         pr1 = PullRequest.get(pr1_id)
         assert pr1.org_ref == 'branch:webvcs:9e6119747791ff886a5abe1193a730b6bf874e1c'
@@ @@ -251,13 +251,13 @@ class TestPullrequestsController(TestCon @@
             url(controller='pullrequests', action='post', repo_name=HG_REPO, pull_request_id=pr1_id),
+            {
                 'updaterev': '5ec21f21aafe95220f1fc4843a4a57c378498b71',
                 'pullrequest_title': 'title',
                 'pullrequest_desc': 'description',
                 'owner': TEST_USER_REGULAR_LOGIN,
-                '_authentication_token': self.authentication_token(),
+                '_session_csrf_secret_token': self.session_csrf_secret_token(),
              },
              status=302)
         pr2_id = int(re.search('/pull-request/(\d+)/', response.location).group(1))
         pr1 = PullRequest.get(pr1_id)
         pr2 = PullRequest.get(pr2_id)
@@ @@ -273,13 +273,13 @@ class TestPullrequestsController(TestCon @@
             url(controller='pullrequests', action='post', repo_name=HG_REPO, pull_request_id=pr2_id),
+            {
                 'updaterev': 'fb95b340e0d03fa51f33c56c991c08077c99303e',
                 'pullrequest_title': 'title',
                 'pullrequest_desc': 'description',
                 'owner': TEST_USER_REGULAR_LOGIN,
-                '_authentication_token': self.authentication_token(),
+                '_session_csrf_secret_token': self.session_csrf_secret_token(),
              },
              status=302)
         pr3_id = int(re.search('/pull-request/(\d+)/', response.location).group(1))
         pr2 = PullRequest.get(pr2_id)
         pr3 = PullRequest.get(pr3_id)

kallithea/tests/functional/test_repo_groups.py

➞

Show inline comments

@@ @@ -17,41 +17,41 @@ class TestRepoGroupsController(TestContr @@
         group_name = 'foo'
         # creation with form error
         response = self.app.post(url('repos_groups'),
                                          {'group_name': group_name,
-                                          '_authentication_token': self.authentication_token()})
+                                          '_session_csrf_secret_token': self.session_csrf_secret_token()})
         response.mustcontain('name="group_name" type="text" value="%s"' % group_name)
         response.mustcontain('<!-- for: group_description -->')
         # creation
         response = self.app.post(url('repos_groups'),
                                          {'group_name': group_name,
                                          'group_description': 'lala',
                                          'parent_group_id': '-1',
                                          'group_copy_permissions': 'True',
-                                          '_authentication_token': self.authentication_token()})
+                                          '_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'Created repository group %s' % group_name)
         # edit form
         response = self.app.get(url('edit_repo_group', group_name=group_name))
         response.mustcontain('>lala<')
         # edit with form error
         response = self.app.post(url('update_repos_group', group_name=group_name),
                                          {'group_name': group_name,
-                                          '_authentication_token': self.authentication_token()})
+                                          '_session_csrf_secret_token': self.session_csrf_secret_token()})
         response.mustcontain('name="group_name" type="text" value="%s"' % group_name)
         response.mustcontain('<!-- for: group_description -->')
         # edit
         response = self.app.post(url('update_repos_group', group_name=group_name),
                                          {'group_name': group_name,
                                          'group_description': 'lolo',
-                                          '_authentication_token': self.authentication_token()})
+                                          '_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'Updated repository group %s' % group_name)
         response = response.follow()
         response.mustcontain('name="group_name" type="text" value="%s"' % group_name)
         response.mustcontain(no='<!-- for: group_description -->')
         response.mustcontain('>lolo<')
@@ @@ -66,12 +66,12 @@ class TestRepoGroupsController(TestContr @@
         # show ignores extra trailing slashes in the URL
         response = self.app.get(url('repos_group', group_name='%s//' % group_name))
         response.mustcontain('href="/_admin/repo_groups/%s/edit"' % group_name)
         # delete
         response = self.app.post(url('delete_repo_group', group_name=group_name),
-                                 {'_authentication_token': self.authentication_token()})
+                                 {'_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'Removed repository group %s' % group_name)
     def test_new_by_regular_user(self):
         self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
         response = self.app.get(url('new_repos_group'), status=403)

0 comments (0 inline, 0 general)