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

@@ @@ -258,385 +258,385 @@ def make_map(config): @@
         m.connect("admin_permissions_perms", "/permissions/perms",
                   action="permission_perms", conditions=dict(method=["GET"]))
     # ADMIN DEFAULTS ROUTES
     with rmap.submapper(path_prefix=ADMIN_PREFIX,
                         controller='admin/defaults') as m:
         m.connect('defaults', 'defaults',
                   action="index")
         m.connect('defaults_update', 'defaults/{id}/update',
                   action="update", conditions=dict(method=["POST"]))
     # ADMIN AUTH SETTINGS
     rmap.connect('auth_settings', '%s/auth' % ADMIN_PREFIX,
                  controller='admin/auth_settings', action='auth_settings',
                  conditions=dict(method=["POST"]))
     rmap.connect('auth_home', '%s/auth' % ADMIN_PREFIX,
                  controller='admin/auth_settings')
     # ADMIN SETTINGS ROUTES
     with rmap.submapper(path_prefix=ADMIN_PREFIX,
                         controller='admin/settings') as m:
         m.connect("admin_settings", "/settings",
                   action="settings_vcs", conditions=dict(method=["POST"]))
         m.connect("admin_settings", "/settings",
                   action="settings_vcs", conditions=dict(method=["GET"]))
         m.connect("admin_settings_mapping", "/settings/mapping",
                   action="settings_mapping", conditions=dict(method=["POST"]))
         m.connect("admin_settings_mapping", "/settings/mapping",
                   action="settings_mapping", conditions=dict(method=["GET"]))
         m.connect("admin_settings_global", "/settings/global",
                   action="settings_global", conditions=dict(method=["POST"]))
         m.connect("admin_settings_global", "/settings/global",
                   action="settings_global", conditions=dict(method=["GET"]))
         m.connect("admin_settings_visual", "/settings/visual",
                   action="settings_visual", conditions=dict(method=["POST"]))
         m.connect("admin_settings_visual", "/settings/visual",
                   action="settings_visual", conditions=dict(method=["GET"]))
         m.connect("admin_settings_email", "/settings/email",
                   action="settings_email", conditions=dict(method=["POST"]))
         m.connect("admin_settings_email", "/settings/email",
                   action="settings_email", conditions=dict(method=["GET"]))
         m.connect("admin_settings_hooks", "/settings/hooks",
                   action="settings_hooks", conditions=dict(method=["POST"]))
         m.connect("admin_settings_hooks_delete", "/settings/hooks/delete",
                   action="settings_hooks", conditions=dict(method=["POST"]))
         m.connect("admin_settings_hooks", "/settings/hooks",
                   action="settings_hooks", conditions=dict(method=["GET"]))
         m.connect("admin_settings_search", "/settings/search",
                   action="settings_search", conditions=dict(method=["POST"]))
         m.connect("admin_settings_search", "/settings/search",
                   action="settings_search", conditions=dict(method=["GET"]))
         m.connect("admin_settings_system", "/settings/system",
                   action="settings_system", conditions=dict(method=["POST"]))
         m.connect("admin_settings_system", "/settings/system",
                   action="settings_system", conditions=dict(method=["GET"]))
         m.connect("admin_settings_system_update", "/settings/system/updates",
                   action="settings_system_update", conditions=dict(method=["GET"]))
     # ADMIN MY ACCOUNT
     with rmap.submapper(path_prefix=ADMIN_PREFIX,
                         controller='admin/my_account') as m:
         m.connect("my_account", "/my_account",
                   action="my_account", conditions=dict(method=["GET"]))
         m.connect("my_account", "/my_account",
                   action="my_account", conditions=dict(method=["POST"]))
         m.connect("my_account_password", "/my_account/password",
                   action="my_account_password", conditions=dict(method=["GET"]))
         m.connect("my_account_password", "/my_account/password",
                   action="my_account_password", conditions=dict(method=["POST"]))
         m.connect("my_account_repos", "/my_account/repos",
                   action="my_account_repos", conditions=dict(method=["GET"]))
         m.connect("my_account_watched", "/my_account/watched",
                   action="my_account_watched", conditions=dict(method=["GET"]))
         m.connect("my_account_perms", "/my_account/perms",
                   action="my_account_perms", conditions=dict(method=["GET"]))
         m.connect("my_account_emails", "/my_account/emails",
                   action="my_account_emails", conditions=dict(method=["GET"]))
         m.connect("my_account_emails", "/my_account/emails",
                   action="my_account_emails_add", conditions=dict(method=["POST"]))
         m.connect("my_account_emails_delete", "/my_account/emails/delete",
                   action="my_account_emails_delete", conditions=dict(method=["POST"]))
         m.connect("my_account_api_keys", "/my_account/api_keys",
                   action="my_account_api_keys", conditions=dict(method=["GET"]))
         m.connect("my_account_api_keys", "/my_account/api_keys",
                   action="my_account_api_keys_add", conditions=dict(method=["POST"]))
         m.connect("my_account_api_keys_delete", "/my_account/api_keys/delete",
                   action="my_account_api_keys_delete", conditions=dict(method=["POST"]))
         m.connect("my_account_ssh_keys", "/my_account/ssh_keys",
                   action="my_account_ssh_keys", conditions=dict(method=["GET"]))
         m.connect("my_account_ssh_keys", "/my_account/ssh_keys",
                   action="my_account_ssh_keys_add", conditions=dict(method=["POST"]))
         m.connect("my_account_ssh_keys_delete", "/my_account/ssh_keys/delete",
                   action="my_account_ssh_keys_delete", conditions=dict(method=["POST"]))
     # ADMIN GIST
     with rmap.submapper(path_prefix=ADMIN_PREFIX,
                         controller='admin/gists') as m:
         m.connect("gists", "/gists",
                   action="create", conditions=dict(method=["POST"]))
         m.connect("gists", "/gists",
                   action="index", conditions=dict(method=["GET"]))
         m.connect("new_gist", "/gists/new",
                   action="new", conditions=dict(method=["GET"]))
         m.connect("gist_delete", "/gists/{gist_id}/delete",
                   action="delete", conditions=dict(method=["POST"]))
         m.connect("edit_gist", "/gists/{gist_id}/edit",
                   action="edit", conditions=dict(method=["GET", "POST"]))
         m.connect("edit_gist_check_revision", "/gists/{gist_id}/edit/check_revision",
                   action="check_revision", conditions=dict(method=["POST"]))
         m.connect("gist", "/gists/{gist_id}",
                   action="show", conditions=dict(method=["GET"]))
         m.connect("gist_rev", "/gists/{gist_id}/{revision}",
                   revision="tip",
                   action="show", conditions=dict(method=["GET"]))
         m.connect("formatted_gist", "/gists/{gist_id}/{revision}/{format}",
                   revision="tip",
                   action="show", conditions=dict(method=["GET"]))
         m.connect("formatted_gist_file", "/gists/{gist_id}/{revision}/{format}/{f_path:.*}",
                   revision='tip',
                   action="show", conditions=dict(method=["GET"]))
     # ADMIN MAIN PAGES
     with rmap.submapper(path_prefix=ADMIN_PREFIX,
                         controller='admin/admin') as m:
         m.connect('admin_home', '', action='index')
         m.connect('admin_add_repo', '/add_repo/{new_repo:[a-z0-9\. _-]*}',
                   action='add_repo')
     #==========================================================================
     # API V2
     #==========================================================================
     with rmap.submapper(path_prefix=ADMIN_PREFIX, controller='api/api',
                         action='_dispatch') as m:
         m.connect('api', '/api')
     # USER JOURNAL
     rmap.connect('journal', '%s/journal' % ADMIN_PREFIX,
                  controller='journal', action='index')
     rmap.connect('journal_rss', '%s/journal/rss' % ADMIN_PREFIX,
                  controller='journal', action='journal_rss')
     rmap.connect('journal_atom', '%s/journal/atom' % ADMIN_PREFIX,
                  controller='journal', action='journal_atom')
     rmap.connect('public_journal', '%s/public_journal' % ADMIN_PREFIX,
                  controller='journal', action="public_journal")
     rmap.connect('public_journal_rss', '%s/public_journal/rss' % ADMIN_PREFIX,
                  controller='journal', action="public_journal_rss")
     rmap.connect('public_journal_rss_old', '%s/public_journal_rss' % ADMIN_PREFIX,
                  controller='journal', action="public_journal_rss")
     rmap.connect('public_journal_atom',
                  '%s/public_journal/atom' % ADMIN_PREFIX, controller='journal',
                  action="public_journal_atom")
     rmap.connect('public_journal_atom_old',
                  '%s/public_journal_atom' % ADMIN_PREFIX, controller='journal',
                  action="public_journal_atom")
     rmap.connect('toggle_following', '%s/toggle_following' % ADMIN_PREFIX,
                  controller='journal', action='toggle_following',
                  conditions=dict(method=["POST"]))
     # SEARCH
     rmap.connect('search', '%s/search' % ADMIN_PREFIX, controller='search',)
     rmap.connect('search_repo_admin', '%s/search/{repo_name:.*}' % ADMIN_PREFIX,
                  controller='search',
                  conditions=dict(function=check_repo))
     rmap.connect('search_repo', '/{repo_name:.*?}/search',
                  controller='search',
                  conditions=dict(function=check_repo),
+                 )
     # LOGIN/LOGOUT/REGISTER/SIGN IN
-    rmap.connect('authentication_token', '%s/authentication_token' % ADMIN_PREFIX, controller='login', action='authentication_token')
+    rmap.connect('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')
     rmap.connect('reset_password', '%s/password_reset' % ADMIN_PREFIX,
                  controller='login', action='password_reset')
     rmap.connect('reset_password_confirmation',
                  '%s/password_reset_confirmation' % ADMIN_PREFIX,
                  controller='login', action='password_reset_confirmation')
     # FEEDS
     rmap.connect('rss_feed_home', '/{repo_name:.*?}/feed/rss',
                 controller='feed', action='rss',
                 conditions=dict(function=check_repo))
     rmap.connect('atom_feed_home', '/{repo_name:.*?}/feed/atom',
                 controller='feed', action='atom',
                 conditions=dict(function=check_repo))
     #==========================================================================
     # REPOSITORY ROUTES
     #==========================================================================
     rmap.connect('repo_creating_home', '/{repo_name:.*?}/repo_creating',
                 controller='admin/repos', action='repo_creating')
     rmap.connect('repo_check_home', '/{repo_name:.*?}/crepo_check',
                 controller='admin/repos', action='repo_check')
     rmap.connect('summary_home', '/{repo_name:.*?}',
                 controller='summary',
                 conditions=dict(function=check_repo))
     # must be here for proper group/repo catching
     rmap.connect('repos_group_home', '/{group_name:.*}',
                 controller='admin/repo_groups', action="show_by_name",
                 conditions=dict(function=check_group))
     rmap.connect('repo_stats_home', '/{repo_name:.*?}/statistics',
                 controller='summary', action='statistics',
                 conditions=dict(function=check_repo))
     rmap.connect('repo_size', '/{repo_name:.*?}/repo_size',
                 controller='summary', action='repo_size',
                 conditions=dict(function=check_repo))
     rmap.connect('repo_refs_data', '/{repo_name:.*?}/refs-data',
                  controller='home', action='repo_refs_data')
     rmap.connect('changeset_home', '/{repo_name:.*?}/changeset/{revision:.*}',
                 controller='changeset', revision='tip',
                 conditions=dict(function=check_repo))
     rmap.connect('changeset_children', '/{repo_name:.*?}/changeset_children/{revision}',
                 controller='changeset', revision='tip', action="changeset_children",
                 conditions=dict(function=check_repo))
     rmap.connect('changeset_parents', '/{repo_name:.*?}/changeset_parents/{revision}',
                 controller='changeset', revision='tip', action="changeset_parents",
                 conditions=dict(function=check_repo))
     # repo edit options
     rmap.connect("edit_repo", "/{repo_name:.*?}/settings",
                  controller='admin/repos', action="edit",
                  conditions=dict(method=["GET"], function=check_repo))
     rmap.connect("edit_repo_perms", "/{repo_name:.*?}/settings/permissions",
                  controller='admin/repos', action="edit_permissions",
                  conditions=dict(method=["GET"], function=check_repo))
     rmap.connect("edit_repo_perms_update", "/{repo_name:.*?}/settings/permissions",
                  controller='admin/repos', action="edit_permissions_update",
                  conditions=dict(method=["POST"], function=check_repo))
     rmap.connect("edit_repo_perms_revoke", "/{repo_name:.*?}/settings/permissions/delete",
                  controller='admin/repos', action="edit_permissions_revoke",
                  conditions=dict(method=["POST"], function=check_repo))
     rmap.connect("edit_repo_fields", "/{repo_name:.*?}/settings/fields",
                  controller='admin/repos', action="edit_fields",
                  conditions=dict(method=["GET"], function=check_repo))
     rmap.connect('create_repo_fields', "/{repo_name:.*?}/settings/fields/new",
                  controller='admin/repos', action="create_repo_field",
                  conditions=dict(method=["POST"], function=check_repo))
     rmap.connect('delete_repo_fields', "/{repo_name:.*?}/settings/fields/{field_id}/delete",
                  controller='admin/repos', action="delete_repo_field",
                  conditions=dict(method=["POST"], function=check_repo))
     rmap.connect("edit_repo_advanced", "/{repo_name:.*?}/settings/advanced",
                  controller='admin/repos', action="edit_advanced",
                  conditions=dict(method=["GET"], function=check_repo))
     rmap.connect("edit_repo_advanced_journal", "/{repo_name:.*?}/settings/advanced/journal",
                  controller='admin/repos', action="edit_advanced_journal",
                  conditions=dict(method=["POST"], function=check_repo))
     rmap.connect("edit_repo_advanced_fork", "/{repo_name:.*?}/settings/advanced/fork",
                  controller='admin/repos', action="edit_advanced_fork",
                  conditions=dict(method=["POST"], function=check_repo))
     rmap.connect("edit_repo_caches", "/{repo_name:.*?}/settings/caches",
                  controller='admin/repos', action="edit_caches",
                  conditions=dict(method=["GET"], function=check_repo))
     rmap.connect("update_repo_caches", "/{repo_name:.*?}/settings/caches",
                  controller='admin/repos', action="edit_caches",
                  conditions=dict(method=["POST"], function=check_repo))
     rmap.connect("edit_repo_remote", "/{repo_name:.*?}/settings/remote",
                  controller='admin/repos', action="edit_remote",
                  conditions=dict(method=["GET"], function=check_repo))
     rmap.connect("edit_repo_remote_update", "/{repo_name:.*?}/settings/remote",
                  controller='admin/repos', action="edit_remote",
                  conditions=dict(method=["POST"], function=check_repo))
     rmap.connect("edit_repo_statistics", "/{repo_name:.*?}/settings/statistics",
                  controller='admin/repos', action="edit_statistics",
                  conditions=dict(method=["GET"], function=check_repo))
     rmap.connect("edit_repo_statistics_update", "/{repo_name:.*?}/settings/statistics",
                  controller='admin/repos', action="edit_statistics",
                  conditions=dict(method=["POST"], function=check_repo))
     # still working url for backward compat.
     rmap.connect('raw_changeset_home_depraced',
                  '/{repo_name:.*?}/raw-changeset/{revision}',
                  controller='changeset', action='changeset_raw',
                  revision='tip', conditions=dict(function=check_repo))
     ## new URLs
     rmap.connect('changeset_raw_home',
                  '/{repo_name:.*?}/changeset-diff/{revision}',
                  controller='changeset', action='changeset_raw',
                  revision='tip', conditions=dict(function=check_repo))
     rmap.connect('changeset_patch_home',
                  '/{repo_name:.*?}/changeset-patch/{revision}',
                  controller='changeset', action='changeset_patch',
                  revision='tip', conditions=dict(function=check_repo))
     rmap.connect('changeset_download_home',
                  '/{repo_name:.*?}/changeset-download/{revision}',
                  controller='changeset', action='changeset_download',
                  revision='tip', conditions=dict(function=check_repo))
     rmap.connect('changeset_comment',
                  '/{repo_name:.*?}/changeset-comment/{revision}',
                 controller='changeset', revision='tip', action='comment',
                 conditions=dict(function=check_repo))
     rmap.connect('changeset_comment_delete',
                  '/{repo_name:.*?}/changeset-comment/{comment_id}/delete',
                 controller='changeset', action='delete_comment',
                 conditions=dict(function=check_repo, method=["POST"]))
     rmap.connect('changeset_info', '/changeset_info/{repo_name:.*?}/{revision}',
                  controller='changeset', action='changeset_info')
     rmap.connect('compare_home',
                  '/{repo_name:.*?}/compare',
                  controller='compare', action='index',
                  conditions=dict(function=check_repo))
     rmap.connect('compare_url',
                  '/{repo_name:.*?}/compare/{org_ref_type}@{org_ref_name:.*?}...{other_ref_type}@{other_ref_name:.*?}',
                  controller='compare', action='compare',
                  conditions=dict(function=check_repo),
                  requirements=dict(
                             org_ref_type='(branch|book|tag|rev|__other_ref_type__)',
                             other_ref_type='(branch|book|tag|rev|__org_ref_type__)')
+                 )
     rmap.connect('pullrequest_home',
                  '/{repo_name:.*?}/pull-request/new', controller='pullrequests',
                  action='index', conditions=dict(function=check_repo,
                                                  method=["GET"]))
     rmap.connect('pullrequest_repo_info',
                  '/{repo_name:.*?}/pull-request-repo-info',
                  controller='pullrequests', action='repo_info',
                  conditions=dict(function=check_repo, method=["GET"]))
     rmap.connect('pullrequest',
                  '/{repo_name:.*?}/pull-request/new', controller='pullrequests',
                  action='create', conditions=dict(function=check_repo,
                                                   method=["POST"]))
     rmap.connect('pullrequest_show',
                  '/{repo_name:.*?}/pull-request/{pull_request_id:\\d+}{extra:(/.*)?}', extra='',
                  controller='pullrequests',
                  action='show', conditions=dict(function=check_repo,
                                                 method=["GET"]))
     rmap.connect('pullrequest_post',
                  '/{repo_name:.*?}/pull-request/{pull_request_id}',
                  controller='pullrequests',
                  action='post', conditions=dict(function=check_repo,
                                                 method=["POST"]))

kallithea/controllers/login.py

➞

Show inline comments

@@ @@ -60,199 +60,199 @@ class LoginController(BaseController): @@
         Determines if a URI reference is valid and relative to the origin;
         or in RFC 3986 terms, whether it matches this production:
           origin-relative-ref = path-absolute [ "?" query ] [ "#" fragment ]
         with the exception that '%' escapes are not validated and '#' is
         allowed inside the fragment part.
         """
         return _re.match(came_from) is not None
     def index(self):
         c.came_from = safe_str(request.GET.get('came_from', ''))
         if c.came_from:
             if not self._validate_came_from(c.came_from):
                 log.error('Invalid came_from (not server-relative): %r', c.came_from)
                 raise HTTPBadRequest()
         else:
             c.came_from = url('home')
         if request.POST:
             # import Login Form validator class
             login_form = LoginForm()()
             try:
                 c.form_result = login_form.to_python(dict(request.POST))
                 # form checks for username/password, now we're authenticated
                 username = c.form_result['username']
                 user = User.get_by_username_or_email(username, case_insensitive=True)
             except formencode.Invalid as errors:
                 defaults = errors.value
                 # remove password from filling in form again
                 defaults.pop('password', None)
                 return htmlfill.render(
                     render('/login.html'),
                     defaults=errors.value,
                     errors=errors.error_dict or {},
                     prefix_error=False,
                     encoding="UTF-8",
                     force_defaults=False)
             except UserCreationError as e:
                 # container auth or other auth functions that create users on
                 # the fly can throw this exception signaling that there's issue
                 # with user creation, explanation should be provided in
                 # Exception itself
                 h.flash(e, 'error')
             else:
                 auth_user = log_in_user(user, c.form_result['remember'], is_external_auth=False, ip_addr=request.ip_addr)
                 # TODO: handle auth_user is None as failed authentication?
                 raise HTTPFound(location=c.came_from)
         else:
             # redirect if already logged in
             if not request.authuser.is_anonymous:
                 raise HTTPFound(location=c.came_from)
             # continue to show login to default user
         return render('/login.html')
     @HasPermissionAnyDecorator('hg.admin', 'hg.register.auto_activate',
                                'hg.register.manual_activate')
     def register(self):
         def_user_perms = AuthUser(dbuser=User.get_default_user()).permissions['global']
         c.auto_active = 'hg.register.auto_activate' in def_user_perms
         settings = Setting.get_app_settings()
         captcha_private_key = settings.get('captcha_private_key')
         c.captcha_active = bool(captcha_private_key)
         c.captcha_public_key = settings.get('captcha_public_key')
         if request.POST:
             register_form = RegisterForm()()
             try:
                 form_result = register_form.to_python(dict(request.POST))
                 form_result['active'] = c.auto_active
                 if c.captcha_active:
                     from kallithea.lib.recaptcha import submit
                     response = submit(request.POST.get('g-recaptcha-response'),
                                       private_key=captcha_private_key,
                                       remoteip=request.ip_addr)
                     if not response.is_valid:
                         _value = form_result
                         _msg = _('Bad captcha')
                         error_dict = {'recaptcha_field': _msg}
                         raise formencode.Invalid(_msg, _value, None,
                                                  error_dict=error_dict)
                 UserModel().create_registration(form_result)
                 h.flash(_('You have successfully registered with %s') % (c.site_name or 'Kallithea'),
                         category='success')
                 Session().commit()
                 raise HTTPFound(location=url('login_home'))
             except formencode.Invalid as errors:
                 return htmlfill.render(
                     render('/register.html'),
                     defaults=errors.value,
                     errors=errors.error_dict or {},
                     prefix_error=False,
                     encoding="UTF-8",
                     force_defaults=False)
             except UserCreationError as e:
                 # container auth or other auth functions that create users on
                 # the fly can throw this exception signaling that there's issue
                 # with user creation, explanation should be provided in
                 # Exception itself
                 h.flash(e, 'error')
         return render('/register.html')
     def password_reset(self):
         settings = Setting.get_app_settings()
         captcha_private_key = settings.get('captcha_private_key')
         c.captcha_active = bool(captcha_private_key)
         c.captcha_public_key = settings.get('captcha_public_key')
         if request.POST:
             password_reset_form = PasswordResetRequestForm()()
             try:
                 form_result = password_reset_form.to_python(dict(request.POST))
                 if c.captcha_active:
                     from kallithea.lib.recaptcha import submit
                     response = submit(request.POST.get('g-recaptcha-response'),
                                       private_key=captcha_private_key,
                                       remoteip=request.ip_addr)
                     if not response.is_valid:
                         _value = form_result
                         _msg = _('Bad captcha')
                         error_dict = {'recaptcha_field': _msg}
                         raise formencode.Invalid(_msg, _value, None,
                                                  error_dict=error_dict)
                 redirect_link = UserModel().send_reset_password_email(form_result)
                 h.flash(_('A password reset confirmation code has been sent'),
                             category='success')
                 raise HTTPFound(location=redirect_link)
             except formencode.Invalid as errors:
                 return htmlfill.render(
                     render('/password_reset.html'),
                     defaults=errors.value,
                     errors=errors.error_dict or {},
                     prefix_error=False,
                     encoding="UTF-8",
                     force_defaults=False)
         return render('/password_reset.html')
     def password_reset_confirmation(self):
         # This controller handles both GET and POST requests, though we
         # only ever perform the actual password change on POST (since
         # GET requests are not allowed to have side effects, and do not
         # receive automatic CSRF protection).
         # The template needs the email address outside of the form.
         c.email = request.params.get('email')
         if not request.POST:
             return htmlfill.render(
                 render('/password_reset_confirmation.html'),
                 defaults=dict(request.params),
                 encoding='UTF-8')
         form = PasswordResetConfirmationForm()()
         try:
             form_result = form.to_python(dict(request.POST))
         except formencode.Invalid as errors:
             return htmlfill.render(
                 render('/password_reset_confirmation.html'),
                 defaults=errors.value,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding='UTF-8')
         if not UserModel().verify_reset_password_token(
             form_result['email'],
             form_result['timestamp'],
             form_result['token'],
         ):
             return htmlfill.render(
                 render('/password_reset_confirmation.html'),
                 defaults=form_result,
                 errors={'token': _('Invalid password reset token')},
                 prefix_error=False,
                 encoding='UTF-8')
         UserModel().reset_password(form_result['email'], form_result['password'])
         h.flash(_('Successfully updated password'), category='success')
         raise HTTPFound(location=url('login_home'))
     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

@@ @@ -1084,211 +1084,211 @@ def urlify_text(s, repo_name=None, link_ @@
         s = s.rstrip()
     else:
         s = truncatef(s, truncate, whole_word=True)
     s = html_escape(s)
     s = _urlify(s)
     if repo_name is not None:
         s = urlify_issues(s, repo_name)
     if link_ is not None:
         # make href around everything that isn't a href already
         s = linkify_others(s, link_)
     s = s.replace('\r\n', '<br/>').replace('\n', '<br/>')
     # Turn HTML5 into more valid HTML4 as required by some mail readers.
     # (This is not done in one step in html_escape, because character codes like
     # &#123; risk to be seen as an issue reference due to the presence of '#'.)
     s = s.replace("&apos;", "&#39;")
     return literal(s)
 def linkify_others(t, l):
     """Add a default link to html with links.
     HTML doesn't allow nesting of links, so the outer link must be broken up
     in pieces and give space for other links.
     """
     urls = re.compile(r'(\<a.*?\<\/a\>)',)
     links = []
     for e in urls.split(t):
         if e.strip() and not urls.match(e):
             links.append('<a class="message-link" href="%s">%s</a>' % (l, e))
         else:
             links.append(e)
     return ''.join(links)
 # Global variable that will hold the actual urlify_issues function body.
 # Will be set on first use when the global configuration has been read.
 _urlify_issues_f = None
 def urlify_issues(newtext, repo_name):
     """Urlify issue references according to .ini configuration"""
     global _urlify_issues_f
     if _urlify_issues_f is None:
         from kallithea import CONFIG
         from kallithea.model.db import URL_SEP
         assert CONFIG['sqlalchemy.url'] # make sure config has been loaded
         # Build chain of urlify functions, starting with not doing any transformation
         tmp_urlify_issues_f = lambda s: s
         issue_pat_re = re.compile(r'issue_pat(.*)')
         for k in CONFIG.keys():
             # Find all issue_pat* settings that also have corresponding server_link and prefix configuration
             m = issue_pat_re.match(k)
             if m is None:
                 continue
             suffix = m.group(1)
             issue_pat = CONFIG.get(k)
             issue_server_link = CONFIG.get('issue_server_link%s' % suffix)
             issue_sub = CONFIG.get('issue_sub%s' % suffix)
             if not issue_pat or not issue_server_link or issue_sub is None: # issue_sub can be empty but should be present
                 log.error('skipping incomplete issue pattern %r: %r -> %r %r', suffix, issue_pat, issue_server_link, issue_sub)
                 continue
             # Wrap tmp_urlify_issues_f with substitution of this pattern, while making sure all loop variables (and compiled regexpes) are bound
             try:
                 issue_re = re.compile(issue_pat)
             except re.error as e:
                 log.error('skipping invalid issue pattern %r: %r -> %r %r. Error: %s', suffix, issue_pat, issue_server_link, issue_sub, str(e))
                 continue
             log.debug('issue pattern %r: %r -> %r %r', suffix, issue_pat, issue_server_link, issue_sub)
             def issues_replace(match_obj,
                                issue_server_link=issue_server_link, issue_sub=issue_sub):
                 try:
                     issue_url = match_obj.expand(issue_server_link)
                 except (IndexError, re.error) as e:
                     log.error('invalid issue_url setting %r -> %r %r. Error: %s', issue_pat, issue_server_link, issue_sub, str(e))
                     issue_url = issue_server_link
                 issue_url = issue_url.replace('{repo}', repo_name)
                 issue_url = issue_url.replace('{repo_name}', repo_name.split(URL_SEP)[-1])
                 # if issue_sub is empty use the matched issue reference verbatim
                 if not issue_sub:
                     issue_text = match_obj.group()
                 else:
                     try:
                         issue_text = match_obj.expand(issue_sub)
                     except (IndexError, re.error) as e:
                         log.error('invalid issue_sub setting %r -> %r %r. Error: %s', issue_pat, issue_server_link, issue_sub, str(e))
                         issue_text = match_obj.group()
                 return (
                     '<a class="issue-tracker-link" href="%(url)s">'
                     '%(text)s'
                     '</a>'
                     ) % {
                      'url': issue_url,
                      'text': issue_text,
+                    }
             tmp_urlify_issues_f = (lambda s,
                                           issue_re=issue_re, issues_replace=issues_replace, chain_f=tmp_urlify_issues_f:
                                    issue_re.sub(issues_replace, chain_f(s)))
         # Set tmp function globally - atomically
         _urlify_issues_f = tmp_urlify_issues_f
     return _urlify_issues_f(newtext)
 def render_w_mentions(source, repo_name=None):
     """
     Render plain text with revision hashes and issue references urlified
     and with @mention highlighting.
     """
     s = safe_unicode(source)
     s = urlify_text(s, repo_name=repo_name)
     return literal('<div class="formatted-fixed">%s</div>' % s)
 def short_ref(ref_type, ref_name):
     if ref_type == 'rev':
         return short_id(ref_name)
     return ref_name
 def link_to_ref(repo_name, ref_type, ref_name, rev=None):
     """
     Return full markup for a href to changeset_home for a changeset.
     If ref_type is branch it will link to changelog.
     ref_name is shortened if ref_type is 'rev'.
     if rev is specified show it too, explicitly linking to that revision.
     """
     txt = short_ref(ref_type, ref_name)
     if ref_type == 'branch':
         u = url('changelog_home', repo_name=repo_name, branch=ref_name)
     else:
         u = url('changeset_home', repo_name=repo_name, revision=ref_name)
     l = link_to(repo_name + '#' + txt, u)
     if rev and ref_type != 'rev':
         l = literal('%s (%s)' % (l, link_to(short_id(rev), url('changeset_home', repo_name=repo_name, revision=rev))))
     return l
 def changeset_status(repo, revision):
     from kallithea.model.changeset_status import ChangesetStatusModel
     return ChangesetStatusModel().get_status(repo, revision)
 def changeset_status_lbl(changeset_status):
     from kallithea.model.db import ChangesetStatus
     return ChangesetStatus.get_status_lbl(changeset_status)
 def get_permission_name(key):
     from kallithea.model.db import Permission
     return dict(Permission.PERMS).get(key)
 def journal_filter_help():
     return _(textwrap.dedent('''
         Example filter terms:
             repository:vcs
             username:developer
             action:*push*
             ip:127.0.0.1
             date:20120101
             date:[20120101100000 TO 20120102]
         Generate wildcards using '*' character:
             "repository:vcs*" - search everything starting with 'vcs'
             "repository:*vcs*" - search for repository containing 'vcs'
         Optional AND / OR operators in queries
             "repository:vcs OR repository:test"
             "username:test AND repository:test*"
     '''))
 def not_mapped_error(repo_name):
     flash(_('%s repository is not mapped to db perhaps'
             ' it was created or renamed from the filesystem'
             ' please run the application again'
             ' in order to rescan repositories') % repo_name, category='error')
 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))
         session.save()
     return session[session_csrf_secret_name]
 def form(url, method="post", **attrs):
     """Like webhelpers.html.tags.form , but automatically adding
     session_csrf_secret_token for POST. The secret is thus never leaked in GET
     URLs.
     """
     form = insecure_form(url, method, **attrs)
     if method.lower() == 'get':
         return form
     return form + HTML.div(hidden(session_csrf_secret_name, session_csrf_secret_token()), style="display: none;")

kallithea/public/js/base.js

➞

Show inline comments

@@ @@ -219,443 +219,443 @@ var pyroutes = (function() { @@
                         case 'X': arg = arg.toString(16).toUpperCase(); break;
+                    }
                     arg = (/[def]/.test(match[8]) && match[3] && arg >= 0 ? '+'+ arg : arg);
                     pad_character = match[4] ? match[4] == '0' ? '0' : match[4].charAt(1) : ' ';
                     pad_length = match[6] - String(arg).length;
                     pad = match[6] ? str_repeat(pad_character, pad_length) : '';
                     output.push(match[5] ? arg + pad : pad + arg);
+                }
+            }
             return output.join('');
         };
         str_format.cache = {};
         str_format.parse = function(fmt) {
             var _fmt = fmt, match = [], parse_tree = [], arg_names = 0;
             while (_fmt) {
                 if ((match = /^[^\x25]+/.exec(_fmt)) !== null) {
                     parse_tree.push(match[0]);
+                }
                 else if ((match = /^\x25{2}/.exec(_fmt)) !== null) {
                     parse_tree.push('%');
+                }
                 else if ((match = /^\x25(?:([1-9]\d*)\$|\(([^\)]+)\))?(\+)?(0|'[^$])?(-)?(\d+)?(?:\.(\d+))?([b-fosuxX])/.exec(_fmt)) !== null) {
                     if (match[2]) {
                         arg_names |= 1;
                         var field_list = [], replacement_field = match[2], field_match = [];
                         if ((field_match = /^([a-z_][a-z_\d]*)/i.exec(replacement_field)) !== null) {
                             field_list.push(field_match[1]);
                             while ((replacement_field = replacement_field.substring(field_match[0].length)) !== '') {
                                 if ((field_match = /^\.([a-z_][a-z_\d]*)/i.exec(replacement_field)) !== null) {
                                     field_list.push(field_match[1]);
+                                }
                                 else if ((field_match = /^\[(\d+)\]/.exec(replacement_field)) !== null) {
                                     field_list.push(field_match[1]);
+                                }
                                 else {
                                     throw('[sprintf] huh?');
+                                }
+                            }
+                        }
                         else {
                             throw('[sprintf] huh?');
+                        }
                         match[2] = field_list;
+                    }
                     else {
                         arg_names |= 2;
+                    }
                     if (arg_names === 3) {
                         throw('[sprintf] mixing positional and named placeholders is not (yet) supported');
+                    }
                     parse_tree.push(match);
+                }
                 else {
                     throw('[sprintf] huh?');
+                }
                 _fmt = _fmt.substring(match[0].length);
+            }
             return parse_tree;
         };
         return str_format;
     })();
     var vsprintf = function(fmt, argv) {
         argv.unshift(fmt);
         return sprintf.apply(null, argv);
     };
     return {
         'url': function(route_name, params) {
             var result = route_name;
             if (typeof(params) != 'object'){
                 params = {};
+            }
             if (matchlist.hasOwnProperty(route_name)) {
                 var route = matchlist[route_name];
                 // param substitution
                 for(var i=0; i < route[1].length; i++) {
                    if (!params.hasOwnProperty(route[1][i]))
                         throw new Error(route[1][i] + ' missing in "' + route_name + '" route generation');
+                }
                 result = sprintf(route[0], params);
                 var ret = [];
                 //extra params => GET
                 for(var param in params){
                     if (route[1].indexOf(param) == -1){
                         ret.push(encodeURIComponent(param) + "=" + encodeURIComponent(params[param]));
+                    }
+                }
                 var _parts = ret.join("&");
                 if(_parts){
                     result = result +'?'+ _parts
+                }
+            }
             return result;
         },
         'register': function(route_name, route_tmpl, req_params) {
             if (typeof(req_params) != 'object') {
                 req_params = [];
+            }
             var keys = [];
             for (var i=0; i < req_params.length; i++) {
                 keys.push(req_params[i]);
+            }
             matchlist[route_name] = [
                 unescape(route_tmpl),
                 keys
+            ]
         },
         '_routes': function(){
             return matchlist;
+        }
+    }
 })();
 /* Invoke all functions in callbacks */
 var _run_callbacks = function(callbacks){
     if (callbacks !== undefined){
         var _l = callbacks.length;
         for (var i=0;i<_l;i++){
             var func = callbacks[i];
             if(typeof(func)=='function'){
                 try{
                     func();
                 }catch (err){};
+            }
+        }
+    }
+}
 /**
  * turns objects into GET query string
  */
 var _toQueryString = function(o) {
     if(typeof o !== 'object') {
         return false;
+    }
     var _p, _qs = [];
     for(_p in o) {
         _qs.push(encodeURIComponent(_p) + '=' + encodeURIComponent(o[_p]));
+    }
     return _qs.join('&');
 };
 /**
  * Load HTML into DOM using Ajax
+ *
  * @param $target: load html async and place it (or an error message) here
  * @param success: success callback function
  * @param args: query parameters to pass to url
  */
 function asynchtml(url, $target, success, args){
     if(args===undefined){
         args=null;
+    }
     $target.html(_TM['Loading ...']).css('opacity','0.3');
     return $.ajax({url: url, data: args, headers: {'X-PARTIAL-XHR': '1'}, cache: false, dataType: 'html'})
         .done(function(html) {
                 $target.html(html);
                 $target.css('opacity','1.0');
                 //execute the given original callback
                 if (success !== undefined && success) {
                     success();
+                }
             })
         .fail(function(jqXHR, textStatus, errorThrown) {
                 if (textStatus == "abort")
                     return;
                 $target.html('<span class="bg-danger">ERROR: {0}</span>'.format(textStatus));
                 $target.css('opacity','1.0');
             })
+        ;
 };
 var ajaxGET = function(url, success, failure) {
     if(failure === undefined) {
         failure = function(jqXHR, textStatus, errorThrown) {
                 if (textStatus != "abort")
                     alert("Ajax GET error: " + textStatus);
             };
+    }
     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);
             };
+    }
     return $.ajax({url: url, data: postData, type: 'POST', headers: {'X-PARTIAL-XHR': '1'}, cache: false})
         .done(success)
         .fail(failure);
 };
 /**
  * activate .show_more links
  * the .show_more must have an id that is the the id of an element to hide prefixed with _
  * the parentnode will be displayed
  */
 var show_more_event = function(){
     $('.show_more').click(function(e){
         var el = e.currentTarget;
         $('#' + el.id.substring(1)).hide();
         $(el.parentNode).show();
     });
 };
 var _onSuccessFollow = function(target){
     var $target = $(target);
     var $f_cnt = $('#current_followers_count');
     if ($target.hasClass('follow')) {
         $target.removeClass('follow').addClass('following');
         $target.prop('title', _TM['Stop following this repository']);
         if ($f_cnt.html()) {
             var cnt = Number($f_cnt.html())+1;
             $f_cnt.html(cnt);
+        }
     } else {
         $target.removeClass('following').addClass('follow');
         $target.prop('title', _TM['Start following this repository']);
         if ($f_cnt.html()) {
             var cnt = Number($f_cnt.html())-1;
             $f_cnt.html(cnt);
+        }
+    }
+}
 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);
             $("#" + target).addClass('loaded');
         });
+    }
     return false;
 };
 /**
  * load tooltips dynamically based on data attributes, used for .lazy-cs changeset links
  */
 var get_changeset_tooltip = function() {
     var $target = $(this);
     var tooltip = $target.data('tooltip');
     if (!tooltip) {
         var raw_id = $target.data('raw_id');
         var repo_name = $target.data('repo_name');
         var url = pyroutes.url('changeset_info', {"repo_name": repo_name, "revision": raw_id});
         $.ajax(url, {
             async: false,
             success: function(data) {
                 tooltip = data["message"];
+            }
         });
         $target.data('tooltip', tooltip);
+    }
     return tooltip;
 };
 /**
  * activate tooltips and popups
  */
 var tooltip_activate = function(){
     function placement(p, e){
         if(e.getBoundingClientRect().top > 2*$(window).height()/3){
             return 'top';
         }else{
             return 'bottom';
+        }
+    }
     $(document).ready(function(){
         $('[data-toggle="tooltip"]').tooltip({
             container: 'body',
             placement: placement
         });
         $('[data-toggle="popover"]').popover({
             html: true,
             container: 'body',
             placement: placement,
             trigger: 'hover',
             template: '<div class="popover cs-popover" role="tooltip"><div class="arrow"></div><h3 class="popover-title"></h3><div class="popover-content"></div></div>'
         });
         $('.lazy-cs').tooltip({
             title: get_changeset_tooltip,
             placement: placement
         });
     });
 };
 /**
  * Quick filter widget
+ *
  * @param target: filter input target
  * @param nodes: list of nodes in html we want to filter.
  * @param display_element function that takes current node from nodes and
  *    does hide or show based on the node
  */
 var q_filter = (function() {
     var _namespace = {};
     var namespace = function (target) {
         if (!(target in _namespace)) {
             _namespace[target] = {};
+        }
         return _namespace[target];
     };
     return function (target, $nodes, display_element) {
         var $nodes = $nodes;
         var $q_filter_field = $('#' + target);
         var F = namespace(target);
         $q_filter_field.keyup(function (e) {
             clearTimeout(F.filterTimeout);
             F.filterTimeout = setTimeout(F.updateFilter, 600);
         });
         F.filterTimeout = null;
         F.updateFilter = function () {
             // Reset timeout
             F.filterTimeout = null;
             var obsolete = [];
             var req = $q_filter_field.val().toLowerCase();
             var showing = 0;
             $nodes.each(function () {
                 var n = this;
                 var target_element = display_element(n);
                 if (req && n.innerHTML.toLowerCase().indexOf(req) == -1) {
                     $(target_element).hide();
+                }
                 else {
                     $(target_element).show();
                     showing += 1;
+                }
             });
             $('#repo_count').html(showing);
             /* FIXME: don't hardcode */
+        }
+    }
 })();
 /**
  * Comment handling
  */
 // move comments to their right location, inside new trs
 function move_comments($anchorcomments) {
     $anchorcomments.each(function(i, anchorcomment) {
         var $anchorcomment = $(anchorcomment);
         var target_id = $anchorcomment.data('target-id');
         var $comment_div = _get_add_comment_div(target_id);
         var f_path = $anchorcomment.data('f_path');
         var line_no = $anchorcomment.data('line_no');
         if ($comment_div[0]) {
             $comment_div.append($anchorcomment.children());
             if (f_path && line_no) {
                 _comment_div_append_add($comment_div, f_path, line_no);
             } else {
                 _comment_div_append_form($comment_div, f_path, line_no);
+            }
         } else {
             $anchorcomment.before("<span class='bg-warning'>Comment to {0} line {1} which is outside the diff context:</span>".format(f_path || '?', line_no || '?'));
+        }
     });
     linkInlineComments($('.firstlink'), $('.comment:first-child'));
+}
 // comment bubble was clicked - insert new tr and show form
 function show_comment_form($bubble) {
     var children = $bubble.closest('tr.line').children('[id]');
     var line_td_id = children[children.length - 1].id;
     var $comment_div = _get_add_comment_div(line_td_id);
     var f_path = $bubble.closest('[data-f_path]').data('f_path');
     var parts = line_td_id.split('_');
     var line_no = parts[parts.length-1];
     comment_div_state($comment_div, f_path, line_no, true);
+}
 // return comment div for target_id - add it if it doesn't exist yet
 function _get_add_comment_div(target_id) {
     var comments_box_id = 'comments-' + target_id;
     var $comments_box = $('#' + comments_box_id);
     if (!$comments_box.length) {
         var html = '<tr><td id="{0}" colspan="3" class="inline-comments"></td></tr>'.format(comments_box_id);
         $('#' + target_id).closest('tr').after(html);
         $comments_box = $('#' + comments_box_id);
+    }
     return $comments_box;
+}
 // Set $comment_div state - showing or not showing form and Add button.
 // An Add button is shown on non-empty forms when no form is shown.
 // The form is controlled by show_form_opt - if undefined, form is only shown for general comments.
 function comment_div_state($comment_div, f_path, line_no, show_form_opt) {
     var show_form = show_form_opt !== undefined ? show_form_opt : !f_path && !line_no;
     var $forms = $comment_div.children('.comment-inline-form');
     var $buttonrow = $comment_div.children('.add-button-row');
     var $comments = $comment_div.children('.comment:not(.submitting)');
     $forms.remove();
     $buttonrow.remove();
     if (show_form) {
         _comment_div_append_form($comment_div, f_path, line_no);
     } else if ($comments.length) {
         _comment_div_append_add($comment_div, f_path, line_no);
     } else {
         $comment_div.parent('tr').remove();
+    }
+}
 // append an Add button to $comment_div and hook it up to show form
 function _comment_div_append_add($comment_div, f_path, line_no) {
     var addlabel = TRANSLATION_MAP['Add Another Comment'];

kallithea/templates/admin/gists/edit.html

➞

Show inline comments

 ## -*- coding: utf-8 -*-
 <%inherit file="/base/base.html"/>
 <%block name="title">
     ${_('Edit Gist')} &middot; ${c.gist.gist_access_id}
 </%block>
 <%block name="js_extra">
   <script type="text/javascript" src="${h.url('/codemirror/lib/codemirror.js')}"></script>
   <script type="text/javascript" src="${h.url('/js/codemirror_loadmode.js')}"></script>
   <script type="text/javascript" src="${h.url('/codemirror/mode/meta.js')}"></script>
 </%block>
 <%block name="css_extra">
   <link rel="stylesheet" type="text/css" href="${h.url('/codemirror/lib/codemirror.css')}"/>
 </%block>
 <%def name="breadcrumbs_links()">
     ${_('Edit Gist')} &middot; ${c.gist.gist_access_id}
 </%def>
 <%block name="header_menu">
     ${self.menu('gists')}
 </%block>
 <%def name="main()">
 <div class="panel panel-primary">
     <div class="panel-heading clearfix">
         ${self.breadcrumbs()}
     </div>
     <div class="panel-body">
         <div id="edit_error" style="display: none" class="flash_msg">
             <div class="alert alert-dismissable alert-warning">
               <button type="button" class="close" data-dismiss="alert" aria-hidden="true"><i class="icon-cancel-circled"></i></button>
               ${(h.HTML(_('Gist was updated since you started editing. Copy your changes and click %(here)s to reload new version.'))
                              % {'here': h.link_to(_('here'),h.url('edit_gist', gist_id=c.gist.gist_access_id))})}
             </div>
             <script>
             if (typeof jQuery != 'undefined') {
                 $(".alert").alert();
+            }
             </script>
         </div>
         <div id="files_data">
           ${h.form(h.url('edit_gist', gist_id=c.gist.gist_access_id), method='post', id='eform')}
             <div>
                 <input type="hidden" value="${c.file_changeset.raw_id}" name="parent_hash">
                 <textarea class="form-control"
                           id="description" name="description"
                           placeholder="${_('Gist description ...')}">${c.gist.gist_description}</textarea>
                 <div>
                     <label>
                         ${_('Gist lifetime')}
                         ${h.select('lifetime', '0', c.lifetime_options)}
                     </label>
                     <span class="text-muted">
                      %if c.gist.gist_expires == -1:
                       ${_('Expires')}: ${_('Never')}
                      %else:
                       ${_('Expires')}: ${h.age(h.time_to_datetime(c.gist.gist_expires))}
                      %endif
                    </span>
                 </div>
             </div>
             % for cnt, file in enumerate(c.files):
                 <div id="body" class="panel panel-default form-inline">
                     <div class="panel-heading">
                         <input type="hidden" value="${h.safe_unicode(file.path)}" name="org_files">
                         <input class="form-control" id="filename_${h.FID('f',file.path)}" name="files" size="30" type="text" value="${h.safe_unicode(file.path)}">
                         <select class="form-control" id="mimetype_${h.FID('f',file.path)}" name="mimetypes"></select>
                     </div>
                     <div class="panel-body no-padding">
                         <div id="editor_container">
                             <textarea id="editor_${h.FID('f',file.path)}" name="contents" style="display:none">${file.content}</textarea>
                         </div>
                     </div>
                 </div>
                 ## dynamic edit box.
                 <script type="text/javascript">
                     $(document).ready(function(){
                         var myCodeMirror = initCodeMirror(${h.js('editor_' + h.FID('f',file.path))}, ${h.jshtml(request.script_name)}, '');
                         //inject new modes
                         var $mimetype_select = $(${h.js('#mimetype_' + h.FID('f',file.path))});
                         $mimetype_select.each(function(){
                             var modes_select = this;
                             var index = 1;
                             for(var i=0;i<CodeMirror.modeInfo.length;i++) {
                                 var m = CodeMirror.modeInfo[i];
                                 var opt = new Option(m.name, m.mime);
                                 $(opt).attr('mode', m.mode);
                                 if (m.mime == 'text/plain') {
                                     // default plain text
                                     $(opt).prop('selected', true);
                                     modes_select.options[0] = opt;
                                 } else {
                                     modes_select.options[index++] = opt;
+                                }
+                            }
                         });
                         var $filename_input = $(${h.js('#filename_' + h.FID('f',file.path))});
                         // on select change set new mode
                         $mimetype_select.change(function(e){
                             var selected = e.currentTarget;
                             var node = selected.options[selected.selectedIndex];
                             var detected_mode = CodeMirror.findModeByMIME(node.value);
                             setCodeMirrorMode(myCodeMirror, detected_mode);
                             var proposed_ext = CodeMirror.findExtensionByMode(detected_mode);
                             var file_data = CodeMirror.getFilenameAndExt($filename_input.val());
                             var filename = file_data['filename'] || 'filename1';
                             $filename_input.val(filename + '.' + proposed_ext);
                         });
                         // on type the new filename set mode
                         $filename_input.keyup(function(e){
                             var file_data = CodeMirror.getFilenameAndExt(this.value);
                             if(file_data['ext'] != null){
                                 var detected_mode = CodeMirror.findModeByExtension(file_data['ext']) || CodeMirror.findModeByMIME('text/plain');
                                 if (detected_mode){
                                     setCodeMirrorMode(myCodeMirror, detected_mode);
                                     $mimetype_select.val(detected_mode.mime);
+                                }
+                            }
                         });
                         // set mode on page load
                         var detected_mode = CodeMirror.findModeByExtension(${h.js(file.extension)});
                         if (detected_mode){
                             setCodeMirrorMode(myCodeMirror, detected_mode);
                             $mimetype_select.val(detected_mode.mime);
+                        }
                     });
                 </script>
             %endfor
             <div>
             ${h.submit('update',_('Update Gist'),class_="btn btn-success")}
             <a class="btn btn-default" href="${h.url('gist', gist_id=c.gist.gist_access_id)}">${_('Cancel')}</a>
             </div>
           ${h.end_form()}
           <script>
               $('#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();
+                      }
                       else{
                         $('#eform').submit();
+                      }
+                    }
                   });
               });
           </script>
         </div>
     </div>
 </div>
 </%def>

kallithea/tests/base.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 import datetime
 import logging
 import os
 import pytest
 import re
 import tempfile
 import time
 from tg import config
 from webtest import TestApp
 from kallithea import model
 from kallithea.model.db import User
 from kallithea.model.meta import Session
 from kallithea.lib.utils2 import safe_str
 log = logging.getLogger(__name__)
 skipif = pytest.mark.skipif
 parametrize = pytest.mark.parametrize
 # Hack: These module global values MUST be set to actual values before running any tests. This is currently done by conftest.py.
 url = None
 testapp = None
 __all__ = [
     'skipif', 'parametrize', 'url', 'TestController',
     'ldap_lib_installed', 'pam_lib_installed', 'invalidate_all_caches',
     'TESTS_TMP_PATH', 'HG_REPO', 'GIT_REPO', 'NEW_HG_REPO', 'NEW_GIT_REPO',
     'HG_FORK', 'GIT_FORK', 'TEST_USER_ADMIN_LOGIN', 'TEST_USER_ADMIN_PASS',
     'TEST_USER_ADMIN_EMAIL', 'TEST_USER_REGULAR_LOGIN', 'TEST_USER_REGULAR_PASS',
     'TEST_USER_REGULAR_EMAIL', 'TEST_USER_REGULAR2_LOGIN',
     'TEST_USER_REGULAR2_PASS', 'TEST_USER_REGULAR2_EMAIL', 'IP_ADDR',
     'TEST_HG_REPO', 'TEST_HG_REPO_CLONE', 'TEST_HG_REPO_PULL', 'TEST_GIT_REPO',
     'TEST_GIT_REPO_CLONE', 'TEST_GIT_REPO_PULL', 'HG_REMOTE_REPO',
     'GIT_REMOTE_REPO', 'HG_TEST_REVISION', 'GIT_TEST_REVISION',
+]
 ## SOME GLOBALS FOR TESTS
 TESTS_TMP_PATH = os.environ.get('KALLITHEA_TESTS_TMP_PATH', tempfile.mkdtemp(prefix='kallithea-test-'))
 TEST_USER_ADMIN_LOGIN = 'test_admin'
 TEST_USER_ADMIN_PASS = 'test12'
 TEST_USER_ADMIN_EMAIL = 'test_admin@example.com'
 TEST_USER_REGULAR_LOGIN = 'test_regular'
 TEST_USER_REGULAR_PASS = 'test12'
 TEST_USER_REGULAR_EMAIL = 'test_regular@example.com'
 TEST_USER_REGULAR2_LOGIN = 'test_regular2'
 TEST_USER_REGULAR2_PASS = 'test12'
 TEST_USER_REGULAR2_EMAIL = 'test_regular2@example.com'
 IP_ADDR = '127.0.0.127'
 HG_REPO = u'vcs_test_hg'
 GIT_REPO = u'vcs_test_git'
 NEW_HG_REPO = u'vcs_test_hg_new'
 NEW_GIT_REPO = u'vcs_test_git_new'
 HG_FORK = u'vcs_test_hg_fork'
 GIT_FORK = u'vcs_test_git_fork'
 HG_TEST_REVISION = u"a53d9201d4bc278910d416d94941b7ea007ecd52"
 GIT_TEST_REVISION = u"7ab37bc680b4aa72c34d07b230c866c28e9fc204"
 ## VCS
 uniq_suffix = str(int(time.mktime(datetime.datetime.now().timetuple())))
 GIT_REMOTE_REPO = os.path.join(TESTS_TMP_PATH, GIT_REPO)
 TEST_GIT_REPO = os.path.join(TESTS_TMP_PATH, GIT_REPO)
 TEST_GIT_REPO_CLONE = os.path.join(TESTS_TMP_PATH, 'vcs-git-clone-%s' % uniq_suffix)
 TEST_GIT_REPO_PULL = os.path.join(TESTS_TMP_PATH, 'vcs-git-pull-%s' % uniq_suffix)
 HG_REMOTE_REPO = os.path.join(TESTS_TMP_PATH, HG_REPO)
 TEST_HG_REPO = os.path.join(TESTS_TMP_PATH, HG_REPO)
 TEST_HG_REPO_CLONE = os.path.join(TESTS_TMP_PATH, 'vcs-hg-clone-%s' % uniq_suffix)
 TEST_HG_REPO_PULL = os.path.join(TESTS_TMP_PATH, 'vcs-hg-pull-%s' % uniq_suffix)
 # By default, some of the tests will utilise locally available
 # repositories stored within tar.gz archives as source for
 # cloning. Should you wish to use some other, remote archive, simply
 # uncomment these entries and/or update the URLs to use.
+#
 # GIT_REMOTE_REPO = 'git://github.com/codeinn/vcs.git'
 # HG_REMOTE_REPO = 'http://bitbucket.org/marcinkuzminski/vcs'
 # skip ldap tests if LDAP lib is not installed
 ldap_lib_installed = False
 try:
     import ldap
     ldap.API_VERSION
     ldap_lib_installed = True
 except ImportError:
     # means that python-ldap is not installed
     pass
 try:
     import pam
     pam.PAM_TEXT_INFO
     pam_lib_installed = True
 except ImportError:
     pam_lib_installed = False
 def invalidate_all_caches():
     """Invalidate all beaker caches currently configured.
     Useful when manipulating IP permissions in a test and changes need to take
     effect immediately.
     Note: Any use of this function is probably a workaround - it should be
     replaced with a more specific cache invalidation in code or test."""
     from beaker.cache import cache_managers
     for cache in cache_managers.values():
         cache.clear()
 class NullHandler(logging.Handler):
     def emit(self, record):
         pass
 class TestController(object):
     """Pytest-style test controller"""
     # Note: pytest base classes cannot have an __init__ method
     @pytest.fixture(autouse=True)
     def app_fixture(self):
         h = NullHandler()
         logging.getLogger("kallithea").addHandler(h)
         self.app = TestApp(testapp)
         return self.app
     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)
         response = response.follow()
         return response.session['authuser']
     def _get_logged_user(self):
         return User.get_by_username(self._logged_username)
     def assert_authenticated_user(self, response, expected_username):
         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]
             if _matcher(msg, m):
                 return
         except IndexError:
             pass
         pytest.fail(safe_str(u'msg `%s` not found in session flash (skipping %s): %s' %
                            (msg, skip,
                             ', '.join('`%s`' % m for level, m in response.session['flash']))))
     def checkSessionFlashRegex(self, response, regex, skip=0):
         self.checkSessionFlash(response, regex, skip=skip, _matcher=re.search)

kallithea/tests/functional/test_admin_auth_settings.py

➞

Show inline comments

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

kallithea/tests/functional/test_admin_defaults.py

➞

Show inline comments

 from kallithea.tests.base import *
 from kallithea.model.db import Setting
 class TestDefaultsController(TestController):
     def test_index(self):
         self.log_user()
         response = self.app.get(url('defaults'))
         response.mustcontain('default_repo_private')
         response.mustcontain('default_repo_enable_statistics')
         response.mustcontain('default_repo_enable_downloads')
     def test_update_params_true_hg(self):
         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

 from kallithea.tests.base import *
 from kallithea.model.gist import GistModel
 from kallithea.model.meta import Session
 from kallithea.model.db import User, Gist
 def _create_gist(f_name, content='some gist', lifetime=-1,
                  description=u'gist-desc', gist_type='public',
                  owner=TEST_USER_ADMIN_LOGIN):
     gist_mapping = {
         f_name: {'content': content}
+    }
     owner = User.get_by_username(owner)
     gist = GistModel().create(description, owner=owner, ip_addr=IP_ADDR,
                        gist_mapping=gist_mapping, gist_type=gist_type,
                        lifetime=lifetime)
     Session().commit()
     return gist
 class TestGistsController(TestController):
     def teardown_method(self, method):
         for g in Gist.query():
             GistModel().delete(g)
         Session().commit()
     def test_index(self):
         self.log_user()
         response = self.app.get(url('gists'))
         # Test response...
         response.mustcontain('There are no gists yet')
         g1 = _create_gist('gist1').gist_access_id
         g2 = _create_gist('gist2', lifetime=1400).gist_access_id
         g3 = _create_gist('gist3', description=u'gist3-desc').gist_access_id
         g4 = _create_gist('gist4', gist_type='private').gist_access_id
         response = self.app.get(url('gists'))
         # Test response...
         response.mustcontain('gist: %s' % g1)
         response.mustcontain('gist: %s' % g2)
         response.mustcontain('Expires: in 23 hours')  # we don't care about the end
         response.mustcontain('gist: %s' % g3)
         response.mustcontain('gist3-desc')
         response.mustcontain(no=['gist: %s' % g4])
     def test_index_private_gists(self):
         self.log_user()
         gist = _create_gist('gist5', gist_type='private')
         response = self.app.get(url('gists', private=1))
         # Test response...
         # and privates
         response.mustcontain('gist: %s' % gist.gist_access_id)
     def test_create_missing_description(self):
         self.log_user()
         response = self.app.post(url('gists'),
-                                 params={'lifetime': -1, '_authentication_token': self.authentication_token()},
+                                 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>')
     def test_create_with_path_with_dirs(self):
         self.log_user()
         response = self.app.post(url('gists'),
                                  params={'lifetime': -1,
                                          'content': 'gist test',
                                          'filename': '/home/foo',
                                          'public': 'public',
-                                         '_authentication_token': self.authentication_token()},
+                                         '_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')
         gist.gist_expires = 0  # 1970
         Session().commit()
         response = self.app.get(url('gist', gist_id=gist.gist_access_id), status=404)
     def test_create_private(self):
         self.log_user()
         response = self.app.post(url('gists'),
                                  params={'lifetime': -1,
                                          'content': 'private gist test',
                                          'filename': 'private-foo',
                                          'private': 'private',
-                                         '_authentication_token': self.authentication_token()},
+                                         '_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>')
     def test_create_with_description(self):
         self.log_user()
         response = self.app.post(url('gists'),
                                  params={'lifetime': -1,
                                          'content': 'gist test',
                                          'filename': 'foo-desc',
                                          'description': 'gist-desc',
                                          'public': 'public',
-                                         '_authentication_token': self.authentication_token()},
+                                         '_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>')
     def test_new(self):
         self.log_user()
         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)
         response.mustcontain('gist-desc')
         response.mustcontain('<div class="label label-success">Public Gist</div>')
     def test_show_as_raw(self):
         gist = _create_gist('gist-show-me', content='GIST CONTENT')
         response = self.app.get(url('formatted_gist',
                                     gist_id=gist.gist_access_id, format='raw'))
         assert response.body == 'GIST CONTENT'
     def test_show_as_raw_individual_file(self):
         gist = _create_gist('gist-show-me-raw', content='GIST BODY')
         response = self.app.get(url('formatted_gist_file',
                                     gist_id=gist.gist_access_id, format='raw',
                                     revision='tip', f_path='gist-show-me-raw'))
         assert response.body == 'GIST BODY'
     def test_edit(self):
         response = self.app.get(url('edit_gist', gist_id=1))

kallithea/tests/functional/test_admin_permissions.py

➞

Show inline comments

 import time
 from kallithea.model.db import User, UserIpMap
 from kallithea.model.user import UserModel
 from kallithea.model.meta import Session
 from kallithea.tests.base import *
 from tg.util.webtest import test_context
 class TestAdminPermissionsController(TestController):
     def test_index(self):
         self.log_user()
         response = self.app.get(url('admin_permissions'))
         # Test response...
     def test_index_ips(self):
         self.log_user()
         response = self.app.get(url('admin_permissions_ips'))
         # Test response...
         response.mustcontain('All IP addresses are allowed')
     def test_add_delete_ips(self, auto_clear_ip_permissions):
         self.log_user()
         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')
         response = self.app.get(url('admin_permissions_ips'),
                                 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):
         self.log_user()
         response = self.app.get(url('admin_permissions_perms'))
         # Test response...
     def test_edit_permissions_permissions(self):
         user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
         # Test unauthenticated access - it will redirect to login page
         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()
         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

 from kallithea.model.db import Repository
 from kallithea.model.meta import Session
 from kallithea.model.repo_group import RepoGroupModel
 from kallithea.tests.base import TestController, url
 from kallithea.tests.fixture import Fixture
 fixture = Fixture()
 class TestRepoGroupsController(TestController):
     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

 # -*- coding: utf-8 -*-
 import os
 import mock
 import urllib
 import pytest
 from sqlalchemy import func
 from kallithea.lib import vcs
 from kallithea.lib.utils2 import safe_str, safe_unicode
 from kallithea.model.db import Repository, RepoGroup, UserRepoToPerm, User, \
     Permission, Ui
 from kallithea.model.user import UserModel
 from kallithea.tests.base import *
 from kallithea.model.repo_group import RepoGroupModel
 from kallithea.model.repo import RepoModel
 from kallithea.model.meta import Session, Base
 from kallithea.tests.fixture import Fixture, error_function
 fixture = Fixture()
 def _get_permission_for_user(user, repo):
     perm = UserRepoToPerm.query() \
                 .filter(UserRepoToPerm.repository ==
                         Repository.get_by_repo_name(repo)) \
                 .filter(UserRepoToPerm.user == User.get_by_username(user)) \
                 .all()
     return perm
 class _BaseTestCase(TestController):
     """
     Write all tests here
     """
     REPO = None
     REPO_TYPE = None
     NEW_REPO = None
     OTHER_TYPE_REPO = None
     OTHER_TYPE = None
     def test_index(self):
         self.log_user()
         response = self.app.get(url('repos'))
     def test_create(self):
         self.log_user()
         repo_name = self.NEW_REPO
         description = u'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
-                                                _authentication_token=self.authentication_token()))
+                                                _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))
         # test if the repo was created in the database
         new_repo = Session().query(Repository) \
             .filter(Repository.repo_name == repo_name).one()
         assert new_repo.repo_name == repo_name
         assert new_repo.description == description
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=repo_name))
         response.mustcontain(repo_name)
         response.mustcontain(self.REPO_TYPE)
         # test if the repository was created on filesystem
         try:
             vcs.get_repo(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name)))
         except vcs.exceptions.VCSError:
             pytest.fail('no repo %s in filesystem' % repo_name)
         RepoModel().delete(repo_name)
         Session().commit()
     def test_case_insensitivity(self):
         self.log_user()
         repo_name = self.NEW_REPO
         description = u'description for newly created repo'
         response = self.app.post(url('repos'),
                                  fixture._get_repo_create_params(repo_private=False,
                                                                  repo_name=repo_name,
                                                                  repo_type=self.REPO_TYPE,
                                                                  repo_description=description,
-                                                                 _authentication_token=self.authentication_token()))
+                                                                 _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):
         self.log_user()
         ## create GROUP
         group_name = u'sometest_%s' % self.REPO_TYPE
         gr = RepoGroupModel().create(group_name=group_name,
                                      group_description=u'test',
                                      owner=TEST_USER_ADMIN_LOGIN)
         Session().commit()
         repo_name = u'ingroup'
         repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])
         description = u'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 repo_group=gr.group_id,
-                                                _authentication_token=self.authentication_token()))
+                                                _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))
         # test if the repo was created in the database
         new_repo = Session().query(Repository) \
             .filter(Repository.repo_name == repo_name_full).one()
         new_repo_id = new_repo.repo_id
         assert new_repo.repo_name == repo_name_full
         assert new_repo.description == description
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=repo_name_full))
         response.mustcontain(repo_name_full)
         response.mustcontain(self.REPO_TYPE)
         inherited_perms = UserRepoToPerm.query() \
             .filter(UserRepoToPerm.repository_id == new_repo_id).all()
         assert len(inherited_perms) == 1
         # test if the repository was created on filesystem
         try:
             vcs.get_repo(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name_full)))
         except vcs.exceptions.VCSError:
             RepoGroupModel().delete(group_name)
             Session().commit()
             pytest.fail('no repo %s in filesystem' % repo_name)
         RepoModel().delete(repo_name_full)
         RepoGroupModel().delete(group_name)
         Session().commit()
     def test_create_in_group_without_needed_permissions(self):
         usr = self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
         # avoid spurious RepoGroup DetachedInstanceError ...
-        authentication_token = self.authentication_token()
+        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')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.fork.none')
         # disable on regular user
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.none')
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.none')
         Session().commit()
         ## create GROUP
         group_name = u'reg_sometest_%s' % self.REPO_TYPE
         gr = RepoGroupModel().create(group_name=group_name,
                                      group_description=u'test',
                                      owner=TEST_USER_ADMIN_LOGIN)
         Session().commit()
         group_name_allowed = u'reg_sometest_allowed_%s' % self.REPO_TYPE
         gr_allowed = RepoGroupModel().create(group_name=group_name_allowed,
                                      group_description=u'test',
                                      owner=TEST_USER_REGULAR_LOGIN)
         Session().commit()
         repo_name = u'ingroup'
         repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])
         description = u'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 repo_group=gr.group_id,
-                                                _authentication_token=authentication_token))
+                                                _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])
         description = u'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 repo_group=gr_allowed.group_id,
-                                                _authentication_token=authentication_token))
+                                                _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>'
                                % (repo_name_full, repo_name_full))
         # test if the repo was created in the database
         new_repo = Session().query(Repository) \
             .filter(Repository.repo_name == repo_name_full).one()
         new_repo_id = new_repo.repo_id
         assert new_repo.repo_name == repo_name_full
         assert new_repo.description == description
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=repo_name_full))
         response.mustcontain(repo_name_full)
         response.mustcontain(self.REPO_TYPE)
         inherited_perms = UserRepoToPerm.query() \
             .filter(UserRepoToPerm.repository_id == new_repo_id).all()
         assert len(inherited_perms) == 1
         # test if the repository was created on filesystem
         try:
             vcs.get_repo(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name_full)))
         except vcs.exceptions.VCSError:
             RepoGroupModel().delete(group_name)
             Session().commit()
             pytest.fail('no repo %s in filesystem' % repo_name)
         RepoModel().delete(repo_name_full)
         RepoGroupModel().delete(group_name)
         RepoGroupModel().delete(group_name_allowed)
         Session().commit()
     def test_create_in_group_inherit_permissions(self):
         self.log_user()
         ## create GROUP
         group_name = u'sometest_%s' % self.REPO_TYPE
         gr = RepoGroupModel().create(group_name=group_name,
                                      group_description=u'test',
                                      owner=TEST_USER_ADMIN_LOGIN)
         perm = Permission.get_by_key('repository.write')
         RepoGroupModel().grant_user_permission(gr, TEST_USER_REGULAR_LOGIN, perm)
         ## add repo permissions
         Session().commit()
         repo_name = u'ingroup_inherited_%s' % self.REPO_TYPE
         repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])
         description = u'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 repo_group=gr.group_id,
                                                 repo_copy_permissions=True,
-                                                _authentication_token=self.authentication_token()))
+                                                _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))
         # test if the repo was created in the database
         new_repo = Session().query(Repository) \
             .filter(Repository.repo_name == repo_name_full).one()
         new_repo_id = new_repo.repo_id
         assert new_repo.repo_name == repo_name_full
         assert new_repo.description == description
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=repo_name_full))
         response.mustcontain(repo_name_full)
         response.mustcontain(self.REPO_TYPE)
         # test if the repository was created on filesystem
         try:
             vcs.get_repo(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name_full)))
         except vcs.exceptions.VCSError:
             RepoGroupModel().delete(group_name)
             Session().commit()
             pytest.fail('no repo %s in filesystem' % repo_name)
         # check if inherited permissiona are applied
         inherited_perms = UserRepoToPerm.query() \
             .filter(UserRepoToPerm.repository_id == new_repo_id).all()
         assert len(inherited_perms) == 2
         assert TEST_USER_REGULAR_LOGIN in [x.user.username
                                                     for x in inherited_perms]
         assert 'repository.write' in [x.permission.permission_name
                                                for x in inherited_perms]
         RepoModel().delete(repo_name_full)
         RepoGroupModel().delete(group_name)
         Session().commit()
     def test_create_remote_repo_wrong_clone_uri(self):
         self.log_user()
         repo_name = self.NEW_REPO
         description = u'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 clone_uri='http://127.0.0.1/repo',
-                                                _authentication_token=self.authentication_token()))
+                                                _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
         new_repo = Session().query(Repository) \
             .filter(Repository.repo_name == repo_name).one()
         assert new_repo.repo_name == repo_name
         assert new_repo.description == description
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=repo_name))
         response.mustcontain(repo_name)
         response.mustcontain(self.REPO_TYPE)
         # test if the repository was created on filesystem
         try:
             vcs.get_repo(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name)))
         except vcs.exceptions.VCSError:
             pytest.fail('no repo %s in filesystem' % repo_name)
         response = self.app.post(url('delete_repo', repo_name=repo_name),
-            params={'_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
         deleted_repo = Session().query(Repository) \
             .filter(Repository.repo_name == repo_name).scalar()
         assert deleted_repo is None
         assert os.path.isdir(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name)) == False
     def test_delete_non_ascii(self):
         self.log_user()
         non_ascii = "ąęł"
         repo_name = "%s%s" % (safe_str(self.NEW_REPO), non_ascii)
         repo_name_unicode = safe_unicode(repo_name)
         description = 'description for newly created repo' + non_ascii
         description_unicode = safe_unicode(description)
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
-                                                _authentication_token=self.authentication_token()))
+                                                _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))
         # test if the repo was created in the database
         new_repo = Session().query(Repository) \
             .filter(Repository.repo_name == repo_name_unicode).one()
         assert new_repo.repo_name == repo_name_unicode
         assert new_repo.description == description_unicode
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=repo_name))
         response.mustcontain(repo_name)
         response.mustcontain(self.REPO_TYPE)
         # test if the repository was created on filesystem
         try:
             vcs.get_repo(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name_unicode)))
         except vcs.exceptions.VCSError:
             pytest.fail('no repo %s in filesystem' % repo_name)
         response = self.app.post(url('delete_repo', repo_name=repo_name),
-            params={'_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()
         assert deleted_repo is None
         assert os.path.isdir(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name_unicode)) == False
     def test_delete_repo_with_group(self):
         # TODO:
         pass
     def test_delete_browser_fakeout(self):
         response = self.app.post(url('delete_repo', repo_name=self.REPO),
-                                 params=dict(_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):
         response = self.app.get(url('edit_repo', repo_name=self.REPO))
     def test_set_private_flag_sets_default_to_none(self):
         self.log_user()
         # initially repository perm should be read
         perm = _get_permission_for_user(user='default', repo=self.REPO)
         assert len(perm), 1
         assert perm[0].permission.permission_name == 'repository.read'
         assert Repository.get_by_repo_name(self.REPO).private == False
         response = self.app.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)
         assert len(perm), 1
         assert perm[0].permission.permission_name == 'repository.none'
         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)
         assert len(perm), 1
         assert perm[0].permission.permission_name == 'repository.none'
         # update this permission back
         perm[0].permission = Permission.get_by_key('repository.read')
         Session().commit()
     def test_set_repo_fork_has_no_self_id(self):
         self.log_user()
         repo = Repository.get_by_repo_name(self.REPO)
         response = self.app.get(url('edit_repo_advanced', repo_name=self.REPO))
         opt = """<option value="%s">%s</option>""" % (repo.repo_id, self.REPO)
         response.mustcontain(no=[opt])
     def test_set_fork_of_other_repo(self):
         self.log_user()
         other_repo = u'other_%s' % self.REPO_TYPE
         fixture.create_repo(other_repo, repo_type=self.REPO_TYPE)
         repo = Repository.get_by_repo_name(self.REPO)
         repo2 = Repository.get_by_repo_name(other_repo)
         response = self.app.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
         response = response.follow()
         # check if given repo is selected
         opt = """<option value="%s" selected="selected">%s</option>""" % (
                     repo2.repo_id, repo2.repo_name)
         response.mustcontain(opt)
         fixture.destroy_repo(other_repo, forks='detach')
     def test_set_fork_of_other_type_repo(self):
         self.log_user()
         repo = Repository.get_by_repo_name(self.REPO)
         repo2 = Repository.get_by_repo_name(self.OTHER_TYPE_REPO)
         response = self.app.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
         user_model = UserModel()
         # disable fork and create on default user
         user_model.revoke_perm(User.DEFAULT_USER, 'hg.create.repository')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.create.none')
         user_model.revoke_perm(User.DEFAULT_USER, 'hg.fork.repository')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.fork.none')
         # disable on regular user
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.none')
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.none')
         Session().commit()
         user = User.get(usr['user_id'])
         repo_name = self.NEW_REPO + u'no_perms'
         description = 'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
-                                                _authentication_token=self.authentication_token()))
+                                                _session_csrf_secret_token=self.session_csrf_secret_token()))
         response.mustcontain('<span class="error-message">Invalid value</span>')
         RepoModel().delete(repo_name)
         Session().commit()
     @mock.patch.object(RepoModel, '_create_filesystem_repo', error_function)
     def test_create_repo_when_filesystem_op_fails(self):
         self.log_user()
         repo_name = self.NEW_REPO
         description = 'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
-                                                _authentication_token=self.authentication_token()))
+                                                _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
         # repo must not be in filesystem !
         assert not os.path.isdir(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name))
 class TestAdminReposControllerGIT(_BaseTestCase):
     REPO = GIT_REPO
     REPO_TYPE = 'git'
     NEW_REPO = NEW_GIT_REPO
     OTHER_TYPE_REPO = HG_REPO
     OTHER_TYPE = 'hg'
 class TestAdminReposControllerHG(_BaseTestCase):
     REPO = HG_REPO
     REPO_TYPE = 'hg'
     NEW_REPO = NEW_HG_REPO
     OTHER_TYPE_REPO = GIT_REPO
     OTHER_TYPE = 'git'
     def test_permanent_url_protocol_access(self):
         repo = Repository.get_by_repo_name(self.REPO)
         permanent_name = '_%d' % repo.repo_id
         # 400 Bad Request - Unable to detect pull/push action
         self.app.get(url('summary_home', repo_name=permanent_name),
             extra_environ={'HTTP_ACCEPT': 'application/mercurial'},
             status=400,
+        )

kallithea/tests/functional/test_admin_settings.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 from kallithea.model.db import Setting, Ui
 from kallithea.tests.base import *
 from kallithea.tests.fixture import Fixture
 fixture = Fixture()
 class TestAdminSettingsController(TestController):
     def test_index_main(self):
         self.log_user()
         response = self.app.get(url('admin_settings'))
     def test_index_mapping(self):
         self.log_user()
         response = self.app.get(url('admin_settings_mapping'))
     def test_index_global(self):
         self.log_user()
         response = self.app.get(url('admin_settings_global'))
     def test_index_visual(self):
         self.log_user()
         response = self.app.get(url('admin_settings_visual'))
     def test_index_email(self):
         self.log_user()
         response = self.app.get(url('admin_settings_email'))
     def test_index_hooks(self):
         self.log_user()
         response = self.app.get(url('admin_settings_hooks'))
     def test_create_custom_hook(self):
         self.log_user()
         response = self.app.post(url('admin_settings_hooks'),
                                 params=dict(new_hook_ui_key='test_hooks_1',
                                             new_hook_ui_value='cd %s' % 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')
     def test_index_search(self):
         self.log_user()
         response = self.app.get(url('admin_settings_search'))
     def test_index_system(self):
         self.log_user()
         response = self.app.get(url('admin_settings_system'))
     def test_ga_code_active(self):
         self.log_user()
         old_title = 'Kallithea'
         old_realm = 'Kallithea authentication'
         new_ga_code = 'ga-test-123456789'
         response = self.app.post(url('admin_settings_global'),
                         params=dict(title=old_title,
                                  realm=old_realm,
                                  ga_code=new_ga_code,
                                  captcha_private_key='',
                                  captcha_public_key='',
-                                 _authentication_token=self.authentication_token(),
+                                 _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()
         response.mustcontain("""_gaq.push(['_setAccount', '%s']);""" % new_ga_code)
     def test_ga_code_inactive(self):
         self.log_user()
         old_title = 'Kallithea'
         old_realm = 'Kallithea authentication'
         new_ga_code = ''
         response = self.app.post(url('admin_settings_global'),
                         params=dict(title=old_title,
                                  realm=old_realm,
                                  ga_code=new_ga_code,
                                  captcha_private_key='',
                                  captcha_public_key='',
-                                 _authentication_token=self.authentication_token(),
+                                 _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()
         response.mustcontain(no=["_gaq.push(['_setAccount', '%s']);" % new_ga_code])
     def test_captcha_activate(self):
         self.log_user()
         old_title = 'Kallithea'
         old_realm = 'Kallithea authentication'
         new_ga_code = ''
         response = self.app.post(url('admin_settings_global'),
                         params=dict(title=old_title,
                                  realm=old_realm,
                                  ga_code=new_ga_code,
                                  captcha_private_key='1234567890',
                                  captcha_public_key='1234567890',
-                                 _authentication_token=self.authentication_token(),
+                                 _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'))
         response.mustcontain('captcha')
     def test_captcha_deactivate(self):
         self.log_user()
         old_title = 'Kallithea'
         old_realm = 'Kallithea authentication'
         new_ga_code = ''
         response = self.app.post(url('admin_settings_global'),
                         params=dict(title=old_title,
                                  realm=old_realm,
                                  ga_code=new_ga_code,
                                  captcha_private_key='',
                                  captcha_public_key='1234567890',
-                                 _authentication_token=self.authentication_token(),
+                                 _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'))
         response.mustcontain(no=['captcha'])
     def test_title_change(self):
         self.log_user()
         old_title = 'Kallithea'
         new_title = old_title + '_changed'
         old_realm = 'Kallithea authentication'
         for new_title in ['Changed', 'Żółwik', old_title]:
             response = self.app.post(url('admin_settings_global'),
                         params=dict(title=new_title,
                                  realm=old_realm,
                                  ga_code='',
                                  captcha_private_key='',
                                  captcha_public_key='',
-                                 _authentication_token=self.authentication_token(),
+                                 _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()
             response.mustcontain("""<span class="branding">%s</span>""" % new_title)

kallithea/tests/functional/test_admin_user_groups.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 from kallithea.tests.base import *
 from kallithea.model.db import UserGroup, UserGroupToPerm, Permission
 from kallithea.model.meta import Session
 TEST_USER_GROUP = u'admins_test'
 class TestAdminUsersGroupsController(TestController):
     def test_index(self):
         self.log_user()
         response = self.app.get(url('users_groups'))
         # Test response...
     def test_create(self):
         self.log_user()
         users_group_name = TEST_USER_GROUP
         response = self.app.post(url('users_groups'),
                                  {'users_group_name': users_group_name,
                                   'user_group_description': 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)
     def test_new(self):
         response = self.app.get(url('new_users_group'))
     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
     def test_default_perms_enable_repository_read_on_group(self):
         self.log_user()
         users_group_name = TEST_USER_GROUP + 'another2'
         response = self.app.post(url('users_groups'),
                                  {'users_group_name': users_group_name,
                                   'user_group_description': u'DESC',
                                   'active': True,
-                                  '_authentication_token': self.authentication_token()})
+                                  '_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
         # defaults are on
         perms = UserGroupToPerm.query() \
             .filter(UserGroupToPerm.users_group == ug).all()
         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')
         # check if user has this perms, they should be here since
         # defaults are on
         perms = UserGroupToPerm.query() \
             .filter(UserGroupToPerm.users_group == ug).all()
         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]])
         # 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')
         perms = UserGroupToPerm.query() \
             .filter(UserGroupToPerm.users_group_id == ugid).all()
         perms = [[x.users_group_id,
                   x.permission_id, ] for x in perms]
         assert perms == []
     def test_default_perms_enable_repository_fork_on_group(self):
         self.log_user()
         users_group_name = TEST_USER_GROUP + 'another2'
         response = self.app.post(url('users_groups'),
                                  {'users_group_name': users_group_name,
                                   'user_group_description': u'DESC',
                                   'active': True,
-                                  '_authentication_token': self.authentication_token()})
+                                  '_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')
         # check if user has this perms, they should be here since
         # defaults are on
         perms = UserGroupToPerm.query() \
             .filter(UserGroupToPerm.users_group == ug).all()
         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')
         # check if user has this perms, they should be here since
         # defaults are on
         perms = UserGroupToPerm.query() \
             .filter(UserGroupToPerm.users_group == ug).all()
         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]])
         # 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.fork.repository')
         perms = UserGroupToPerm.query() \
             .filter(UserGroupToPerm.users_group_id == ugid).all()
         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

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 from sqlalchemy.orm.exc import NoResultFound, ObjectDeletedError
 import pytest
 from kallithea.tests.base import *
 from kallithea.tests.fixture import Fixture
 from kallithea.controllers.admin.users import UsersController
 from kallithea.model.db import User, Permission, UserIpMap, UserApiKeys, RepoGroup, UserSshKeys
 from kallithea.lib.auth import check_password
 from kallithea.model.user import UserModel
 from kallithea.model import validators
 from kallithea.lib import helpers as h
 from kallithea.model.meta import Session
 from webob.exc import HTTPNotFound
 from tg.util.webtest import test_context
 fixture = Fixture()
 @pytest.fixture
 def user_and_repo_group_fail():
     username = 'repogrouperr'
     groupname = u'repogroup_fail'
     user = fixture.create_user(name=username)
     repo_group = fixture.create_repo_group(name=groupname, cur_user=username)
     yield user, repo_group
     # cleanup
     if RepoGroup.get_by_group_name(groupname):
         fixture.destroy_repo_group(repo_group)
 class TestAdminUsersController(TestController):
     test_user_1 = 'testme'
     @classmethod
     def teardown_class(cls):
         if User.get_by_username(cls.test_user_1):
             UserModel().delete(cls.test_user_1)
             Session().commit()
     def test_index(self):
         self.log_user()
         response = self.app.get(url('users'))
         # TODO: Test response...
     def test_create(self):
         self.log_user()
         username = 'newtestuser'
         password = 'test12'
         password_confirmation = password
         name = u'name'
         lastname = u'lastname'
         email = 'mail@example.com'
         response = self.app.post(url('new_user'),
             {'username': username,
              'password': password,
              'password_confirmation': password_confirmation,
              'firstname': name,
              'active': True,
              'lastname': lastname,
              'extern_name': 'internal',
              'extern_type': 'internal',
              'email': email,
-             '_authentication_token': self.authentication_token()})
+             '_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()
         response.mustcontain("""%s user settings""" % username) # in <title>
         new_user = Session().query(User). \
             filter(User.username == username).one()
         assert new_user.username == username
         assert check_password(password, new_user.password) == True
         assert new_user.name == name
         assert new_user.lastname == lastname
         assert new_user.email == email
     def test_create_err(self):
         self.log_user()
         username = 'new_user'
         password = ''
         name = u'name'
         lastname = u'lastname'
         email = 'errmail.example.com'
         response = self.app.post(url('new_user'),
             {'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>""")
         response.mustcontain("""<span class="error-message">An email address must contain a single @</span>""")
         def get_user():
             Session().query(User).filter(User.username == username).one()
         with pytest.raises(NoResultFound):
             get_user(), 'found user in database'
     def test_new(self):
         self.log_user()
         response = self.app.get(url('new_user'))
     @parametrize('name,attrs',
         [('firstname', {'firstname': 'new_username'}),
          ('lastname', {'lastname': 'new_username'}),
          ('admin', {'admin': True}),
          ('admin', {'admin': False}),
          ('extern_type', {'extern_type': 'ldap'}),
          ('extern_type', {'extern_type': None}),
          ('extern_name', {'extern_name': 'test'}),
          ('extern_name', {'extern_name': None}),
          ('active', {'active': False}),
          ('active', {'active': True}),
          ('email', {'email': 'someemail@example.com'}),
         # ('new_password', {'new_password': 'foobar123',
         #                   'password_confirmation': 'foobar123'})
         ])
     def test_update(self, name, attrs):
         self.log_user()
         usr = fixture.create_user(self.test_user_1, password='qweqwe',
                                   email='testme@example.com',
                                   extern_type='internal',
                                   extern_name=self.test_user_1,
                                   skip_if_exists=True)
         Session().commit()
         params = usr.get_api_data(True)
         params.update({'password_confirmation': ''})
         params.update({'new_password': ''})
         params.update(attrs)
         if name == 'email':
             params['emails'] = [attrs['email']]
         if name == 'extern_type':
             # cannot update this via form, expected value is original one
             params['extern_type'] = "internal"
         if name == 'extern_name':
             # cannot update this via form, expected value is original one
             params['extern_name'] = self.test_user_1
             # special case since this user is not logged in yet his data is
             # not filled so we use creation data
-        params.update({'_authentication_token': self.authentication_token()})
+        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': ''})
         assert params == updated_params
     def test_delete(self):
         self.log_user()
         username = 'newtestuserdeleteme'
         fixture.create_user(name=username)
         new_user = Session().query(User) \
             .filter(User.username == username).one()
         response = self.app.post(url('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'
         reponame = u'repoerr_fail'
         fixture.create_user(name=username)
         fixture.create_repo(name=reponame, cur_user=username)
         new_user = Session().query(User) \
             .filter(User.username == username).one()
         response = self.app.post(url('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'
         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))
     def test_add_perm_create_repo(self):
         self.log_user()
         perm_none = Permission.get_by_key('hg.create.none')
         perm_create = Permission.get_by_key('hg.create.repository')
         user = UserModel().create_or_update(username='dummy', password='qwe',
                                             email='dummy', firstname=u'a',
                                             lastname=u'b')
         Session().commit()
         uid = user.user_id
         try:
             # User should have None permission on creation repository
             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
             assert UserModel().has_perm(uid, perm_create) == True
         finally:
             UserModel().delete(uid)
             Session().commit()
     def test_revoke_perm_create_repo(self):
         self.log_user()
         perm_none = Permission.get_by_key('hg.create.none')
         perm_create = Permission.get_by_key('hg.create.repository')
         user = UserModel().create_or_update(username='dummy', password='qwe',
                                             email='dummy', firstname=u'a',
                                             lastname=u'b')
         Session().commit()
         uid = user.user_id
         try:
             # User should have None permission on creation repository
             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
             assert UserModel().has_perm(uid, perm_create) == False
         finally:
             UserModel().delete(uid)
             Session().commit()
     def test_add_perm_fork_repo(self):
         self.log_user()
         perm_none = Permission.get_by_key('hg.fork.none')
         perm_fork = Permission.get_by_key('hg.fork.repository')
         user = UserModel().create_or_update(username='dummy', password='qwe',
                                             email='dummy', firstname=u'a',
                                             lastname=u'b')
         Session().commit()
         uid = user.user_id
         try:
             # User should have None permission on creation repository
             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
             assert UserModel().has_perm(uid, perm_create) == True
         finally:
             UserModel().delete(uid)
             Session().commit()
     def test_revoke_perm_fork_repo(self):
         self.log_user()
         perm_none = Permission.get_by_key('hg.fork.none')
         perm_fork = Permission.get_by_key('hg.fork.repository')
         user = UserModel().create_or_update(username='dummy', password='qwe',
                                             email='dummy', firstname=u'a',
                                             lastname=u'b')
         Session().commit()
         uid = user.user_id
         try:
             # User should have None permission on creation repository
             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
             assert UserModel().has_perm(uid, perm_create) == False
         finally:
             UserModel().delete(uid)
             Session().commit()
     def test_ips(self):
         self.log_user()
         user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
         response = self.app.get(url('edit_user_ips', id=user.user_id))
         response.mustcontain('All IP addresses are allowed')
     @parametrize('test_name,ip,ip_range,failure', [
         ('127/24', '127.0.0.1/24', '127.0.0.0 - 127.0.0.255', False),
         ('10/32', '10.0.0.10/32', '10.0.0.10 - 10.0.0.10', False),
         ('0/16', '0.0.0.0/16', '0.0.0.0 - 0.0.255.255', False),
         ('0/8', '0.0.0.0/8', '0.0.0.0 - 0.255.255.255', False),
         ('127_bad_mask', '127.0.0.1/99', '127.0.0.1 - 127.0.0.1', True),
         ('127_bad_ip', 'foobar', 'foobar', True),
     ])
     def test_add_ip(self, test_name, ip, ip_range, failure, 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])
         else:
             response = self.app.get(url('edit_user_ips', id=user_id))
             response.mustcontain(ip)
             response.mustcontain(ip_range)
     def test_delete_ip(self, auto_clear_ip_permissions):
         self.log_user()
         user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
         user_id = user.user_id
         ip = '127.0.0.1/32'
         ip_range = '127.0.0.1 - 127.0.0.1'
         with test_context(self.app):
             new_ip = UserModel().add_extra_ip(user_id, ip)
             Session().commit()
         new_ip_id = new_ip.ip_id
         response = self.app.get(url('edit_user_ips', id=user_id))
         response.mustcontain(ip)
         response.mustcontain(ip_range)
         self.app.post(url('edit_user_ips_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])
     def test_api_keys(self):
         self.log_user()
         user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
         response = self.app.get(url('edit_user_api_keys', id=user.user_id))
         response.mustcontain(user.api_key)
         response.mustcontain('Expires: Never')
     @parametrize('desc,lifetime', [
         ('forever', -1),
         ('5mins', 60*5),
         ('30days', 60*60*24*30),
     ])
     def test_add_api_keys(self, desc, lifetime):
         self.log_user()
         user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
         user_id = user.user_id
         response = self.app.post(url('edit_user_api_keys_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)
         finally:
             for api_key in UserApiKeys.query().filter(UserApiKeys.user_id == user_id).all():
                 Session().delete(api_key)
                 Session().commit()
     def test_remove_api_key(self):
         self.log_user()
         user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
         user_id = user.user_id
         response = self.app.post(url('edit_user_api_keys_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()
         user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
         user_id = user.user_id
         api_key = user.api_key
         response = self.app.get(url('edit_user_api_keys', id=user_id))
         response.mustcontain(api_key)
         response.mustcontain('Expires: Never')
         response = self.app.post(url('edit_user_api_keys_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'
         public_key = u'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQC6Ycnc2oUZHQnQwuqgZqTTdMDZD7ataf3JM7oG2Fw8JR6cdmz4QZLe5mfDwaFwG2pWHLRpVqzfrD/Pn3rIO++bgCJH5ydczrl1WScfryV1hYMJ/4EzLGM657J1/q5EI+b9SntKjf4ax+KP322L0TNQGbZUHLbfG2MwHMrYBQpHUQ== me@localhost'
         fingerprint = u'Ke3oUCNJM87P0jJTb3D+e3shjceP2CqMpQKVd75E9I8'
         self.log_user()
         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
         assert ssh_key.description == description
         Session().delete(ssh_key)
         Session().commit()
     def test_remove_ssh_key(self):
         description = u''
         public_key = u'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQC6Ycnc2oUZHQnQwuqgZqTTdMDZD7ataf3JM7oG2Fw8JR6cdmz4QZLe5mfDwaFwG2pWHLRpVqzfrD/Pn3rIO++bgCJH5ydczrl1WScfryV1hYMJ/4EzLGM657J1/q5EI+b9SntKjf4ax+KP322L0TNQGbZUHLbfG2MwHMrYBQpHUQ== me@localhost'
         fingerprint = u'Ke3oUCNJM87P0jJTb3D+e3shjceP2CqMpQKVd75E9I8'
         self.log_user()
         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):
     """ Unit tests for the users controller """
     def test_get_user_or_raise_if_default(self, monkeypatch, test_context_fixture):
         # flash complains about an non-existing session
         def flash_mock(*args, **kwargs):
             pass
         monkeypatch.setattr(h, 'flash', flash_mock)
         u = UsersController()
         # a regular user should work correctly
         user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
         assert u._get_user_or_raise_if_default(user.user_id) == user
         # the default user should raise
         with pytest.raises(HTTPNotFound):
             u._get_user_or_raise_if_default(User.get_default_user().user_id)
 class TestAdminUsersControllerForDefaultUser(TestController):
     """
     Edit actions on the default user are not allowed.
     Validate that they throw a 404 exception.
     """
     def test_edit_default_user(self):
         self.log_user()
         user = User.get_default_user()
         response = self.app.get(url('edit_user', id=user.user_id), status=404)
     def test_edit_advanced_default_user(self):
         self.log_user()
         user = User.get_default_user()
         response = self.app.get(url('edit_user_advanced', id=user.user_id), status=404)
     # API keys
     def test_edit_api_keys_default_user(self):
         self.log_user()
         user = User.get_default_user()
         response = self.app.get(url('edit_user_api_keys', id=user.user_id), status=404)
     def test_add_api_keys_default_user(self):
         self.log_user()
         user = User.get_default_user()
         response = self.app.post(url('edit_user_api_keys_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()
         user = User.get_default_user()
         response = self.app.get(url('edit_user_ips', id=user.user_id), status=404)

kallithea/tests/functional/test_changeset_pullrequests_comments.py

➞

Show inline comments

 import re
 from kallithea.tests.base import *
 from kallithea.model.changeset_status import ChangesetStatusModel
 from kallithea.model.db import ChangesetComment, PullRequest
 from kallithea.model.meta import Session
 class TestChangeSetCommentsController(TestController):
     def setup_method(self, method):
         for x in ChangesetComment.query().all():
             Session().delete(x)
         Session().commit()
     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'
         response = self.app.get(url(controller='changeset', action='index',
                                 repo_name=HG_REPO, revision=rev))
         response.mustcontain(
             '''<div class="comments-number">'''
             ''' 1 comment (0 inline, 1 general)'''
+        )
         response.mustcontain(text)
         # test DB
         assert ChangesetComment.query().count() == 1
     def test_create_inline(self):
         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'
         response = self.app.get(url(controller='changeset', action='index',
                                 repo_name=HG_REPO, revision=rev))
         response.mustcontain(
             '''<div class="comments-number">'''
             ''' 1 comment (1 inline, 0 general)'''
+        )
         response.mustcontain(
             '''<div class="comments-list-chunk" '''
             '''data-f_path="vcs/web/simplevcs/views/repository.py" '''
             '''data-line_no="n1" data-target-id="vcswebsimplevcsviewsrepositorypy_n1">'''
+        )
         response.mustcontain(text)
         # test DB
         assert ChangesetComment.query().count() == 1
     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'
         response = self.app.get(url(controller='changeset', action='index',
                                 repo_name=HG_REPO, revision=rev))
         response.mustcontain(
             '''<div class="comments-number">'''
             ''' 1 comment (0 inline, 1 general)'''
+        )
         response.mustcontain('<b>@%s</b> check CommentOnRevision' % TEST_USER_REGULAR_LOGIN)
         # test DB
         assert ChangesetComment.query().count() == 1
     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'
         response = self.app.get(url(controller='changeset', action='index',
                                 repo_name=HG_REPO, revision=rev))
         response.mustcontain(
             '''<div class="comments-number">'''
             ''' 1 comment (0 inline, 1 general)'''
+        )
         response.mustcontain(text)
         # test DB
         assert ChangesetComment.query().count() == 1
         # check status
         status = ChangesetStatusModel().get_status(repo=HG_REPO, revision=rev)
         assert status == 'rejected'
     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))
         response.mustcontain(
             '''<div class="comments-number">'''
             ''' 0 comments (0 inline, 0 general)'''
+        )
         response.mustcontain(no=text)
 class TestPullrequestsCommentsController(TestController):
     def setup_method(self, method):
         for x in ChangesetComment.query().all():
             Session().delete(x)
         Session().commit()
     def _create_pr(self):
         response = self.app.post(url(controller='pullrequests', action='create',
                                      repo_name=HG_REPO),
                                  {'org_repo': HG_REPO,
                                   'org_ref': 'branch:stable:4f7e2131323e0749a740c0a56ab68ae9269c562a',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
-                                  '_authentication_token': self.authentication_token(),
+                                  '_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'
         response = self.app.get(url(controller='pullrequests', action='show',
                                 repo_name=HG_REPO, pull_request_id=pr_id, extra=''))
         # PRs currently always have an initial 'Under Review' status change
         # that counts as a general comment, hence '2' in the test below. That
         # could be counted as a misfeature, to be reworked later.
         response.mustcontain(
             '''<div class="comments-number">'''
             ''' 2 comments (0 inline, 2 general)'''
+        )
         response.mustcontain(text)
         # test DB
         assert ChangesetComment.query().count() == 2
     def test_create_inline(self):
         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'
         response = self.app.get(url(controller='pullrequests', action='show',
                                 repo_name=HG_REPO, pull_request_id=pr_id, extra=''))
         response.mustcontain(
             '''<div class="comments-number">'''
             ''' 2 comments (1 inline, 1 general)'''
+        )
         response.mustcontain(
             '''<div class="comments-list-chunk" '''
             '''data-f_path="vcs/web/simplevcs/views/repository.py" '''
             '''data-line_no="n1" data-target-id="vcswebsimplevcsviewsrepositorypy_n1">'''
+        )
         response.mustcontain(text)
         # test DB
         assert ChangesetComment.query().count() == 2
     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'
         response = self.app.get(url(controller='pullrequests', action='show',
                                 repo_name=HG_REPO, pull_request_id=pr_id, extra=''))
         response.mustcontain(
             '''<div class="comments-number">'''
             ''' 2 comments (0 inline, 2 general)'''
+        )
         response.mustcontain('<b>@%s</b> check CommentOnRevision' % TEST_USER_REGULAR_LOGIN)
         # test DB
         assert ChangesetComment.query().count() == 2
     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'
         response = self.app.get(url(controller='pullrequests', action='show',
                                 repo_name=HG_REPO, pull_request_id=pr_id, extra=''))
         # PRs currently always have an initial 'Under Review' status change
         # that counts as a general comment, hence '2' in the test below. That
         # could be counted as a misfeature, to be reworked later.
         response.mustcontain(
             '''<div class="comments-number">'''
             ''' 2 comments (0 inline, 2 general)'''
+        )
         response.mustcontain(text)
         # test DB
         assert ChangesetComment.query().count() == 2
         # check status
         status = ChangesetStatusModel().get_status(repo=HG_REPO, pull_request=pr_id)
         assert status == 'rejected'
     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=''))
         response.mustcontain(
             '''<div class="comments-number">'''
             ''' 1 comment (0 inline, 1 general)'''
+        )
         response.mustcontain(no=text)
     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'
         response = self.app.get(url(controller='pullrequests', action='show',
                                 repo_name=HG_REPO, pull_request_id=pr_id, extra=''))
         response.mustcontain(
             '''title (Closed)'''
+        )
         response.mustcontain(text)
         # test DB
         assert PullRequest.get(pr_id).status == PullRequest.STATUS_CLOSED
     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'
         response = self.app.get(url(controller='pullrequests', action='show',
                                 repo_name=HG_REPO, pull_request_id=pr_id, extra=''), status=404)
         # test DB
         assert PullRequest.get(pr_id) is None
     def test_delete_closed_pr(self):
         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

@@ @@ -144,640 +144,640 @@ class TestFilesController(TestController @@
         response = self.app.get(url(controller='files', action='index',
                                     repo_name=GIT_REPO,
                                     revision='master',
                                     f_path='vcs/nodes.py',
                                     annotate=True))
         response.mustcontain("""r345:c994f0de03b2""")
     def test_file_annotation_history(self):
         self.log_user()
         response = self.app.get(url(controller='files', action='history',
                                     repo_name=HG_REPO,
                                     revision='tip',
                                     f_path='vcs/nodes.py',
                                     annotate=True),
                                 extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         assert response.body == HG_NODE_HISTORY
     def test_file_annotation_history_git(self):
         self.log_user()
         response = self.app.get(url(controller='files', action='history',
                                     repo_name=GIT_REPO,
                                     revision='master',
                                     f_path='vcs/nodes.py',
                                     annotate=True),
                                 extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         assert response.body == GIT_NODE_HISTORY
     def test_file_authors(self):
         self.log_user()
         response = self.app.get(url(controller='files', action='authors',
                                     repo_name=HG_REPO,
                                     revision='tip',
                                     f_path='vcs/nodes.py',
                                     annotate=True))
         response.mustcontain('Marcin Kuzminski')
         response.mustcontain('Lukasz Balcerzak')
     def test_file_authors_git(self):
         self.log_user()
         response = self.app.get(url(controller='files', action='authors',
                                     repo_name=GIT_REPO,
                                     revision='master',
                                     f_path='vcs/nodes.py',
                                     annotate=True))
         response.mustcontain('Marcin Kuzminski')
         response.mustcontain('Lukasz Balcerzak')
     def test_archival(self):
         self.log_user()
         _set_downloads(HG_REPO, set_to=True)
         for arch_ext, info in ARCHIVE_SPECS.items():
             short = '27cd5cce30c9%s' % arch_ext
             fname = '27cd5cce30c96924232dffcd24178a07ffeb5dfc%s' % arch_ext
             filename = '%s-%s' % (HG_REPO, short)
             response = self.app.get(url(controller='files',
                                         action='archivefile',
                                         repo_name=HG_REPO,
                                         fname=fname))
             assert response.status == '200 OK'
             heads = [
                 ('Pragma', 'no-cache'),
                 ('Cache-Control', 'no-cache'),
                 ('Content-Disposition', 'attachment; filename=%s' % filename),
                 ('Content-Type', info[0]),
+            ]
             assert response.response._headers.items() == heads
     def test_archival_wrong_ext(self):
         self.log_user()
         _set_downloads(HG_REPO, set_to=True)
         for arch_ext in ['tar', 'rar', 'x', '..ax', '.zipz']:
             fname = '27cd5cce30c96924232dffcd24178a07ffeb5dfc%s' % arch_ext
             response = self.app.get(url(controller='files',
                                         action='archivefile',
                                         repo_name=HG_REPO,
                                         fname=fname))
             response.mustcontain('Unknown archive type')
     def test_archival_wrong_revision(self):
         self.log_user()
         _set_downloads(HG_REPO, set_to=True)
         for rev in ['00x000000', 'tar', 'wrong', '@##$@$42413232', '232dffcd']:
             fname = '%s.zip' % rev
             response = self.app.get(url(controller='files',
                                         action='archivefile',
                                         repo_name=HG_REPO,
                                         fname=fname))
             response.mustcontain('Unknown revision')
     #==========================================================================
     # RAW FILE
     #==========================================================================
     def test_raw_file_ok(self):
         self.log_user()
         response = self.app.get(url(controller='files', action='rawfile',
                                     repo_name=HG_REPO,
                                     revision='27cd5cce30c96924232dffcd24178a07ffeb5dfc',
                                     f_path='vcs/nodes.py'))
         assert response.content_disposition == "attachment; filename=nodes.py"
         assert response.content_type == mimetypes.guess_type("nodes.py")[0]
     def test_raw_file_wrong_cs(self):
         self.log_user()
         rev = u'ERRORce30c96924232dffcd24178a07ffeb5dfc'
         f_path = 'vcs/nodes.py'
         response = self.app.get(url(controller='files', action='rawfile',
                                     repo_name=HG_REPO,
                                     revision=rev,
                                     f_path=f_path), status=404)
         msg = """Such revision does not exist for this repository"""
         response.mustcontain(msg)
     def test_raw_file_wrong_f_path(self):
         self.log_user()
         rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
         f_path = 'vcs/ERRORnodes.py'
         response = self.app.get(url(controller='files', action='rawfile',
                                     repo_name=HG_REPO,
                                     revision=rev,
                                     f_path=f_path), status=404)
         msg = "There is no file nor directory at the given path: &#39;%s&#39; at revision %s" % (f_path, rev[:12])
         response.mustcontain(msg)
     #==========================================================================
     # RAW RESPONSE - PLAIN
     #==========================================================================
     def test_raw_ok(self):
         self.log_user()
         response = self.app.get(url(controller='files', action='raw',
                                     repo_name=HG_REPO,
                                     revision='27cd5cce30c96924232dffcd24178a07ffeb5dfc',
                                     f_path='vcs/nodes.py'))
         assert response.content_type == "text/plain"
     def test_raw_wrong_cs(self):
         self.log_user()
         rev = u'ERRORcce30c96924232dffcd24178a07ffeb5dfc'
         f_path = 'vcs/nodes.py'
         response = self.app.get(url(controller='files', action='raw',
                                     repo_name=HG_REPO,
                                     revision=rev,
                                     f_path=f_path), status=404)
         msg = """Such revision does not exist for this repository"""
         response.mustcontain(msg)
     def test_raw_wrong_f_path(self):
         self.log_user()
         rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
         f_path = 'vcs/ERRORnodes.py'
         response = self.app.get(url(controller='files', action='raw',
                                     repo_name=HG_REPO,
                                     revision=rev,
                                     f_path=f_path), status=404)
         msg = "There is no file nor directory at the given path: &#39;%s&#39; at revision %s" % (f_path, rev[:12])
         response.mustcontain(msg)
     def test_ajaxed_files_list(self):
         self.log_user()
         rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
         response = self.app.get(
             url('files_nodelist_home', repo_name=HG_REPO, f_path='/',
                 revision=rev),
             extra_environ={'HTTP_X_PARTIAL_XHR': '1'},
+        )
         response.mustcontain("vcs/web/simplevcs/views/repository.py")
     # Hg - ADD FILE
     def test_add_file_view_hg(self):
         self.log_user()
         response = self.app.get(url('files_add_home',
                                       repo_name=HG_REPO,
                                       revision='tip', f_path='/'))
     def test_add_file_into_hg_missing_content(self):
         self.log_user()
         response = self.app.post(url('files_add_home',
                                       repo_name=HG_REPO,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': '',
-                                    '_authentication_token': self.authentication_token(),
+                                    '_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', [
         ('/abs', 'foo'),
         ('../rel', 'foo'),
         ('file/../foo', 'foo'),
     ])
     def test_add_file_into_hg_bad_filenames(self, location, filename):
         self.log_user()
         response = self.app.post(url('files_add_home',
                                       repo_name=HG_REPO,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "foo",
                                     'filename': filename,
                                     'location': location,
-                                    '_authentication_token': self.authentication_token(),
+                                    '_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', [
         (1, '', 'foo.txt'),
         (2, 'dir', 'foo.rst'),
         (3, 'rel/dir', 'foo.bar'),
     ])
     def test_add_file_into_hg(self, cnt, location, filename):
         self.log_user()
         repo = fixture.create_repo(u'commit-test-%s' % cnt, repo_type='hg')
         response = self.app.post(url('files_add_home',
                                       repo_name=repo.repo_name,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "foo",
                                     'filename': filename,
                                     'location': location,
-                                    '_authentication_token': self.authentication_token(),
+                                    '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
         finally:
             fixture.destroy_repo(repo.repo_name)
     # Git - add file
     def test_add_file_view_git(self):
         self.log_user()
         response = self.app.get(url('files_add_home',
                                       repo_name=GIT_REPO,
                                       revision='tip', f_path='/'))
     def test_add_file_into_git_missing_content(self):
         self.log_user()
         response = self.app.post(url('files_add_home',
                                       repo_name=GIT_REPO,
                                       revision='tip', f_path='/'),
                                  params={
                                      'content': '',
-                                     '_authentication_token': self.authentication_token(),
+                                     '_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', [
         ('/abs', 'foo'),
         ('../rel', 'foo'),
         ('file/../foo', 'foo'),
     ])
     def test_add_file_into_git_bad_filenames(self, location, filename):
         self.log_user()
         response = self.app.post(url('files_add_home',
                                       repo_name=GIT_REPO,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "foo",
                                     'filename': filename,
                                     'location': location,
-                                    '_authentication_token': self.authentication_token(),
+                                    '_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', [
         (1, '', 'foo.txt'),
         (2, 'dir', 'foo.rst'),
         (3, 'rel/dir', 'foo.bar'),
     ])
     def test_add_file_into_git(self, cnt, location, filename):
         self.log_user()
         repo = fixture.create_repo(u'commit-test-%s' % cnt, repo_type='git')
         response = self.app.post(url('files_add_home',
                                       repo_name=repo.repo_name,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "foo",
                                     'filename': filename,
                                     'location': location,
-                                    '_authentication_token': self.authentication_token(),
+                                    '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
         finally:
             fixture.destroy_repo(repo.repo_name)
     # Hg - EDIT
     def test_edit_file_view_hg(self):
         self.log_user()
         response = self.app.get(url('files_edit_home',
                                       repo_name=HG_REPO,
                                       revision='tip', f_path='vcs/nodes.py'))
     def test_edit_file_view_not_on_branch_hg(self):
         self.log_user()
         repo = fixture.create_repo(u'test-edit-repo', repo_type='hg')
         ## add file
         location = 'vcs'
         filename = 'nodes.py'
         response = self.app.post(url('files_add_home',
                                       repo_name=repo.repo_name,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
-                                    '_authentication_token': self.authentication_token(),
+                                    '_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.get(url('files_edit_home',
                                           repo_name=repo.repo_name,
                                           revision='tip', f_path=posixpath.join(location, filename)),
                                     status=302)
             self.checkSessionFlash(response,
                 'You can only edit files with revision being a valid branch')
         finally:
             fixture.destroy_repo(repo.repo_name)
     def test_edit_file_view_commit_changes_hg(self):
         self.log_user()
         repo = fixture.create_repo(u'test-edit-repo', repo_type='hg')
         ## add file
         location = 'vcs'
         filename = 'nodes.py'
         response = self.app.post(url('files_add_home',
                                       repo_name=repo.repo_name,
                                       revision='tip',
                                       f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
-                                    '_authentication_token': self.authentication_token(),
+                                    '_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_edit_home',
                                           repo_name=repo.repo_name,
                                           revision=repo.scm_instance.DEFAULT_BRANCH_NAME,
                                           f_path=posixpath.join(location, filename)),
                                      params={
                                         'content': "def py():\n print 'hello world'\n",
                                         'message': 'i committed',
-                                        '_authentication_token': self.authentication_token(),
+                                        '_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)
     # Git - edit
     def test_edit_file_view_git(self):
         self.log_user()
         response = self.app.get(url('files_edit_home',
                                       repo_name=GIT_REPO,
                                       revision='tip', f_path='vcs/nodes.py'))
     def test_edit_file_view_not_on_branch_git(self):
         self.log_user()
         repo = fixture.create_repo(u'test-edit-repo', repo_type='git')
         ## add file
         location = 'vcs'
         filename = 'nodes.py'
         response = self.app.post(url('files_add_home',
                                       repo_name=repo.repo_name,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
-                                    '_authentication_token': self.authentication_token(),
+                                    '_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.get(url('files_edit_home',
                                           repo_name=repo.repo_name,
                                           revision='tip', f_path=posixpath.join(location, filename)),
                                     status=302)
             self.checkSessionFlash(response,
                 'You can only edit files with revision being a valid branch')
         finally:
             fixture.destroy_repo(repo.repo_name)
     def test_edit_file_view_commit_changes_git(self):
         self.log_user()
         repo = fixture.create_repo(u'test-edit-repo', repo_type='git')
         ## add file
         location = 'vcs'
         filename = 'nodes.py'
         response = self.app.post(url('files_add_home',
                                       repo_name=repo.repo_name,
                                       revision='tip',
                                       f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
-                                    '_authentication_token': self.authentication_token(),
+                                    '_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_edit_home',
                                           repo_name=repo.repo_name,
                                           revision=repo.scm_instance.DEFAULT_BRANCH_NAME,
                                           f_path=posixpath.join(location, filename)),
                                      params={
                                         'content': "def py():\n print 'hello world'\n",
                                         'message': 'i committed',
-                                        '_authentication_token': self.authentication_token(),
+                                        '_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)
     # Hg - delete
     def test_delete_file_view_hg(self):
         self.log_user()
         response = self.app.get(url('files_delete_home',
                                      repo_name=HG_REPO,
                                      revision='tip', f_path='vcs/nodes.py'))
     def test_delete_file_view_not_on_branch_hg(self):
         self.log_user()
         repo = fixture.create_repo(u'test-delete-repo', repo_type='hg')
         ## add file
         location = 'vcs'
         filename = 'nodes.py'
         response = self.app.post(url('files_add_home',
                                       repo_name=repo.repo_name,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
-                                    '_authentication_token': self.authentication_token(),
+                                    '_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.get(url('files_delete_home',
                                           repo_name=repo.repo_name,
                                           revision='tip', f_path=posixpath.join(location, filename)),
                                     status=302)
             self.checkSessionFlash(response,
                 'You can only delete files with revision being a valid branch')
         finally:
             fixture.destroy_repo(repo.repo_name)
     def test_delete_file_view_commit_changes_hg(self):
         self.log_user()
         repo = fixture.create_repo(u'test-delete-repo', repo_type='hg')
         ## add file
         location = 'vcs'
         filename = 'nodes.py'
         response = self.app.post(url('files_add_home',
                                       repo_name=repo.repo_name,
                                       revision='tip',
                                       f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
-                                    '_authentication_token': self.authentication_token(),
+                                    '_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)
     # Git - delete
     def test_delete_file_view_git(self):
         self.log_user()
         response = self.app.get(url('files_delete_home',
                                      repo_name=HG_REPO,
                                      revision='tip', f_path='vcs/nodes.py'))
     def test_delete_file_view_not_on_branch_git(self):
         self.log_user()
         repo = fixture.create_repo(u'test-delete-repo', repo_type='git')
         ## add file
         location = 'vcs'
         filename = 'nodes.py'
         response = self.app.post(url('files_add_home',
                                       repo_name=repo.repo_name,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
-                                    '_authentication_token': self.authentication_token(),
+                                    '_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.get(url('files_delete_home',
                                           repo_name=repo.repo_name,
                                           revision='tip', f_path=posixpath.join(location, filename)),
                                     status=302)
             self.checkSessionFlash(response,
                 'You can only delete files with revision being a valid branch')
         finally:
             fixture.destroy_repo(repo.repo_name)
     def test_delete_file_view_commit_changes_git(self):
         self.log_user()
         repo = fixture.create_repo(u'test-delete-repo', repo_type='git')
         ## add file
         location = 'vcs'
         filename = 'nodes.py'
         response = self.app.post(url('files_add_home',
                                       repo_name=repo.repo_name,
                                       revision='tip',
                                       f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
-                                    '_authentication_token': self.authentication_token(),
+                                    '_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)
     def test_png_diff_no_crash_hg(self):
         self.log_user()
         response = self.app.get(url('files_diff_home',
                                     repo_name=HG_REPO,
                                     f_path='docs/theme/ADC/static/documentation.png',
                                     diff1='tip', diff2='tip'))
         response.mustcontain("""<pre>Binary file</pre>""")
     def test_png_diff_no_crash_git(self):
         self.log_user()
         response = self.app.get(url('files_diff_home',
                                     repo_name=GIT_REPO,
                                     f_path='docs/theme/ADC/static/documentation.png',
                                     diff1='master', diff2='master'))
         response.mustcontain("""<pre>Binary file</pre>""")

kallithea/tests/functional/test_forks.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 import urllib
 import unittest
 from kallithea.tests.base import *
 from kallithea.tests.fixture import Fixture
 from kallithea.lib.utils2 import safe_str, safe_unicode
 from kallithea.model.db import Repository, User
 from kallithea.model.repo import RepoModel
 from kallithea.model.user import UserModel
 from kallithea.model.meta import Session
 fixture = Fixture()
 class _BaseTestCase(TestController):
     """
     Write all tests here
     """
     REPO = None
     REPO_TYPE = None
     NEW_REPO = None
     REPO_FORK = None
     def setup_method(self, method):
         self.username = u'forkuser'
         self.password = u'qweqwe'
         u1 = fixture.create_user(self.username, password=self.password, email=u'fork_king@example.com')
         self.u1_id = u1.user_id
         Session().commit()
     def teardown_method(self, method):
         fixture.destroy_user(self.u1_id)
         Session().commit()
     def test_index(self):
         self.log_user()
         repo_name = self.REPO
         response = self.app.get(url(controller='forks', action='forks',
                                     repo_name=repo_name))
         response.mustcontain("""There are no forks yet""")
     def test_no_permissions_to_fork(self):
         self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)['user_id']
         try:
             user_model = UserModel()
             usr = User.get_default_user()
             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()
     def test_index_with_fork(self):
         self.log_user()
         # create a fork
         fork_name = self.REPO_FORK
         description = 'fork of vcs test'
         repo_name = self.REPO
         org_repo = Repository.get_by_repo_name(repo_name)
         creation_args = {
             'repo_name': fork_name,
             'repo_group': u'-1',
             'fork_parent_id': org_repo.repo_id,
             'repo_type': self.REPO_TYPE,
             'description': description,
             'private': 'False',
             'landing_rev': 'rev:tip',
-            '_authentication_token': self.authentication_token()}
+            '_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>""" % (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
         fork_name_full = 'vc/%s' % fork_name
         description = 'fork of vcs test'
         repo_name = self.REPO
         org_repo = Repository.get_by_repo_name(repo_name)
         creation_args = {
             'repo_name': fork_name,
             'repo_group': group_id,
             'fork_parent_id': org_repo.repo_id,
             'repo_type': self.REPO_TYPE,
             'description': description,
             'private': 'False',
             'landing_rev': 'rev:tip',
-            '_authentication_token': self.authentication_token()}
+            '_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
         response = self.app.get(url('repo_check_home', repo_name=fork_name_full))
         # test if we have a message that fork is ok
         self.checkSessionFlash(response,
                 'Forked repository %s as <a href="/%s">%s</a>'
                 % (repo_name, fork_name_full, fork_name_full))
         # test if the fork was created in the database
         fork_repo = Session().query(Repository) \
             .filter(Repository.repo_name == fork_name_full).one()
         assert fork_repo.repo_name == fork_name_full
         assert fork_repo.fork.repo_name == repo_name
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=fork_name_full))
         response.mustcontain(fork_name_full)
         response.mustcontain(self.REPO_TYPE)
         response.mustcontain('Fork of "<a href="/%s">%s</a>"' % (repo_name, repo_name))
         fixture.destroy_repo(fork_name_full)
         fixture.destroy_repo_group(group_id)
     def test_fork_unicode(self):
         self.log_user()
         # create a fork
         repo_name = self.REPO
         org_repo = Repository.get_by_repo_name(repo_name)
         fork_name = safe_str(self.REPO_FORK + u'-rødgrød')
         creation_args = {
             'repo_name': fork_name,
             '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)
+        )
         fork_repo = Repository.get_by_repo_name(safe_unicode(fork_name))
         assert fork_repo
         # fork the fork
         fork_name_2 = safe_str(self.REPO_FORK + u'-blåbærgrød')
         creation_args = {
             'repo_name': fork_name_2,
             '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
         org_repo = Repository.get_by_repo_name(repo_name)
         creation_args = {
             'repo_name': fork_name,
             'repo_group': u'-1',
             'fork_parent_id': org_repo.repo_id,
             'repo_type': self.REPO_TYPE,
             'description': description,
             'private': 'False',
             'landing_rev': 'rev:tip',
-            '_authentication_token': self.authentication_token()}
+            '_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
         response = self.app.get(url('repo_check_home', repo_name=fork_name))
         # test if we have a message that fork is ok
         self.checkSessionFlash(response,
                 'Forked repository %s as <a href="/%s">%s</a>'
                 % (repo_name, fork_name, fork_name))
         # test if the fork was created in the database
         fork_repo = Session().query(Repository) \
             .filter(Repository.repo_name == fork_name).one()
         assert fork_repo.repo_name == fork_name
         assert fork_repo.fork.repo_name == repo_name
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=fork_name))
         response.mustcontain(fork_name)
         response.mustcontain(self.REPO_TYPE)
         response.mustcontain('Fork of "<a href="/%s">%s</a>"' % (repo_name, repo_name))
         usr = self.log_user(self.username, self.password)['user_id']
         forks = Repository.query() \
             .filter(Repository.repo_type == self.REPO_TYPE) \
             .filter(Repository.fork_id != None).all()
         assert 1 == len(forks)
         # set read permissions for this
         RepoModel().grant_user_permission(repo=forks[0],
                                           user=usr,
                                           perm='repository.read')
         Session().commit()
         response = self.app.get(url(controller='forks', action='forks',
                                     repo_name=repo_name))
         response.mustcontain('<div>fork of vcs test</div>')
         # remove permissions
         default_user = User.get_default_user()
         try:
             RepoModel().grant_user_permission(repo=forks[0],
                                               user=usr, perm='repository.none')
             RepoModel().grant_user_permission(repo=forks[0],
                                               user=default_user, perm='repository.none')
             Session().commit()
             # fork shouldn't be visible
             response = self.app.get(url(controller='forks', action='forks',
                                         repo_name=repo_name))
             response.mustcontain('There are no forks yet')
         finally:
             RepoModel().grant_user_permission(repo=forks[0],
                                               user=usr, perm='repository.read')
             RepoModel().grant_user_permission(repo=forks[0],
                                               user=default_user, perm='repository.read')
             RepoModel().delete(repo=forks[0])
 class TestGIT(_BaseTestCase):
     REPO = GIT_REPO
     NEW_REPO = NEW_GIT_REPO
     REPO_TYPE = 'git'
     REPO_FORK = GIT_FORK
 class TestHG(_BaseTestCase):
     REPO = HG_REPO
     NEW_REPO = NEW_HG_REPO
     REPO_TYPE = 'hg'
     REPO_FORK = HG_FORK

kallithea/tests/functional/test_login.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 import re
 import time
 import urlparse
 import mock
 from kallithea.tests.base import *
 from kallithea.tests.fixture import Fixture
 from kallithea.lib.utils2 import generate_api_key
 from kallithea.lib.auth import check_password
 from kallithea.lib import helpers as h
 from kallithea.model.api_key import ApiKeyModel
 from kallithea.model import validators
 from kallithea.model.db import User
 from kallithea.model.meta import Session
 from kallithea.model.user import UserModel
 from tg.util.webtest import test_context
 fixture = Fixture()
 class TestLoginController(TestController):
     def test_index(self):
         response = self.app.get(url(controller='login', action='index'))
         assert response.status == '200 OK'
         # Test response...
     def test_login_admin_ok(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_ADMIN_LOGIN,
                                   'password': TEST_USER_ADMIN_PASS,
-                                  '_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)
     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
         response.click('Log Out')
         # Verify that the login session has been terminated.
         response = self.app.get(url(controller='login', action='index'))
         assert 'authuser' not in response.session
     @parametrize('url_came_from', [
           ('data:text/html,<script>window.alert("xss")</script>',),
           ('mailto:test@example.com',),
           ('file:///etc/passwd',),
           ('ftp://ftp.example.com',),
           ('http://other.example.com/bl%C3%A5b%C3%A6rgr%C3%B8d',),
           ('//evil.example.com/',),
           ('/\r\nX-Header-Injection: boo',),
           ('/invälid_url_bytes',),
           ('non-absolute-path',),
     ])
     def test_login_bad_came_froms(self, url_came_from):
         response = self.app.post(url(controller='login', action='index',
                                      came_from=url_came_from),
                                  {'username': TEST_USER_ADMIN_LOGIN,
                                   'password': TEST_USER_ADMIN_PASS,
-                                  '_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', [
         ({'foo':'one', 'bar':'two'}, (('foo', 'one'), ('bar', 'two'))),
         ({'blue': u'blå'.encode('utf-8'), 'green':u'grøn'},
              (('blue', u'blå'.encode('utf-8')), ('green', u'grøn'.encode('utf-8')))),
     ])
     def test_redirection_to_login_form_preserves_get_args(self, args, args_encoded):
         with fixture.anon_access(False):
             response = self.app.get(url(controller='summary', action='index',
                                         repo_name=HG_REPO,
                                         **args))
             assert response.status == '302 Found'
             came_from = urlparse.parse_qs(urlparse.urlparse(response.location).query)['came_from'][0]
             came_from_qs = urlparse.parse_qsl(urlparse.urlparse(came_from).query)
             for encoded in args_encoded:
                 assert encoded in came_from_qs
     @parametrize('args,args_encoded', [
         ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),
         ({'blue': u'blå', 'green':u'grøn'},
              ('blue=bl%C3%A5', 'green=gr%C3%B8n')),
     ])
     def test_login_form_preserves_get_args(self, args, args_encoded):
         response = self.app.get(url(controller='login', action='index',
                                     came_from=url('/_admin/users', **args)))
         came_from = urlparse.parse_qs(urlparse.urlparse(response.form.action).query)['came_from'][0]
         for encoded in args_encoded:
             assert encoded in came_from
     @parametrize('args,args_encoded', [
         ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),
         ({'blue': u'blå', 'green':u'grøn'},
              ('blue=bl%C3%A5', 'green=gr%C3%B8n')),
     ])
     def test_redirection_after_successful_login_preserves_get_args(self, args, args_encoded):
         response = self.app.post(url(controller='login', action='index',
                                      came_from=url('/_admin/users', **args)),
                                  {'username': TEST_USER_ADMIN_LOGIN,
                                   'password': TEST_USER_ADMIN_PASS,
-                                  '_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')),
         ({'blue': u'blå', 'green':u'grøn'},
              ('blue=bl%C3%A5', 'green=gr%C3%B8n')),
     ])
     def test_login_form_after_incorrect_login_preserves_get_args(self, args, args_encoded):
         response = self.app.post(url(controller='login', action='index',
                                      came_from=url('/_admin/users', **args)),
                                  {'username': 'error',
                                   'password': 'test12',
-                                  '_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
     #==========================================================================
     # REGISTRATIONS
     #==========================================================================
     def test_register(self):
         response = self.app.get(url(controller='login', action='register'))
         response.mustcontain('Sign Up')
     def test_register_err_same_username(self):
         uname = TEST_USER_ADMIN_LOGIN
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': uname,
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': 'goodmail@example.com',
                                              'firstname': 'test',
                                              'lastname': 'test',
-                                             '_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)
     def test_register_err_same_email(self):
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': 'test_admin_0',
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': TEST_USER_ADMIN_EMAIL,
                                              'firstname': 'test',
                                              'lastname': 'test',
-                                             '_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):
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': 'test_admin_1',
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': TEST_USER_ADMIN_EMAIL.title(),
                                              'firstname': 'test',
                                              'lastname': 'test',
-                                             '_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')
     def test_register_err_case_sensitive(self):
         usr = TEST_USER_ADMIN_LOGIN.title()
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': usr,
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': 'goodmailm',
                                              'firstname': 'test',
                                              'lastname': 'test',
-                                             '_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)
     def test_register_special_chars(self):
         response = self.app.post(url(controller='login', action='register'),
                                         {'username': 'xxxaxn',
                                          'password': 'ąćźżąśśśś',
                                          'password_confirmation': 'ąćźżąśśśś',
                                          'email': 'goodmailm@test.plx',
                                          'firstname': 'test',
                                          'lastname': 'test',
-                                         '_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):
         response = self.app.post(url(controller='login', action='register'),
                                             {'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'
         password = 'qweqwe'
         email = 'user4@example.com'
         name = 'testname'
         lastname = 'testlastname'
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': username,
                                              'password': password,
                                              'password_confirmation': password,
                                              'email': email,
                                              'firstname': name,
                                              'lastname': lastname,
                                              'admin': True,
-                                             '_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
         assert ret.email == email
         assert ret.name == name
         assert ret.lastname == lastname
         assert ret.api_key is not None
         assert ret.admin == False
     #==========================================================================
     # PASSWORD RESET
     #==========================================================================
     def test_forgot_password_wrong_mail(self):
         bad_email = 'username%wrongmail.org'
         response = self.app.post(
                         url(controller='login', action='password_reset'),
                             {'email': bad_email,
-                             '_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'))
         assert response.status == '200 OK'
         username = 'test_password_reset_1'
         password = 'qweqwe'
         email = 'username@example.com'
         name = u'passwd'
         lastname = u'reset'
         timestamp = int(time.time())
         new = User()
         new.username = username
         new.password = password
         new.email = email
         new.name = name
         new.lastname = lastname
         new.api_key = generate_api_key()
         Session().add(new)
         Session().commit()
         response = self.app.post(url(controller='login',
                                      action='password_reset'),
                                  {'email': email,
-                                  '_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
         token = "bad"
         response = self.app.post(url(controller='login',
                                      action='password_reset_confirmation'),
                                  {'email': email,
                                   'timestamp': timestamp,
                                   'password': "p@ssw0rd",
                                   'password_confirm': "p@ssw0rd",
                                   'token': token,
-                                  '_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))
         assert response.status == '200 OK'
         response.mustcontain("You are about to set a new password for the email address %s" % email)
         response = self.app.post(url(controller='login',
                                      action='password_reset_confirmation'),
                                  {'email': email,
                                   'timestamp': timestamp,
                                   'password': "p@ssw0rd",
                                   'password_confirm': "p@ssw0rd",
                                   'token': token,
-                                  '_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()
     #==========================================================================
     # API
     #==========================================================================
     def _api_key_test(self, api_key, status):
         """Verifies HTTP status code for accessing an auth-requiring page,
         using the given api_key URL parameter as well as using the API key
         with bearer authentication.
         If api_key is None, no api_key is passed at all. If api_key is True,
         a real, working API key is used.
         """
         with fixture.anon_access(False):
             if api_key is None:
                 params = {}
                 headers = {}
             else:
                 if api_key is True:
                     api_key = User.get_first_admin().api_key
                 params = {'api_key': api_key}
                 headers = {'Authorization': 'Bearer ' + str(api_key)}
             self.app.get(url(controller='changeset', action='changeset_raw',
                              repo_name=HG_REPO, revision='tip', **params),
                          status=status)
             self.app.get(url(controller='changeset', action='changeset_raw',
                              repo_name=HG_REPO, revision='tip'),
                          headers=headers,
                          status=status)
     @parametrize('test_name,api_key,code', [
         ('none', None, 302),
         ('empty_string', '', 403),
         ('fake_number', '123456', 403),
         ('fake_not_alnum', 'a-z', 403),
         ('fake_api_key', '0123456789abcdef0123456789ABCDEF01234567', 403),
         ('proper_api_key', True, 200)
     ])
     def test_access_page_via_api_key(self, test_name, api_key, code):
         self._api_key_test(api_key, code)
     def test_access_page_via_extra_api_key(self):
         new_api_key = ApiKeyModel().create(TEST_USER_ADMIN_LOGIN, u'test')
         Session().commit()
         self._api_key_test(new_api_key.api_key, status=200)
     def test_access_page_via_expired_api_key(self):
         new_api_key = ApiKeyModel().create(TEST_USER_ADMIN_LOGIN, u'test')
         Session().commit()
         # patch the API key and make it expired
         new_api_key.expires = 0
         Session().commit()
         self._api_key_test(new_api_key.api_key, status=403)

kallithea/tests/functional/test_my_account.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 from kallithea.model.db import User, UserFollowing, Repository, UserApiKeys, UserSshKeys
 from kallithea.tests.base import *
 from kallithea.tests.fixture import Fixture
 from kallithea.lib import helpers as h
 from kallithea.model.user import UserModel
 from kallithea.model.meta import Session
 from tg.util.webtest import test_context
 fixture = Fixture()
 class TestMyAccountController(TestController):
     test_user_1 = 'testme'
     @classmethod
     def teardown_class(cls):
         if User.get_by_username(cls.test_user_1):
             UserModel().delete(cls.test_user_1)
             Session().commit()
     def test_my_account(self):
         self.log_user()
         response = self.app.get(url('my_account'))
         response.mustcontain('value="%s' % TEST_USER_ADMIN_LOGIN)
     def test_my_account_my_repos(self):
         self.log_user()
         response = self.app.get(url('my_account_repos'))
         cnt = Repository.query().filter(Repository.owner ==
                            User.get_by_username(TEST_USER_ADMIN_LOGIN)).count()
         response.mustcontain('"raw_name": "%s"' % HG_REPO)
         response.mustcontain('"just_name": "%s"' % GIT_REPO)
     def test_my_account_my_watched(self):
         self.log_user()
         response = self.app.get(url('my_account_watched'))
         cnt = UserFollowing.query().filter(UserFollowing.user ==
                             User.get_by_username(TEST_USER_ADMIN_LOGIN)).count()
         response.mustcontain('"raw_name": "%s"' % HG_REPO)
         response.mustcontain('"just_name": "%s"' % GIT_REPO)
     def test_my_account_my_emails(self):
         self.log_user()
         response = self.app.get(url('my_account_emails'))
         response.mustcontain('No additional emails specified')
     def test_my_account_my_emails_add_existing_email(self):
         self.log_user()
         response = self.app.get(url('my_account_emails'))
         response.mustcontain('No additional emails specified')
         response = self.app.post(url('my_account_emails'),
-                                 {'new_email': TEST_USER_REGULAR_EMAIL, '_authentication_token': self.authentication_token()})
+                                 {'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',
         [('firstname', {'firstname': 'new_username'}),
          ('lastname', {'lastname': 'new_username'}),
          ('admin', {'admin': True}),
          ('admin', {'admin': False}),
          ('extern_type', {'extern_type': 'ldap'}),
          ('extern_type', {'extern_type': None}),
          #('extern_name', {'extern_name': 'test'}),
          #('extern_name', {'extern_name': None}),
          ('active', {'active': False}),
          ('active', {'active': True}),
          ('email', {'email': 'someemail@example.com'}),
         # ('new_password', {'new_password': 'foobar123',
         #                   'password_confirmation': 'foobar123'})
         ])
     def test_my_account_update(self, name, attrs):
         usr = fixture.create_user(self.test_user_1, password='qweqwe',
                                   email='testme@example.com',
                                   extern_type='internal',
                                   extern_name=self.test_user_1,
                                   skip_if_exists=True)
         params = usr.get_api_data(True)  # current user data
         user_id = usr.user_id
         self.log_user(username=self.test_user_1, password='qweqwe')
         params.update({'password_confirmation': ''})
         params.update({'new_password': ''})
         params.update({'extern_type': 'internal'})
         params.update({'extern_name': self.test_user_1})
-        params.update({'_authentication_token': self.authentication_token()})
+        params.update({'_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')
         updated_user = User.get_by_username(self.test_user_1)
         updated_params = updated_user.get_api_data(True)
         updated_params.update({'password_confirmation': ''})
         updated_params.update({'new_password': ''})
         params['last_login'] = updated_params['last_login']
         if name == 'email':
             params['emails'] = [attrs['email']]
         if name == 'extern_type':
             # cannot update this via form, expected value is original one
             params['extern_type'] = "internal"
         if name == 'extern_name':
             # cannot update this via form, expected value is original one
             params['extern_name'] = str(user_id)
         if name == 'active':
             # my account cannot deactivate account
             params['active'] = True
         if name == 'admin':
             # my account cannot make you an admin !
             params['admin'] = False
-        params.pop('_authentication_token')
+        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
         response = self.app.post(url('my_account'),
                                 params=dict(
                                     username=TEST_USER_ADMIN_LOGIN,
                                     new_password=TEST_USER_ADMIN_PASS,
                                     password_confirmation='test122',
                                     firstname=u'NewName',
                                     lastname=u'NewLastname',
                                     email=new_email,
-                                    _authentication_token=self.authentication_token())
+                                    _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)
         new_email = 'newmail.pl'
         response = self.app.post(url('my_account'),
                                  params=dict(
                                             username=TEST_USER_ADMIN_LOGIN,
                                             new_password=TEST_USER_ADMIN_PASS,
                                             password_confirmation='test122',
                                             firstname=u'NewName',
                                             lastname=u'NewLastname',
                                             email=new_email,
-                                            _authentication_token=self.authentication_token()))
+                                            _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']
         msg = h.html_escape(msg % {'username': TEST_USER_ADMIN_LOGIN})
         response.mustcontain(msg)
     def test_my_account_api_keys(self):
         usr = self.log_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS)
         user = User.get(usr['user_id'])
         response = self.app.get(url('my_account_api_keys'))
         response.mustcontain(user.api_key)
         response.mustcontain('Expires: Never')
     @parametrize('desc,lifetime', [
         ('forever', -1),
         ('5mins', 60*5),
         ('30days', 60*60*24*30),
     ])
     def test_my_account_add_api_keys(self, desc, lifetime):
         usr = self.log_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS)
         user = User.get(usr['user_id'])
         response = self.app.post(url('my_account_api_keys'),
-                                 {'description': desc, 'lifetime': lifetime, '_authentication_token': self.authentication_token()})
+                                 {'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)
         finally:
             for api_key in UserApiKeys.query().all():
                 Session().delete(api_key)
                 Session().commit()
     def test_my_account_remove_api_key(self):
         usr = self.log_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS)
         user = User.get(usr['user_id'])
         response = self.app.post(url('my_account_api_keys'),
-                                 {'description': 'desc', 'lifetime': -1, '_authentication_token': self.authentication_token()})
+                                 {'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)
         user = User.get(usr['user_id'])
         api_key = user.api_key
         response = self.app.get(url('my_account_api_keys'))
         response.mustcontain(api_key)
         response.mustcontain('Expires: Never')
         response = self.app.post(url('my_account_api_keys_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'
         public_key = u'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQC6Ycnc2oUZHQnQwuqgZqTTdMDZD7ataf3JM7oG2Fw8JR6cdmz4QZLe5mfDwaFwG2pWHLRpVqzfrD/Pn3rIO++bgCJH5ydczrl1WScfryV1hYMJ/4EzLGM657J1/q5EI+b9SntKjf4ax+KP322L0TNQGbZUHLbfG2MwHMrYBQpHUQ== me@localhost'
         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()
         assert ssh_key.fingerprint == fingerprint
         assert ssh_key.description == description
         Session().delete(ssh_key)
         Session().commit()
     def test_my_account_remove_ssh_key(self):
         description = u''
         public_key = u'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQC6Ycnc2oUZHQnQwuqgZqTTdMDZD7ataf3JM7oG2Fw8JR6cdmz4QZLe5mfDwaFwG2pWHLRpVqzfrD/Pn3rIO++bgCJH5ydczrl1WScfryV1hYMJ/4EzLGM657J1/q5EI+b9SntKjf4ax+KP322L0TNQGbZUHLbfG2MwHMrYBQpHUQ== me@localhost'
         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

 import re
 import pytest
 from tg.util.webtest import test_context
 from kallithea.tests.base import *
 from kallithea.tests.fixture import Fixture
 from kallithea.model.db import PullRequest, User
 from kallithea.model.meta import Session
 from kallithea.controllers.pullrequests import PullrequestsController
 fixture = Fixture()
 class TestPullrequestsController(TestController):
     def test_index(self):
         self.log_user()
         response = self.app.get(url(controller='pullrequests', action='index',
                                     repo_name=HG_REPO))
     def test_create_trivial(self):
         self.log_user()
         response = self.app.post(url(controller='pullrequests', action='create',
                                      repo_name=HG_REPO),
                                  {'org_repo': HG_REPO,
                                   'org_ref': 'branch:stable:4f7e2131323e0749a740c0a56ab68ae9269c562a',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
-                                  '_authentication_token': self.authentication_token(),
+                                  '_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')
         response.mustcontain('href="/vcs_test_hg/changeset/4f7e2131323e0749a740c0a56ab68ae9269c562a"')
     def test_available(self):
         self.log_user()
         response = self.app.post(url(controller='pullrequests', action='create',
                                      repo_name=HG_REPO),
                                  {'org_repo': HG_REPO,
                                   'org_ref': 'rev:94f45ed825a1:94f45ed825a113e61af7e141f44ca578374abef0',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
-                                  '_authentication_token': self.authentication_token(),
+                                  '_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:')
         response.mustcontain('<input id="updaterev_4f7e2131323e0749a740c0a56ab68ae9269c562a" name="updaterev" type="radio" value="4f7e2131323e0749a740c0a56ab68ae9269c562a" />')
         response.mustcontain('href="/vcs_test_hg/changeset/4f7e2131323e0749a740c0a56ab68ae9269c562a"') # as update
     def test_range(self):
         self.log_user()
         response = self.app.post(url(controller='pullrequests', action='create',
                                      repo_name=HG_REPO),
                                  {'org_repo': HG_REPO,
                                   'org_ref': 'branch:stable:4f7e2131323e0749a740c0a56ab68ae9269c562a',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'rev:94f45ed825a1:94f45ed825a113e61af7e141f44ca578374abef0',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
-                                  '_authentication_token': self.authentication_token(),
+                                  '_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"')
     def test_update_reviewers(self):
         self.log_user()
         regular_user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
         regular_user2 = User.get_by_username(TEST_USER_REGULAR2_LOGIN)
         admin_user = User.get_by_username(TEST_USER_ADMIN_LOGIN)
         # create initial PR
         response = self.app.post(url(controller='pullrequests', action='create',
                                      repo_name=HG_REPO),
                                  {'org_repo': HG_REPO,
                                   'org_ref': 'rev:94f45ed825a1:94f45ed825a113e61af7e141f44ca578374abef0',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
-                                  '_authentication_token': self.authentication_token(),
+                                  '_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
         response = self.app.post(url(controller='pullrequests', action='post',
                                      repo_name=HG_REPO, pull_request_id=pull_request1_id),
+                                 {
                                   'updaterev': '4f7e2131323e0749a740c0a56ab68ae9269c562a',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   'owner': TEST_USER_ADMIN_LOGIN,
-                                  '_authentication_token': self.authentication_token(),
+                                  '_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)
         response = response.follow()
         # verify reviewer was added
         response.mustcontain('<input type="hidden" value="%s" name="review_members" />' % regular_user.user_id)
         # update without creating new iteration
         response = self.app.post(url(controller='pullrequests', action='post',
                                      repo_name=HG_REPO, pull_request_id=pull_request2_id),
+                                 {
                                   'pullrequest_title': 'Title',
                                   'pullrequest_desc': 'description',
                                   'owner': TEST_USER_ADMIN_LOGIN,
-                                  '_authentication_token': self.authentication_token(),
+                                  '_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()
         # verify reviewers were added / removed
         response.mustcontain('Meanwhile, the following reviewers have been added: test_regular')
         response.mustcontain('Meanwhile, the following reviewers have been removed: test_admin')
         response.mustcontain('<input type="hidden" value="%s" name="review_members" />' % regular_user.user_id)
         response.mustcontain('<input type="hidden" value="%s" name="review_members" />' % regular_user2.user_id)
         response.mustcontain(no='<input type="hidden" value="%s" name="review_members" />' % admin_user.user_id)
     def test_update_with_invalid_reviewer(self):
         invalid_user_id = 99999
         self.log_user()
         # create a valid pull request
         response = self.app.post(url(controller='pullrequests', action='create',
                                      repo_name=HG_REPO),
+                                 {
                                   'org_repo': HG_REPO,
                                   'org_ref': 'rev:94f45ed825a1:94f45ed825a113e61af7e141f44ca578374abef0',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
-                                  '_authentication_token': self.authentication_token(),
+                                  '_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
         pull_request_id = m.group(1)
         # update it
         response = self.app.post(url(controller='pullrequests', action='post',
                                      repo_name=HG_REPO, pull_request_id=pull_request_id),
+                                 {
                                   'updaterev': '4f7e2131323e0749a740c0a56ab68ae9269c562a',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   'owner': TEST_USER_ADMIN_LOGIN,
-                                  '_authentication_token': self.authentication_token(),
+                                  '_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):
         invalid_user_id = 99999
         self.log_user()
         # create a valid pull request
         response = self.app.post(url(controller='pullrequests', action='create',
                                      repo_name=HG_REPO),
+                                 {
                                   'org_repo': HG_REPO,
                                   'org_ref': 'branch:stable:4f7e2131323e0749a740c0a56ab68ae9269c562a',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
-                                  '_authentication_token': self.authentication_token(),
+                                  '_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
         pull_request_id = m.group(1)
         # edit it
         response = self.app.post(url(controller='pullrequests', action='post',
                                      repo_name=HG_REPO, pull_request_id=pull_request_id),
+                                 {
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   'owner': TEST_USER_ADMIN_LOGIN,
-                                  '_authentication_token': self.authentication_token(),
+                                  '_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):
         # Repo graph excerpt:
         #   o   fb95b340e0d0 webvcs
         #  /:
         # o :   41d2568309a0 default
         # : :
         # : o   5ec21f21aafe webvcs
         # : :
         # : o   9e6119747791 webvcs
         # : :
         # o :   3d1091ee5a53 default
         # :/
         # o     948da46b29c1 default
         self.log_user()
         # create initial PR
         response = self.app.post(
             url(controller='pullrequests', action='create', repo_name=HG_REPO),
+            {
                 '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'
         assert pr1.other_ref == 'branch:default:948da46b29c125838a717f6a8496eb409717078d'
         Session().rollback() # invalidate loaded PR objects before issuing next request.
         # create PR 2 (new iteration with same ancestor)
         response = self.app.post(
             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)
         assert pr2_id != pr1_id
         assert pr1.status == PullRequest.STATUS_CLOSED
         assert pr2.org_ref == 'branch:webvcs:5ec21f21aafe95220f1fc4843a4a57c378498b71'
         assert pr2.other_ref == pr1.other_ref
         Session().rollback() # invalidate loaded PR objects before issuing next request.
         # create PR 3 (new iteration with new ancestor)
         response = self.app.post(
             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)
         assert pr3_id != pr2_id
         assert pr2.status == PullRequest.STATUS_CLOSED
         assert pr3.org_ref == 'branch:webvcs:fb95b340e0d03fa51f33c56c991c08077c99303e'
         assert pr3.other_ref == 'branch:default:41d2568309a05f422cffb8008e599d385f8af439'
 @pytest.mark.usefixtures("test_context_fixture") # apply fixture for all test methods
 class TestPullrequestsGetRepoRefs(TestController):
     def setup_method(self, method):
         self.repo_name = u'main'
         repo = fixture.create_repo(self.repo_name, repo_type='hg')
         self.repo_scm_instance = repo.scm_instance
         Session().commit()
         self.c = PullrequestsController()
     def teardown_method(self, method):
         fixture.destroy_repo(u'main')
         Session().commit()
         Session.remove()
     def test_repo_refs_empty_repo(self):
         # empty repo with no commits, no branches, no bookmarks, just one tag
         refs, default = self.c._get_repo_refs(self.repo_scm_instance)
         assert default == 'tag:null:0000000000000000000000000000000000000000'
     def test_repo_refs_one_commit_no_hints(self):
         cs0 = fixture.commit_change(self.repo_name, filename='file1',
                 content='line1\n', message='commit1', vcs_type='hg',
                 parent=None, newfile=True)
         refs, default = self.c._get_repo_refs(self.repo_scm_instance)
         assert default == 'branch:default:%s' % cs0.raw_id
         assert ([('branch:default:%s' % cs0.raw_id, 'default (current tip)')],
                 'Branches') in refs
     def test_repo_refs_one_commit_rev_hint(self):
         cs0 = fixture.commit_change(self.repo_name, filename='file1',
                 content='line1\n', message='commit1', vcs_type='hg',
                 parent=None, newfile=True)
         refs, default = self.c._get_repo_refs(self.repo_scm_instance, rev=cs0.raw_id)
         expected = 'branch:default:%s' % cs0.raw_id
         assert default == expected
         assert ([(expected, 'default (current tip)')], 'Branches') in refs
     def test_repo_refs_two_commits_no_hints(self):
         cs0 = fixture.commit_change(self.repo_name, filename='file1',
                 content='line1\n', message='commit1', vcs_type='hg',
                 parent=None, newfile=True)
         cs1 = fixture.commit_change(self.repo_name, filename='file2',
                 content='line2\n', message='commit2', vcs_type='hg',
                 parent=None, newfile=True)
         refs, default = self.c._get_repo_refs(self.repo_scm_instance)
         expected = 'branch:default:%s' % cs1.raw_id
         assert default == expected
         assert ([(expected, 'default (current tip)')], 'Branches') in refs
     def test_repo_refs_two_commits_rev_hints(self):
         cs0 = fixture.commit_change(self.repo_name, filename='file1',
                 content='line1\n', message='commit1', vcs_type='hg',
                 parent=None, newfile=True)
         cs1 = fixture.commit_change(self.repo_name, filename='file2',
                 content='line2\n', message='commit2', vcs_type='hg',
                 parent=None, newfile=True)
         refs, default = self.c._get_repo_refs(self.repo_scm_instance, rev=cs0.raw_id)
         expected = 'rev:%s:%s' % (cs0.raw_id, cs0.raw_id)
         assert default == expected
         assert ([(expected, 'Changeset: %s' % cs0.raw_id[0:12])], 'Special') in refs
         assert ([('branch:default:%s' % cs1.raw_id, 'default (current tip)')], 'Branches') in refs
         refs, default = self.c._get_repo_refs(self.repo_scm_instance, rev=cs1.raw_id)
         expected = 'branch:default:%s' % cs1.raw_id
         assert default == expected
         assert ([(expected, 'default (current tip)')], 'Branches') in refs
     def test_repo_refs_two_commits_branch_hint(self):
         cs0 = fixture.commit_change(self.repo_name, filename='file1',
                 content='line1\n', message='commit1', vcs_type='hg',
                 parent=None, newfile=True)
         cs1 = fixture.commit_change(self.repo_name, filename='file2',
                 content='line2\n', message='commit2', vcs_type='hg',
                 parent=None, newfile=True)
         refs, default = self.c._get_repo_refs(self.repo_scm_instance, branch='default')
         expected = 'branch:default:%s' % cs1.raw_id
         assert default == expected
         assert ([(expected, 'default (current tip)')], 'Branches') in refs
     def test_repo_refs_one_branch_no_hints(self):
         cs0 = fixture.commit_change(self.repo_name, filename='file1',
                 content='line1\n', message='commit1', vcs_type='hg',
                 parent=None, newfile=True)
         # TODO

kallithea/tests/functional/test_repo_groups.py

➞

Show inline comments

 from kallithea.tests.base import *
 class TestRepoGroupsController(TestController):
     def test_index(self):
         self.log_user()
         response = self.app.get(url('repos_groups'))
         response.mustcontain('"records": []')
     def test_new(self):
         self.log_user()
         response = self.app.get(url('new_repos_group'))
     def test_create(self):
         self.log_user()
         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<')
         # listing
         response = self.app.get(url('repos_groups'))
         response.mustcontain('raw_name": "%s"' % group_name)
         # show
         response = self.app.get(url('repos_group', group_name=group_name))
         response.mustcontain('href="/_admin/repo_groups/%s/edit"' % group_name)
         # 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)