kallithea Changeset - e9ac5698281d

Changeset - e9ac5698281d

Parent rev.

Child rev.

[Not reviewed]

default

0 53 0

Mads Kiilerich - 9 years ago 2016-12-24 01:27:47
mads@kiilerich.com

tg: minimize future diff by some mocking and replacing some pylons imports with tg

No actual tg dependency yet, just a temporary hack faking tg as an alias for
pylons.

Based on work by Alessandro Molina.

17 files changed:

kallithea/__init__.py

kallithea/config/routing.py

kallithea/controllers/admin/admin.py

kallithea/controllers/admin/auth_settings.py

kallithea/controllers/admin/defaults.py

kallithea/controllers/admin/gists.py

kallithea/controllers/admin/my_account.py

kallithea/controllers/admin/notifications.py

kallithea/controllers/admin/permissions.py

kallithea/controllers/admin/repo_groups.py

kallithea/controllers/admin/repos.py

kallithea/controllers/admin/settings.py

kallithea/controllers/admin/user_groups.py

kallithea/controllers/admin/users.py

kallithea/controllers/api/__init__.py

kallithea/controllers/api/api.py

kallithea/controllers/changelog.py

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

0 comments (0 inline, 0 general)

kallithea/__init__.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/>.
 """
 kallithea
 ~~~~~~~~~
 Kallithea, a web based repository management system.
 Versioning implementation: http://www.python.org/dev/peps/pep-0386/
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 9, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, (C) 2014 Bradley M. Kuhn, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import sys
 import platform
 # temporary aliasing to allow early introduction of imports like 'from tg import request'
 import pylons
 sys.modules['tg'] = pylons
 VERSION = (0, 3, 99)
 BACKENDS = {
     'hg': 'Mercurial repository',
     'git': 'Git repository',
+}
 CELERY_ON = False
 CELERY_EAGER = False
 CONFIG = {}
 # Linked module for extensions
 EXTENSIONS = {}
 try:
     import kallithea.brand
 except ImportError:
     pass
 else:
     assert False, 'Database rebranding is no longer supported; see README.'
 __version__ = '.'.join(str(each) for each in VERSION)
 __platform__ = platform.system()
 __license__ = 'GPLv3'
 __py_version__ = sys.version_info
 __author__ = "Various Authors"
 __url__ = 'https://kallithea-scm.org/'
 is_windows = __platform__ in ['Windows']
 is_unix = not is_windows

kallithea/config/routing.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/>.
 """
 Routes configuration
 The more specific and detailed routes should be defined first so they
 may take precedent over the more generic routes. For more information
 refer to the routes manual at http://routes.groovie.org/docs/
 """
-from pylons import request
+from tg import request
 from routes import Mapper
 # prefix for non repository related links needs to be prefixed with `/`
 ADMIN_PREFIX = '/_admin'
 def make_map(config):
     """Create, configure and return the routes Mapper"""
     rmap = Mapper(directory=config['pylons.paths']['controllers'],
                   always_scan=config['debug'])
     rmap.minimization = False
     rmap.explicit = False
     from kallithea.lib.utils import (is_valid_repo, is_valid_repo_group,
                                      get_repo_by_id)
     def check_repo(environ, match_dict):
         """
         check for valid repository for proper 404 handling
         :param environ:
         :param match_dict:
         """
         repo_name = match_dict.get('repo_name')
         if match_dict.get('f_path'):
             #fix for multiple initial slashes that causes errors
             match_dict['f_path'] = match_dict['f_path'].lstrip('/')
         by_id_match = get_repo_by_id(repo_name)
         if by_id_match:
             repo_name = by_id_match
             match_dict['repo_name'] = repo_name
         return is_valid_repo(repo_name, config['base_path'])
     def check_group(environ, match_dict):
         """
         check for valid repository group for proper 404 handling
         :param environ:
         :param match_dict:
         """
         repo_group_name = match_dict.get('group_name')
         return is_valid_repo_group(repo_group_name, config['base_path'])
     def check_group_skip_path(environ, match_dict):
         """
         check for valid repository group for proper 404 handling, but skips
         verification of existing path
         :param environ:
         :param match_dict:
         """
         repo_group_name = match_dict.get('group_name')
         return is_valid_repo_group(repo_group_name, config['base_path'],
                                    skip_path_check=True)
     def check_user_group(environ, match_dict):
         """
         check for valid user group for proper 404 handling
         :param environ:
         :param match_dict:
         """
         return True
     def check_int(environ, match_dict):
         return match_dict.get('id').isdigit()
     # The ErrorController route (handles 404/500 error pages); it should
     # likely stay at the top, ensuring it can always be resolved
     rmap.connect('/error/{action}', controller='error')
     rmap.connect('/error/{action}/{id}', controller='error')
     #==========================================================================
     # CUSTOM ROUTES HERE
     #==========================================================================
     #MAIN PAGE
     rmap.connect('home', '/', controller='home', action='index')
     rmap.connect('about', '/about', controller='home', action='about')
     rmap.connect('repo_switcher_data', '/_repos', controller='home',
                  action='repo_switcher_data')
     rmap.connect('rst_help',
                  "http://docutils.sourceforge.net/docs/user/rst/quickref.html",
                  _static=True)
     rmap.connect('kallithea_project_url', "https://kallithea-scm.org/", _static=True)
     rmap.connect('issues_url', 'https://bitbucket.org/conservancy/kallithea/issues', _static=True)
     #ADMIN REPOSITORY ROUTES
     with rmap.submapper(path_prefix=ADMIN_PREFIX,
                         controller='admin/repos') as m:
         m.connect("repos", "/repos",
                   action="create", conditions=dict(method=["POST"]))
         m.connect("repos", "/repos",
                   action="index", conditions=dict(method=["GET"]))
         m.connect("new_repo", "/create_repository",
                   action="create_repository", conditions=dict(method=["GET"]))
         m.connect("update_repo", "/repos/{repo_name:.*?}",
                   action="update", conditions=dict(method=["POST"],
                   function=check_repo))
         m.connect("delete_repo", "/repos/{repo_name:.*?}/delete",
                   action="delete", conditions=dict(method=["POST"]))
     #ADMIN REPOSITORY GROUPS ROUTES
     with rmap.submapper(path_prefix=ADMIN_PREFIX,
                         controller='admin/repo_groups') as m:
         m.connect("repos_groups", "/repo_groups",
                   action="create", conditions=dict(method=["POST"]))
         m.connect("repos_groups", "/repo_groups",
                   action="index", conditions=dict(method=["GET"]))
         m.connect("new_repos_group", "/repo_groups/new",
                   action="new", conditions=dict(method=["GET"]))
         m.connect("update_repos_group", "/repo_groups/{group_name:.*?}",
                   action="update", conditions=dict(method=["POST"],
                                                    function=check_group))
         m.connect("repos_group", "/repo_groups/{group_name:.*?}",
                   action="show", conditions=dict(method=["GET"],
                                                  function=check_group))
         #EXTRAS REPO GROUP ROUTES
         m.connect("edit_repo_group", "/repo_groups/{group_name:.*?}/edit",
                   action="edit",
                   conditions=dict(method=["GET"], function=check_group))
         m.connect("edit_repo_group_advanced", "/repo_groups/{group_name:.*?}/edit/advanced",
                   action="edit_repo_group_advanced",
                   conditions=dict(method=["GET"], function=check_group))
         m.connect("edit_repo_group_perms", "/repo_groups/{group_name:.*?}/edit/permissions",
                   action="edit_repo_group_perms",
                   conditions=dict(method=["GET"], function=check_group))
         m.connect("edit_repo_group_perms_update", "/repo_groups/{group_name:.*?}/edit/permissions",
                   action="update_perms",
                   conditions=dict(method=["POST"], function=check_group))
         m.connect("edit_repo_group_perms_delete", "/repo_groups/{group_name:.*?}/edit/permissions/delete",
                   action="delete_perms",
                   conditions=dict(method=["POST"], function=check_group))
         m.connect("delete_repo_group", "/repo_groups/{group_name:.*?}/delete",
                   action="delete", conditions=dict(method=["POST"],
                                                    function=check_group_skip_path))
     #ADMIN USER ROUTES
     with rmap.submapper(path_prefix=ADMIN_PREFIX,
                         controller='admin/users') as m:
         m.connect("new_user", "/users/new",
                   action="create", conditions=dict(method=["POST"]))
         m.connect("users", "/users",
                   action="index", conditions=dict(method=["GET"]))
         m.connect("formatted_users", "/users.{format}",
                   action="index", conditions=dict(method=["GET"]))
         m.connect("new_user", "/users/new",
                   action="new", conditions=dict(method=["GET"]))
         m.connect("update_user", "/users/{id}",
                   action="update", conditions=dict(method=["POST"]))
         m.connect("delete_user", "/users/{id}/delete",
                   action="delete", conditions=dict(method=["POST"]))
         m.connect("edit_user", "/users/{id}/edit",
                   action="edit", conditions=dict(method=["GET"]))
         #EXTRAS USER ROUTES
         m.connect("edit_user_advanced", "/users/{id}/edit/advanced",
                   action="edit_advanced", conditions=dict(method=["GET"]))
         m.connect("edit_user_api_keys", "/users/{id}/edit/api_keys",
                   action="edit_api_keys", conditions=dict(method=["GET"]))
         m.connect("edit_user_api_keys_update", "/users/{id}/edit/api_keys",
                   action="add_api_key", conditions=dict(method=["POST"]))
         m.connect("edit_user_api_keys_delete", "/users/{id}/edit/api_keys/delete",
                   action="delete_api_key", conditions=dict(method=["POST"]))
         m.connect("edit_user_perms", "/users/{id}/edit/permissions",
                   action="edit_perms", conditions=dict(method=["GET"]))
         m.connect("edit_user_perms_update", "/users/{id}/edit/permissions",
                   action="update_perms", conditions=dict(method=["POST"]))
         m.connect("edit_user_emails", "/users/{id}/edit/emails",
                   action="edit_emails", conditions=dict(method=["GET"]))
         m.connect("edit_user_emails_update", "/users/{id}/edit/emails",
                   action="add_email", conditions=dict(method=["POST"]))
         m.connect("edit_user_emails_delete", "/users/{id}/edit/emails/delete",
                   action="delete_email", conditions=dict(method=["POST"]))
         m.connect("edit_user_ips", "/users/{id}/edit/ips",
                   action="edit_ips", conditions=dict(method=["GET"]))
         m.connect("edit_user_ips_update", "/users/{id}/edit/ips",
                   action="add_ip", conditions=dict(method=["POST"]))
         m.connect("edit_user_ips_delete", "/users/{id}/edit/ips/delete",
                   action="delete_ip", conditions=dict(method=["POST"]))
     #ADMIN USER GROUPS REST ROUTES
     with rmap.submapper(path_prefix=ADMIN_PREFIX,
                         controller='admin/user_groups') as m:
         m.connect("users_groups", "/user_groups",
                   action="create", conditions=dict(method=["POST"]))
         m.connect("users_groups", "/user_groups",
                   action="index", conditions=dict(method=["GET"]))
         m.connect("new_users_group", "/user_groups/new",
                   action="new", conditions=dict(method=["GET"]))
         m.connect("update_users_group", "/user_groups/{id}",
                   action="update", conditions=dict(method=["POST"]))
         m.connect("delete_users_group", "/user_groups/{id}/delete",
                   action="delete", conditions=dict(method=["POST"]))
         m.connect("edit_users_group", "/user_groups/{id}/edit",
                   action="edit", conditions=dict(method=["GET"]),
                   function=check_user_group)
         #EXTRAS USER GROUP ROUTES
         m.connect("edit_user_group_default_perms", "/user_groups/{id}/edit/default_perms",
                   action="edit_default_perms", conditions=dict(method=["GET"]))
         m.connect("edit_user_group_default_perms_update", "/user_groups/{id}/edit/default_perms",
                   action="update_default_perms", conditions=dict(method=["POST"]))
         m.connect("edit_user_group_perms", "/user_groups/{id}/edit/perms",
                   action="edit_perms", conditions=dict(method=["GET"]))
         m.connect("edit_user_group_perms_update", "/user_groups/{id}/edit/perms",
                   action="update_perms", conditions=dict(method=["POST"]))
         m.connect("edit_user_group_perms_delete", "/user_groups/{id}/edit/perms/delete",
                   action="delete_perms", conditions=dict(method=["POST"]))
         m.connect("edit_user_group_advanced", "/user_groups/{id}/edit/advanced",
                   action="edit_advanced", conditions=dict(method=["GET"]))
         m.connect("edit_user_group_members", "/user_groups/{id}/edit/members",
                   action="edit_members", conditions=dict(method=["GET"]))
     #ADMIN PERMISSIONS ROUTES
     with rmap.submapper(path_prefix=ADMIN_PREFIX,
                         controller='admin/permissions') as m:
         m.connect("admin_permissions", "/permissions",
                   action="permission_globals", conditions=dict(method=["POST"]))
         m.connect("admin_permissions", "/permissions",
                   action="permission_globals", conditions=dict(method=["GET"]))
         m.connect("admin_permissions_ips", "/permissions/ips",
                   action="permission_ips", conditions=dict(method=["GET"]))
         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"]))
     #NOTIFICATION REST ROUTES
     with rmap.submapper(path_prefix=ADMIN_PREFIX,
                         controller='admin/notifications') as m:
         m.connect("notifications", "/notifications",
                   action="index", conditions=dict(method=["GET"]))
         m.connect("notifications_mark_all_read", "/notifications/mark_all_read",
                   action="mark_all_read", conditions=dict(method=["GET"]))
         m.connect("formatted_notifications", "/notifications.{format}",
                   action="index", conditions=dict(method=["GET"]))
         m.connect("notification_update", "/notifications/{notification_id}/update",
                   action="update", conditions=dict(method=["POST"]))
         m.connect("notification_delete", "/notifications/{notification_id}/delete",
                   action="delete", conditions=dict(method=["POST"]))
         m.connect("notification", "/notifications/{notification_id}",
                   action="show", conditions=dict(method=["GET"]))
         m.connect("formatted_notification", "/notifications/{notification_id}.{format}",
                   action="show", conditions=dict(method=["GET"]))
     #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') 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('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_locking", "/{repo_name:.*?}/settings/advanced/locking",
                  controller='admin/repos', action="edit_advanced_locking",
                  conditions=dict(method=["POST"], function=check_repo))
     rmap.connect('toggle_locking', "/{repo_name:.*?}/settings/advanced/locking_toggle",
                  controller='admin/repos', action="toggle_locking",
                  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"]))
     rmap.connect('pullrequest_delete',
                  '/{repo_name:.*?}/pull-request/{pull_request_id}/delete',
                  controller='pullrequests',
                  action='delete', conditions=dict(function=check_repo,
                                                   method=["POST"]))
     rmap.connect('pullrequest_show_all',
                  '/{repo_name:.*?}/pull-request',
                  controller='pullrequests',
                  action='show_all', conditions=dict(function=check_repo,
                                                 method=["GET"]))
     rmap.connect('my_pullrequests',
                  '/my_pullrequests',
                  controller='pullrequests',
                  action='show_my', conditions=dict(method=["GET"]))
     rmap.connect('pullrequest_comment',
                  '/{repo_name:.*?}/pull-request-comment/{pull_request_id}',
                  controller='pullrequests',
                  action='comment', conditions=dict(function=check_repo,
                                                 method=["POST"]))
     rmap.connect('pullrequest_comment_delete',
                  '/{repo_name:.*?}/pull-request-comment/{comment_id}/delete',
                 controller='pullrequests', action='delete_comment',
                 conditions=dict(function=check_repo, method=["POST"]))
     rmap.connect('summary_home_summary', '/{repo_name:.*?}/summary',
                 controller='summary', conditions=dict(function=check_repo))
     rmap.connect('changelog_home', '/{repo_name:.*?}/changelog',
                 controller='changelog', conditions=dict(function=check_repo))
     rmap.connect('changelog_summary_home', '/{repo_name:.*?}/changelog_summary',
                 controller='changelog', action='changelog_summary',
                 conditions=dict(function=check_repo))
     rmap.connect('changelog_file_home', '/{repo_name:.*?}/changelog/{revision}/{f_path:.*}',
                 controller='changelog', f_path=None,
                 conditions=dict(function=check_repo))
     rmap.connect('changelog_details', '/{repo_name:.*?}/changelog_details/{cs}',
                 controller='changelog', action='changelog_details',
                 conditions=dict(function=check_repo))
     rmap.connect('files_home', '/{repo_name:.*?}/files/{revision}/{f_path:.*}',
                 controller='files', revision='tip', f_path='',
                 conditions=dict(function=check_repo))
     rmap.connect('files_home_nopath', '/{repo_name:.*?}/files/{revision}',
                 controller='files', revision='tip', f_path='',
                 conditions=dict(function=check_repo))
     rmap.connect('files_history_home',
                  '/{repo_name:.*?}/history/{revision}/{f_path:.*}',
                  controller='files', action='history', revision='tip', f_path='',
                  conditions=dict(function=check_repo))
     rmap.connect('files_authors_home',
                  '/{repo_name:.*?}/authors/{revision}/{f_path:.*}',
                  controller='files', action='authors', revision='tip', f_path='',
                  conditions=dict(function=check_repo))
     rmap.connect('files_diff_home', '/{repo_name:.*?}/diff/{f_path:.*}',
                 controller='files', action='diff', revision='tip', f_path='',
                 conditions=dict(function=check_repo))
     rmap.connect('files_diff_2way_home', '/{repo_name:.*?}/diff-2way/{f_path:.+}',
                 controller='files', action='diff_2way', revision='tip', f_path='',
                 conditions=dict(function=check_repo))
     rmap.connect('files_rawfile_home',
                  '/{repo_name:.*?}/rawfile/{revision}/{f_path:.*}',
                  controller='files', action='rawfile', revision='tip',
                  f_path='', conditions=dict(function=check_repo))
     rmap.connect('files_raw_home',
                  '/{repo_name:.*?}/raw/{revision}/{f_path:.*}',
                  controller='files', action='raw', revision='tip', f_path='',
                  conditions=dict(function=check_repo))
     rmap.connect('files_annotate_home',
                  '/{repo_name:.*?}/annotate/{revision}/{f_path:.*}',
                  controller='files', action='index', revision='tip',
                  f_path='', annotate=True, conditions=dict(function=check_repo))
     rmap.connect('files_edit_home',
                  '/{repo_name:.*?}/edit/{revision}/{f_path:.*}',
                  controller='files', action='edit', revision='tip',
                  f_path='', conditions=dict(function=check_repo))
     rmap.connect('files_add_home',
                  '/{repo_name:.*?}/add/{revision}/{f_path:.*}',
                  controller='files', action='add', revision='tip',
                  f_path='', conditions=dict(function=check_repo))
     rmap.connect('files_delete_home',
                  '/{repo_name:.*?}/delete/{revision}/{f_path:.*}',
                  controller='files', action='delete', revision='tip',
                  f_path='', conditions=dict(function=check_repo))
     rmap.connect('files_archive_home', '/{repo_name:.*?}/archive/{fname}',
                 controller='files', action='archivefile',
                 conditions=dict(function=check_repo))
     rmap.connect('files_nodelist_home',
                  '/{repo_name:.*?}/nodelist/{revision}/{f_path:.*}',
                 controller='files', action='nodelist',
                 conditions=dict(function=check_repo))
     rmap.connect('repo_fork_create_home', '/{repo_name:.*?}/fork',
                 controller='forks', action='fork_create',
                 conditions=dict(function=check_repo, method=["POST"]))

kallithea/controllers/admin/admin.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/>.
 """
 kallithea.controllers.admin.admin
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Controller for Admin panel of Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 7, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
-from pylons import request, tmpl_context as c
+from tg import request, tmpl_context as c
 from sqlalchemy.orm import joinedload
 from whoosh.qparser.default import QueryParser
 from whoosh.qparser.dateparse import DateParserPlugin
 from whoosh import query
 from sqlalchemy.sql.expression import or_, and_, func
 from kallithea.config.routing import url
 from kallithea.model.db import UserLog
 from kallithea.lib.auth import LoginRequired, HasPermissionAnyDecorator
 from kallithea.lib.base import BaseController, render
 from kallithea.lib.utils2 import safe_int, remove_prefix, remove_suffix
 from kallithea.lib.indexers import JOURNAL_SCHEMA
 from kallithea.lib.page import Page
 log = logging.getLogger(__name__)
 def _journal_filter(user_log, search_term):
     """
     Filters sqlalchemy user_log based on search_term with whoosh Query language
     http://packages.python.org/Whoosh/querylang.html
     :param user_log:
     :param search_term:
     """
     log.debug('Initial search term: %r', search_term)
     qry = None
     if search_term:
         qp = QueryParser('repository', schema=JOURNAL_SCHEMA)
         qp.add_plugin(DateParserPlugin())
         qry = qp.parse(unicode(search_term))
         log.debug('Filtering using parsed query %r', qry)
     def wildcard_handler(col, wc_term):
         if wc_term.startswith('*') and not wc_term.endswith('*'):
             #postfix == endswith
             wc_term = remove_prefix(wc_term, prefix='*')
             return func.lower(col).endswith(func.lower(wc_term))
         elif wc_term.startswith('*') and wc_term.endswith('*'):
             #wildcard == ilike
             wc_term = remove_prefix(wc_term, prefix='*')
             wc_term = remove_suffix(wc_term, suffix='*')
             return func.lower(col).contains(func.lower(wc_term))
     def get_filterion(field, val, term):
         if field == 'repository':
             field = getattr(UserLog, 'repository_name')
         elif field == 'ip':
             field = getattr(UserLog, 'user_ip')
         elif field == 'date':
             field = getattr(UserLog, 'action_date')
         elif field == 'username':
             field = getattr(UserLog, 'username')
         else:
             field = getattr(UserLog, field)
         log.debug('filter field: %s val=>%s', field, val)
         #sql filtering
         if isinstance(term, query.Wildcard):
             return wildcard_handler(field, val)
         elif isinstance(term, query.Prefix):
             return func.lower(field).startswith(func.lower(val))
         elif isinstance(term, query.DateRange):
             return and_(field >= val[0], field <= val[1])
         return func.lower(field) == func.lower(val)
     if isinstance(qry, (query.And, query.Term, query.Prefix, query.Wildcard,
                         query.DateRange)):
         if not isinstance(qry, query.And):
             qry = [qry]
         for term in qry:
             field = term.fieldname
             val = (term.text if not isinstance(term, query.DateRange)
                    else [term.startdate, term.enddate])
             user_log = user_log.filter(get_filterion(field, val, term))
     elif isinstance(qry, query.Or):
         filters = []
         for term in qry:
             field = term.fieldname
             val = (term.text if not isinstance(term, query.DateRange)
                    else [term.startdate, term.enddate])
             filters.append(get_filterion(field, val, term))
         user_log = user_log.filter(or_(*filters))
     return user_log
 class AdminController(BaseController):
     @LoginRequired()
     def __before__(self):
         super(AdminController, self).__before__()
     @HasPermissionAnyDecorator('hg.admin')
     def index(self):
         users_log = UserLog.query() \
                 .options(joinedload(UserLog.user)) \
                 .options(joinedload(UserLog.repository))
         #FILTERING
         c.search_term = request.GET.get('filter')
         users_log = _journal_filter(users_log, c.search_term)
         users_log = users_log.order_by(UserLog.action_date.desc())
         p = safe_int(request.GET.get('page'), 1)
         def url_generator(**kw):
             return url.current(filter=c.search_term, **kw)
         c.users_log = Page(users_log, page=p, items_per_page=10, url=url_generator)
         if request.environ.get('HTTP_X_PARTIAL_XHR'):
             return render('admin/admin_log.html')
         return render('admin/admin.html')

kallithea/controllers/admin/auth_settings.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/>.
 """
 kallithea.controllers.admin.auth_settings
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 pluggable authentication controller for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Nov 26, 2010
 :author: akesterson
 """
 import logging
 import formencode.htmlfill
 import traceback
 from pylons import request, tmpl_context as c
 from pylons.i18n.translation import _
 from tg import request, tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPFound
 from kallithea.config.routing import url
 from kallithea.lib import helpers as h
 from kallithea.lib.compat import formatted_json
 from kallithea.lib.base import BaseController, render
 from kallithea.lib.auth import LoginRequired, HasPermissionAnyDecorator
 from kallithea.lib import auth_modules
 from kallithea.model.forms import AuthSettingsForm
 from kallithea.model.db import Setting
 from kallithea.model.meta import Session
 log = logging.getLogger(__name__)
 class AuthSettingsController(BaseController):
     @LoginRequired()
     @HasPermissionAnyDecorator('hg.admin')
     def __before__(self):
         super(AuthSettingsController, self).__before__()
     def __load_defaults(self):
         c.available_plugins = [
             'kallithea.lib.auth_modules.auth_internal',
             'kallithea.lib.auth_modules.auth_container',
             'kallithea.lib.auth_modules.auth_ldap',
             'kallithea.lib.auth_modules.auth_crowd',
             'kallithea.lib.auth_modules.auth_pam'
+        ]
         c.enabled_plugins = Setting.get_auth_plugins()
     def __render(self, defaults, errors):
         c.defaults = {}
         c.plugin_settings = {}
         c.plugin_shortnames = {}
         for module in c.enabled_plugins:
             plugin = auth_modules.loadplugin(module)
             plugin_name = plugin.name
             c.plugin_shortnames[module] = plugin_name
             c.plugin_settings[module] = plugin.plugin_settings()
             for v in c.plugin_settings[module]:
                 fullname = ("auth_" + plugin_name + "_" + v["name"])
                 if "default" in v:
                     c.defaults[fullname] = v["default"]
                 # Current values will be the default on the form, if there are any
                 setting = Setting.get_by_name(fullname)
                 if setting is not None:
                     c.defaults[fullname] = setting.app_settings_value
         # we want to show , separated list of enabled plugins
         c.defaults['auth_plugins'] = ','.join(c.enabled_plugins)
         if defaults:
             c.defaults.update(defaults)
         log.debug(formatted_json(defaults))
         return formencode.htmlfill.render(
             render('admin/auth/auth_settings.html'),
             defaults=c.defaults,
             errors=errors,
             prefix_error=False,
             encoding="UTF-8",
             force_defaults=False)
     def index(self):
         self.__load_defaults()
         return self.__render(defaults=None, errors=None)
     def auth_settings(self):
         """POST create and store auth settings"""
         self.__load_defaults()
         log.debug("POST Result: %s", formatted_json(dict(request.POST)))
         # First, parse only the plugin list (not the plugin settings).
         _auth_plugins_validator = AuthSettingsForm([]).fields['auth_plugins']
         try:
             new_enabled_plugins = _auth_plugins_validator.to_python(request.POST.get('auth_plugins'))
         except formencode.Invalid:
             # User provided an invalid plugin list. Just fall back to
             # the list of currently enabled plugins. (We'll re-validate
             # and show an error message to the user, below.)
             pass
         else:
             # Hide plugins that the user has asked to be disabled, but
             # do not show plugins that the user has asked to be enabled
             # (yet), since that'll cause validation errors and/or wrong
             # settings being applied (e.g. checkboxes being cleared),
             # since the plugin settings will not be in the POST data.
             c.enabled_plugins = [ p for p in c.enabled_plugins if p in new_enabled_plugins ]
         # Next, parse everything including plugin settings.
         _form = AuthSettingsForm(c.enabled_plugins)()
         try:
             form_result = _form.to_python(dict(request.POST))
             for k, v in form_result.items():
                 if k == 'auth_plugins':
                     # we want to store it comma separated inside our settings
                     v = ','.join(v)
                 log.debug("%s = %s", k, str(v))
                 setting = Setting.create_or_update(k, v)
             Session().commit()
             h.flash(_('Auth settings updated successfully'),
                        category='success')
         except formencode.Invalid as errors:
             log.error(traceback.format_exc())
             e = errors.error_dict or {}
             return self.__render(
                 defaults=errors.value,
                 errors=e,
+            )
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('error occurred during update of auth settings'),
                     category='error')
         raise HTTPFound(location=url('auth_home'))

kallithea/controllers/admin/defaults.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/>.
 """
 kallithea.controllers.admin.defaults
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 default settings controller for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 27, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 import formencode
 from formencode import htmlfill
 from pylons import request, tmpl_context as c
 from pylons.i18n.translation import _
 from tg import request, tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPFound
 from kallithea.config.routing import url
 from kallithea.lib import helpers as h
 from kallithea.lib.auth import LoginRequired, HasPermissionAnyDecorator
 from kallithea.lib.base import BaseController, render
 from kallithea.model.forms import DefaultsForm
 from kallithea.model.meta import Session
 from kallithea import BACKENDS
 from kallithea.model.db import Setting
 log = logging.getLogger(__name__)
 class DefaultsController(BaseController):
     @LoginRequired()
     @HasPermissionAnyDecorator('hg.admin')
     def __before__(self):
         super(DefaultsController, self).__before__()
     def index(self, format='html'):
         c.backends = BACKENDS.keys()
         defaults = Setting.get_default_repo_settings()
         return htmlfill.render(
             render('admin/defaults/defaults.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False
+        )
     def update(self, id):
         _form = DefaultsForm()()
         try:
             form_result = _form.to_python(dict(request.POST))
             for k, v in form_result.iteritems():
                 setting = Setting.create_or_update(k, v)
             Session().commit()
             h.flash(_('Default settings updated successfully'),
                     category='success')
         except formencode.Invalid as errors:
             defaults = errors.value
             return htmlfill.render(
                 render('admin/defaults/defaults.html'),
                 defaults=defaults,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False)
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('Error occurred during update of defaults'),
                     category='error')
         raise HTTPFound(location=url('defaults'))

kallithea/controllers/admin/gists.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/>.
 """
 kallithea.controllers.admin.gists
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 gist controller for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: May 9, 2013
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import time
 import logging
 import traceback
 import formencode.htmlfill
 from pylons import request, response, tmpl_context as c
 from pylons.i18n.translation import _
 from tg import request, response, tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPFound, HTTPNotFound, HTTPForbidden
 from kallithea.config.routing import url
 from kallithea.model.forms import GistForm
 from kallithea.model.gist import GistModel
 from kallithea.model.meta import Session
 from kallithea.model.db import Gist, User
 from kallithea.lib import helpers as h
 from kallithea.lib.base import BaseController, render, jsonify
 from kallithea.lib.auth import LoginRequired, NotAnonymous
 from kallithea.lib.utils2 import safe_int, safe_unicode, time_to_datetime
 from kallithea.lib.page import Page
 from sqlalchemy.sql.expression import or_
 from kallithea.lib.vcs.exceptions import VCSError, NodeNotChangedError
 log = logging.getLogger(__name__)
 class GistsController(BaseController):
     """REST Controller styled on the Atom Publishing Protocol"""
     def __load_defaults(self, extra_values=None):
         c.lifetime_values = [
             (str(-1), _('Forever')),
             (str(5), _('5 minutes')),
             (str(60), _('1 hour')),
             (str(60 * 24), _('1 day')),
             (str(60 * 24 * 30), _('1 month')),
+        ]
         if extra_values:
             c.lifetime_values.append(extra_values)
         c.lifetime_options = [(c.lifetime_values, _("Lifetime"))]
     @LoginRequired()
     def index(self):
         not_default_user = not request.authuser.is_default_user
         c.show_private = request.GET.get('private') and not_default_user
         c.show_public = request.GET.get('public') and not_default_user
         gists = Gist().query() \
             .filter_by(is_expired=False) \
             .order_by(Gist.created_on.desc())
         # MY private
         if c.show_private and not c.show_public:
             gists = gists.filter(Gist.gist_type == Gist.GIST_PRIVATE) \
                              .filter(Gist.owner_id == request.authuser.user_id)
         # MY public
         elif c.show_public and not c.show_private:
             gists = gists.filter(Gist.gist_type == Gist.GIST_PUBLIC) \
                              .filter(Gist.owner_id == request.authuser.user_id)
         # MY public+private
         elif c.show_private and c.show_public:
             gists = gists.filter(or_(Gist.gist_type == Gist.GIST_PUBLIC,
                                      Gist.gist_type == Gist.GIST_PRIVATE)) \
                              .filter(Gist.owner_id == request.authuser.user_id)
         # default show ALL public gists
         if not c.show_public and not c.show_private:
             gists = gists.filter(Gist.gist_type == Gist.GIST_PUBLIC)
         c.gists = gists
         p = safe_int(request.GET.get('page'), 1)
         c.gists_pager = Page(c.gists, page=p, items_per_page=10)
         return render('admin/gists/index.html')
     @LoginRequired()
     @NotAnonymous()
     def create(self):
         self.__load_defaults()
         gist_form = GistForm([x[0] for x in c.lifetime_values])()
         try:
             form_result = gist_form.to_python(dict(request.POST))
             #TODO: multiple files support, from the form
             filename = form_result['filename'] or Gist.DEFAULT_FILENAME
             nodes = {
                 filename: {
                     'content': form_result['content'],
                     'lexer': form_result['mimetype']  # None is autodetect
+                }
+            }
             _public = form_result['public']
             gist_type = Gist.GIST_PUBLIC if _public else Gist.GIST_PRIVATE
             gist = GistModel().create(
                 description=form_result['description'],
                 owner=request.authuser.user_id,
                 gist_mapping=nodes,
                 gist_type=gist_type,
                 lifetime=form_result['lifetime']
+            )
             Session().commit()
             new_gist_id = gist.gist_access_id
         except formencode.Invalid as errors:
             defaults = errors.value
             return formencode.htmlfill.render(
                 render('admin/gists/new.html'),
                 defaults=defaults,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False)
         except Exception as e:
             log.error(traceback.format_exc())
             h.flash(_('Error occurred during gist creation'), category='error')
             raise HTTPFound(location=url('new_gist'))
         raise HTTPFound(location=url('gist', gist_id=new_gist_id))
     @LoginRequired()
     @NotAnonymous()
     def new(self, format='html'):
         self.__load_defaults()
         return render('admin/gists/new.html')
     @LoginRequired()
     @NotAnonymous()
     def delete(self, gist_id):
         gist = GistModel().get_gist(gist_id)
         owner = gist.owner_id == request.authuser.user_id
         if h.HasPermissionAny('hg.admin')() or owner:
             GistModel().delete(gist)
             Session().commit()
             h.flash(_('Deleted gist %s') % gist.gist_access_id, category='success')
         else:
             raise HTTPForbidden()
         raise HTTPFound(location=url('gists'))
     @LoginRequired()
     def show(self, gist_id, revision='tip', format='html', f_path=None):
         c.gist = Gist.get_or_404(gist_id)
         if c.gist.is_expired:
             log.error('Gist expired at %s',
                       time_to_datetime(c.gist.gist_expires))
             raise HTTPNotFound()
         try:
             c.file_changeset, c.files = GistModel().get_gist_files(gist_id,
                                                             revision=revision)
         except VCSError:
             log.error(traceback.format_exc())
             raise HTTPNotFound()
         if format == 'raw':
             content = '\n\n'.join([f.content for f in c.files if (f_path is None or safe_unicode(f.path) == f_path)])
             response.content_type = 'text/plain'
             return content
         return render('admin/gists/show.html')
     @LoginRequired()
     @NotAnonymous()
     def edit(self, gist_id, format='html'):
         c.gist = Gist.get_or_404(gist_id)
         if c.gist.is_expired:
             log.error('Gist expired at %s',
                       time_to_datetime(c.gist.gist_expires))
             raise HTTPNotFound()
         try:
             c.file_changeset, c.files = GistModel().get_gist_files(gist_id)
         except VCSError:
             log.error(traceback.format_exc())
             raise HTTPNotFound()
         self.__load_defaults(extra_values=('0', _('Unmodified')))
         rendered = render('admin/gists/edit.html')
         if request.POST:
             rpost = request.POST
             nodes = {}
             for org_filename, filename, mimetype, content in zip(
                                                     rpost.getall('org_files'),
                                                     rpost.getall('files'),
                                                     rpost.getall('mimetypes'),
                                                     rpost.getall('contents')):
                 nodes[org_filename] = {
                     'org_filename': org_filename,
                     'filename': filename,
                     'content': content,
                     'lexer': mimetype,
+                }
             try:
                 GistModel().update(
                     gist=c.gist,
                     description=rpost['description'],
                     owner=c.gist.owner,
                     gist_mapping=nodes,
                     gist_type=c.gist.gist_type,
                     lifetime=rpost['lifetime']
+                )
                 Session().commit()
                 h.flash(_('Successfully updated gist content'), category='success')
             except NodeNotChangedError:
                 # raised if nothing was changed in repo itself. We anyway then
                 # store only DB stuff for gist
                 Session().commit()
                 h.flash(_('Successfully updated gist data'), category='success')
             except Exception:
                 log.error(traceback.format_exc())
                 h.flash(_('Error occurred during update of gist %s') % gist_id,
                         category='error')
             raise HTTPFound(location=url('gist', gist_id=gist_id))
         return rendered
     @LoginRequired()
     @NotAnonymous()
     @jsonify
     def check_revision(self, gist_id):
         c.gist = Gist.get_or_404(gist_id)
         last_rev = c.gist.scm_instance.get_changeset()
         success = True
         revision = request.POST.get('revision')
         ##TODO: maybe move this to model ?
         if revision != last_rev.raw_id:
             log.error('Last revision %s is different than submitted %s',
                       revision, last_rev)
             # our gist has newer version than we
             success = False
         return {'success': success}

kallithea/controllers/admin/my_account.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/>.
 """
 kallithea.controllers.admin.my_account
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 my account controller for Kallithea admin
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: August 20, 2013
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 import formencode
 from sqlalchemy import func
 from formencode import htmlfill
 from pylons import request, tmpl_context as c
 from pylons.i18n.translation import _
 from tg import request, tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPFound
 from kallithea.config.routing import url
 from kallithea.lib import helpers as h
 from kallithea.lib import auth_modules
 from kallithea.lib.auth import LoginRequired, NotAnonymous, AuthUser
 from kallithea.lib.base import BaseController, render
 from kallithea.lib.utils2 import generate_api_key, safe_int
 from kallithea.model.db import Repository, UserEmailMap, User, UserFollowing
 from kallithea.model.forms import UserForm, PasswordChangeForm
 from kallithea.model.user import UserModel
 from kallithea.model.repo import RepoModel
 from kallithea.model.api_key import ApiKeyModel
 from kallithea.model.meta import Session
 log = logging.getLogger(__name__)
 class MyAccountController(BaseController):
     """REST Controller styled on the Atom Publishing Protocol"""
     # To properly map this controller, ensure your config/routing.py
     # file has a resource setup:
     #     map.resource('setting', 'settings', controller='admin/settings',
     #         path_prefix='/admin', name_prefix='admin_')
     @LoginRequired()
     @NotAnonymous()
     def __before__(self):
         super(MyAccountController, self).__before__()
     def __load_data(self):
         c.user = User.get(request.authuser.user_id)
         if c.user.is_default_user:
             h.flash(_("You can't edit this user since it's"
                       " crucial for entire application"), category='warning')
             raise HTTPFound(location=url('users'))
     def _load_my_repos_data(self, watched=False):
         if watched:
             admin = False
             repos_list = Session().query(Repository) \
                          .join(UserFollowing) \
                          .filter(UserFollowing.user_id ==
                                  request.authuser.user_id).all()
         else:
             admin = True
             repos_list = Session().query(Repository) \
                          .filter(Repository.owner_id ==
                                  request.authuser.user_id).all()
         return RepoModel().get_repos_as_dict(repos_list=repos_list,
                                                    admin=admin)
     def my_account(self):
         c.active = 'profile'
         self.__load_data()
         c.perm_user = AuthUser(user_id=request.authuser.user_id)
         managed_fields = auth_modules.get_managed_fields(c.user)
         def_user_perms = User.get_default_user().AuthUser.permissions['global']
         if 'hg.register.none' in def_user_perms:
             managed_fields.extend(['username', 'firstname', 'lastname', 'email'])
         c.readonly = lambda n: 'readonly' if n in managed_fields else None
         defaults = c.user.get_dict()
         update = False
         if request.POST:
             _form = UserForm(edit=True,
                              old_data={'user_id': request.authuser.user_id,
                                        'email': request.authuser.email})()
             form_result = {}
             try:
                 post_data = dict(request.POST)
                 post_data['new_password'] = ''
                 post_data['password_confirmation'] = ''
                 form_result = _form.to_python(post_data)
                 # skip updating those attrs for my account
                 skip_attrs = ['admin', 'active', 'extern_type', 'extern_name',
                               'new_password', 'password_confirmation',
                              ] + managed_fields
                 UserModel().update(request.authuser.user_id, form_result,
                                    skip_attrs=skip_attrs)
                 h.flash(_('Your account was updated successfully'),
                         category='success')
                 Session().commit()
                 update = True
             except formencode.Invalid as errors:
                 return htmlfill.render(
                     render('admin/my_account/my_account.html'),
                     defaults=errors.value,
                     errors=errors.error_dict or {},
                     prefix_error=False,
                     encoding="UTF-8",
                     force_defaults=False)
             except Exception:
                 log.error(traceback.format_exc())
                 h.flash(_('Error occurred during update of user %s') \
                         % form_result.get('username'), category='error')
         if update:
             raise HTTPFound(location='my_account')
         return htmlfill.render(
             render('admin/my_account/my_account.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     def my_account_password(self):
         c.active = 'password'
         self.__load_data()
         managed_fields = auth_modules.get_managed_fields(c.user)
         c.can_change_password = 'password' not in managed_fields
         if request.POST and c.can_change_password:
             _form = PasswordChangeForm(request.authuser.username)()
             try:
                 form_result = _form.to_python(request.POST)
                 UserModel().update(request.authuser.user_id, form_result)
                 Session().commit()
                 h.flash(_("Successfully updated password"), category='success')
             except formencode.Invalid as errors:
                 return htmlfill.render(
                     render('admin/my_account/my_account.html'),
                     defaults=errors.value,
                     errors=errors.error_dict or {},
                     prefix_error=False,
                     encoding="UTF-8",
                     force_defaults=False)
             except Exception:
                 log.error(traceback.format_exc())
                 h.flash(_('Error occurred during update of user password'),
                         category='error')
         return render('admin/my_account/my_account.html')
     def my_account_repos(self):
         c.active = 'repos'
         self.__load_data()
         #data used to render the grid
         c.data = self._load_my_repos_data()
         return render('admin/my_account/my_account.html')
     def my_account_watched(self):
         c.active = 'watched'
         self.__load_data()
         #data used to render the grid
         c.data = self._load_my_repos_data(watched=True)
         return render('admin/my_account/my_account.html')
     def my_account_perms(self):
         c.active = 'perms'
         self.__load_data()
         c.perm_user = AuthUser(user_id=request.authuser.user_id)
         return render('admin/my_account/my_account.html')
     def my_account_emails(self):
         c.active = 'emails'
         self.__load_data()
         c.user_email_map = UserEmailMap.query() \
             .filter(UserEmailMap.user == c.user).all()
         return render('admin/my_account/my_account.html')
     def my_account_emails_add(self):
         email = request.POST.get('new_email')
         try:
             UserModel().add_extra_email(request.authuser.user_id, email)
             Session().commit()
             h.flash(_("Added email %s to user") % email, category='success')
         except formencode.Invalid as error:
             msg = error.error_dict['email']
             h.flash(msg, category='error')
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('An error occurred during email saving'),
                     category='error')
         raise HTTPFound(location=url('my_account_emails'))
     def my_account_emails_delete(self):
         email_id = request.POST.get('del_email_id')
         user_model = UserModel()
         user_model.delete_extra_email(request.authuser.user_id, email_id)
         Session().commit()
         h.flash(_("Removed email from user"), category='success')
         raise HTTPFound(location=url('my_account_emails'))
     def my_account_api_keys(self):
         c.active = 'api_keys'
         self.__load_data()
         show_expired = True
         c.lifetime_values = [
             (str(-1), _('Forever')),
             (str(5), _('5 minutes')),
             (str(60), _('1 hour')),
             (str(60 * 24), _('1 day')),
             (str(60 * 24 * 30), _('1 month')),
+        ]
         c.lifetime_options = [(c.lifetime_values, _("Lifetime"))]
         c.user_api_keys = ApiKeyModel().get_api_keys(request.authuser.user_id,
                                                      show_expired=show_expired)
         return render('admin/my_account/my_account.html')
     def my_account_api_keys_add(self):
         lifetime = safe_int(request.POST.get('lifetime'), -1)
         description = request.POST.get('description')
         ApiKeyModel().create(request.authuser.user_id, description, lifetime)
         Session().commit()
         h.flash(_("API key successfully created"), category='success')
         raise HTTPFound(location=url('my_account_api_keys'))
     def my_account_api_keys_delete(self):
         api_key = request.POST.get('del_api_key')
         if request.POST.get('del_api_key_builtin'):
             user = User.get(request.authuser.user_id)
             user.api_key = generate_api_key()
             Session().commit()
             h.flash(_("API key successfully reset"), category='success')
         elif api_key:
             ApiKeyModel().delete(api_key, request.authuser.user_id)
             Session().commit()
             h.flash(_("API key successfully deleted"), category='success')
         raise HTTPFound(location=url('my_account_api_keys'))

kallithea/controllers/admin/notifications.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/>.
 """
 kallithea.controllers.admin.notifications
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 notifications controller for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Nov 23, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 from pylons import request
 from pylons import tmpl_context as c
 from tg import request
 from tg import tmpl_context as c
 from webob.exc import HTTPBadRequest, HTTPForbidden
 from kallithea.model.db import Notification
 from kallithea.model.notification import NotificationModel
 from kallithea.model.meta import Session
 from kallithea.lib.auth import LoginRequired, NotAnonymous
 from kallithea.lib.base import BaseController, render
 from kallithea.lib import helpers as h
 from kallithea.lib.page import Page
 from kallithea.lib.utils2 import safe_int
 log = logging.getLogger(__name__)
 class NotificationsController(BaseController):
     """REST Controller styled on the Atom Publishing Protocol"""
     # To properly map this controller, ensure your config/routing.py
     # file has a resource setup:
     #     map.resource('notification', 'notifications', controller='_admin/notifications',
     #         path_prefix='/_admin', name_prefix='_admin_')
     @LoginRequired()
     @NotAnonymous()
     def __before__(self):
         super(NotificationsController, self).__before__()
     def index(self, format='html'):
         c.user = request.authuser
         notif = NotificationModel().query_for_user(request.authuser.user_id,
                                             filter_=request.GET.getall('type'))
         p = safe_int(request.GET.get('page'), 1)
         c.notifications = Page(notif, page=p, items_per_page=10)
         c.pull_request_type = Notification.TYPE_PULL_REQUEST
         c.comment_type = [Notification.TYPE_CHANGESET_COMMENT,
                           Notification.TYPE_PULL_REQUEST_COMMENT]
         _current_filter = request.GET.getall('type')
         c.current_filter = 'all'
         if _current_filter == [c.pull_request_type]:
             c.current_filter = 'pull_request'
         elif _current_filter == c.comment_type:
             c.current_filter = 'comment'
         return render('admin/notifications/notifications.html')
     def mark_all_read(self):
         if request.environ.get('HTTP_X_PARTIAL_XHR'):
             nm = NotificationModel()
             # mark all read
             nm.mark_all_read_for_user(request.authuser.user_id,
                                       filter_=request.GET.getall('type'))
             Session().commit()
             c.user = request.authuser
             notif = nm.query_for_user(request.authuser.user_id,
                                       filter_=request.GET.getall('type'))
             c.notifications = Page(notif, page=1, items_per_page=10)
             return render('admin/notifications/notifications_data.html')
     def update(self, notification_id):
         try:
             no = Notification.get(notification_id)
             owner = all(un.user_id == request.authuser.user_id
                         for un in no.notifications_to_users)
             if h.HasPermissionAny('hg.admin')() or owner:
                 # deletes only notification2user
                 NotificationModel().mark_read(request.authuser.user_id, no)
                 Session().commit()
                 return 'ok'
         except Exception:
             Session().rollback()
             log.error(traceback.format_exc())
         raise HTTPBadRequest()
     def delete(self, notification_id):
         try:
             no = Notification.get(notification_id)
             owner = any(un.user_id == request.authuser.user_id
                         for un in no.notifications_to_users)
             if h.HasPermissionAny('hg.admin')() or owner:
                 # deletes only notification2user
                 NotificationModel().delete(request.authuser.user_id, no)
                 Session().commit()
                 return 'ok'
         except Exception:
             Session().rollback()
             log.error(traceback.format_exc())
         raise HTTPBadRequest()
     def show(self, notification_id, format='html'):
         notification = Notification.get_or_404(notification_id)
         unotification = NotificationModel() \
             .get_user_notification(request.authuser.user_id, notification)
         # if this association to user is not valid, we don't want to show
         # this message
         if unotification is None:
             raise HTTPForbidden()
         if not unotification.read:
             unotification.mark_as_read()
             Session().commit()
         c.notification = notification
         c.user = request.authuser
         return render('admin/notifications/show_notification.html')

kallithea/controllers/admin/permissions.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/>.
 """
 kallithea.controllers.admin.permissions
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 permissions controller for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 27, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 import formencode
 from formencode import htmlfill
 from pylons import request, tmpl_context as c
 from pylons.i18n.translation import _
 from tg import request, tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPFound
 from kallithea.config.routing import url
 from kallithea.lib import helpers as h
 from kallithea.lib.auth import LoginRequired, HasPermissionAnyDecorator
 from kallithea.lib.base import BaseController, render
 from kallithea.model.forms import DefaultPermissionsForm
 from kallithea.model.permission import PermissionModel
 from kallithea.model.db import User, UserIpMap
 from kallithea.model.meta import Session
 log = logging.getLogger(__name__)
 class PermissionsController(BaseController):
     """REST Controller styled on the Atom Publishing Protocol"""
     # To properly map this controller, ensure your config/routing.py
     # file has a resource setup:
     #     map.resource('permission', 'permissions')
     @LoginRequired()
     @HasPermissionAnyDecorator('hg.admin')
     def __before__(self):
         super(PermissionsController, self).__before__()
     def __load_data(self):
         c.repo_perms_choices = [('repository.none', _('None'),),
                                    ('repository.read', _('Read'),),
                                    ('repository.write', _('Write'),),
                                    ('repository.admin', _('Admin'),)]
         c.group_perms_choices = [('group.none', _('None'),),
                                  ('group.read', _('Read'),),
                                  ('group.write', _('Write'),),
                                  ('group.admin', _('Admin'),)]
         c.user_group_perms_choices = [('usergroup.none', _('None'),),
                                       ('usergroup.read', _('Read'),),
                                       ('usergroup.write', _('Write'),),
                                       ('usergroup.admin', _('Admin'),)]
         c.register_choices = [
             ('hg.register.none',
                 _('Disabled')),
             ('hg.register.manual_activate',
                 _('Allowed with manual account activation')),
             ('hg.register.auto_activate',
                 _('Allowed with automatic account activation')), ]
         c.extern_activate_choices = [
             ('hg.extern_activate.manual', _('Manual activation of external account')),
             ('hg.extern_activate.auto', _('Automatic activation of external account')),
+        ]
         c.repo_create_choices = [('hg.create.none', _('Disabled')),
                                  ('hg.create.repository', _('Enabled'))]
         c.repo_create_on_write_choices = [
             ('hg.create.write_on_repogroup.true', _('Enabled')),
             ('hg.create.write_on_repogroup.false', _('Disabled')),
+        ]
         c.user_group_create_choices = [('hg.usergroup.create.false', _('Disabled')),
                                        ('hg.usergroup.create.true', _('Enabled'))]
         c.repo_group_create_choices = [('hg.repogroup.create.false', _('Disabled')),
                                        ('hg.repogroup.create.true', _('Enabled'))]
         c.fork_choices = [('hg.fork.none', _('Disabled')),
                           ('hg.fork.repository', _('Enabled'))]
     def permission_globals(self):
         c.active = 'globals'
         self.__load_data()
         if request.POST:
             _form = DefaultPermissionsForm(
                 [x[0] for x in c.repo_perms_choices],
                 [x[0] for x in c.group_perms_choices],
                 [x[0] for x in c.user_group_perms_choices],
                 [x[0] for x in c.repo_create_choices],
                 [x[0] for x in c.repo_create_on_write_choices],
                 [x[0] for x in c.repo_group_create_choices],
                 [x[0] for x in c.user_group_create_choices],
                 [x[0] for x in c.fork_choices],
                 [x[0] for x in c.register_choices],
                 [x[0] for x in c.extern_activate_choices])()
             try:
                 form_result = _form.to_python(dict(request.POST))
                 form_result.update({'perm_user_name': 'default'})
                 PermissionModel().update(form_result)
                 Session().commit()
                 h.flash(_('Global permissions updated successfully'),
                         category='success')
             except formencode.Invalid as errors:
                 defaults = errors.value
                 return htmlfill.render(
                     render('admin/permissions/permissions.html'),
                     defaults=defaults,
                     errors=errors.error_dict or {},
                     prefix_error=False,
                     encoding="UTF-8",
                     force_defaults=False)
             except Exception:
                 log.error(traceback.format_exc())
                 h.flash(_('Error occurred during update of permissions'),
                         category='error')
             raise HTTPFound(location=url('admin_permissions'))
         c.user = User.get_default_user()
         defaults = {'anonymous': c.user.active}
         for p in c.user.user_perms:
             if p.permission.permission_name.startswith('repository.'):
                 defaults['default_repo_perm'] = p.permission.permission_name
             if p.permission.permission_name.startswith('group.'):
                 defaults['default_group_perm'] = p.permission.permission_name
             if p.permission.permission_name.startswith('usergroup.'):
                 defaults['default_user_group_perm'] = p.permission.permission_name
             if p.permission.permission_name.startswith('hg.create.write_on_repogroup'):
                 defaults['create_on_write'] = p.permission.permission_name
             elif p.permission.permission_name.startswith('hg.create.'):
                 defaults['default_repo_create'] = p.permission.permission_name
             if p.permission.permission_name.startswith('hg.repogroup.'):
                 defaults['default_repo_group_create'] = p.permission.permission_name
             if p.permission.permission_name.startswith('hg.usergroup.'):
                 defaults['default_user_group_create'] = p.permission.permission_name
             if p.permission.permission_name.startswith('hg.register.'):
                 defaults['default_register'] = p.permission.permission_name
             if p.permission.permission_name.startswith('hg.extern_activate.'):
                 defaults['default_extern_activate'] = p.permission.permission_name
             if p.permission.permission_name.startswith('hg.fork.'):
                 defaults['default_fork'] = p.permission.permission_name
         return htmlfill.render(
             render('admin/permissions/permissions.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     def permission_ips(self):
         c.active = 'ips'
         c.user = User.get_default_user()
         c.user_ip_map = UserIpMap.query() \
                         .filter(UserIpMap.user == c.user).all()
         return render('admin/permissions/permissions.html')
     def permission_perms(self):
         c.active = 'perms'
         c.user = User.get_default_user()
         c.perm_user = c.user.AuthUser
         return render('admin/permissions/permissions.html')

kallithea/controllers/admin/repo_groups.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/>.
 """
 kallithea.controllers.admin.repo_groups
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Repository groups controller for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Mar 23, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 import formencode
 import itertools
 from formencode import htmlfill
 from pylons import request, tmpl_context as c
 from pylons.i18n.translation import _, ungettext
 from tg import request, tmpl_context as c
 from tg.i18n import ugettext as _, ungettext
 from webob.exc import HTTPFound, HTTPForbidden, HTTPNotFound, HTTPInternalServerError
 import kallithea
 from kallithea.config.routing import url
 from kallithea.lib import helpers as h
 from kallithea.lib.auth import LoginRequired, \
     HasRepoGroupPermissionLevelDecorator, HasRepoGroupPermissionLevel, \
     HasPermissionAny
 from kallithea.lib.base import BaseController, render
 from kallithea.model.db import RepoGroup, Repository
 from kallithea.model.scm import RepoGroupList, AvailableRepoGroupChoices
 from kallithea.model.repo_group import RepoGroupModel
 from kallithea.model.forms import RepoGroupForm, RepoGroupPermsForm
 from kallithea.model.meta import Session
 from kallithea.model.repo import RepoModel
 from kallithea.lib.utils2 import safe_int
 from sqlalchemy.sql.expression import func
 log = logging.getLogger(__name__)
 class RepoGroupsController(BaseController):
     @LoginRequired()
     def __before__(self):
         super(RepoGroupsController, self).__before__()
     def __load_defaults(self, extras=(), exclude=()):
         """extras is used for keeping current parent ignoring permissions
         exclude is used for not moving group to itself TODO: also exclude descendants
         Note: only admin can create top level groups
         """
         repo_groups = AvailableRepoGroupChoices([], 'admin', extras)
         exclude_group_ids = set(rg.group_id for rg in exclude)
         c.repo_groups = [rg for rg in repo_groups
                          if rg[0] not in exclude_group_ids]
         repo_model = RepoModel()
         c.users_array = repo_model.get_users_js()
         c.user_groups_array = repo_model.get_user_groups_js()
     def __load_data(self, group_id):
         """
         Load defaults settings for edit, and update
         :param group_id:
         """
         repo_group = RepoGroup.get_or_404(group_id)
         data = repo_group.get_dict()
         data['group_name'] = repo_group.name
         # fill repository group users
         for p in repo_group.repo_group_to_perm:
             data.update({'u_perm_%s' % p.user.username:
                              p.permission.permission_name})
         # fill repository group groups
         for p in repo_group.users_group_to_perm:
             data.update({'g_perm_%s' % p.users_group.users_group_name:
                              p.permission.permission_name})
         return data
     def _revoke_perms_on_yourself(self, form_result):
         _up = filter(lambda u: request.authuser.username == u[0],
                      form_result['perms_updates'])
         _new = filter(lambda u: request.authuser.username == u[0],
                       form_result['perms_new'])
         if _new and _new[0][1] != 'group.admin' or _up and _up[0][1] != 'group.admin':
             return True
         return False
     def index(self, format='html'):
         _list = RepoGroup.query(sorted=True).all()
         group_iter = RepoGroupList(_list, perm_level='admin')
         repo_groups_data = []
         total_records = len(group_iter)
         _tmpl_lookup = kallithea.CONFIG['pylons.app_globals'].mako_lookup
         template = _tmpl_lookup.get_template('data_table/_dt_elements.html')
         repo_group_name = lambda repo_group_name, children_groups: (
             template.get_def("repo_group_name")
             .render(repo_group_name, children_groups, _=_, h=h, c=c)
+        )
         repo_group_actions = lambda repo_group_id, repo_group_name, gr_count: (
             template.get_def("repo_group_actions")
             .render(repo_group_id, repo_group_name, gr_count, _=_, h=h, c=c,
                     ungettext=ungettext)
+        )
         for repo_gr in group_iter:
             children_groups = map(h.safe_unicode,
                 itertools.chain((g.name for g in repo_gr.parents),
                                 (x.name for x in [repo_gr])))
             repo_count = repo_gr.repositories.count()
             repo_groups_data.append({
                 "raw_name": repo_gr.group_name,
                 "group_name": repo_group_name(repo_gr.group_name, children_groups),
                 "desc": h.escape(repo_gr.group_description),
                 "repos": repo_count,
                 "owner": h.person(repo_gr.owner),
                 "action": repo_group_actions(repo_gr.group_id, repo_gr.group_name,
                                              repo_count)
             })
         c.data = {
             "totalRecords": total_records,
             "startIndex": 0,
             "sort": None,
             "dir": "asc",
             "records": repo_groups_data
+        }
         return render('admin/repo_groups/repo_groups.html')
     def create(self):
         self.__load_defaults()
         # permissions for can create group based on parent_id are checked
         # here in the Form
         repo_group_form = RepoGroupForm(repo_groups=c.repo_groups)
         try:
             form_result = repo_group_form.to_python(dict(request.POST))
             gr = RepoGroupModel().create(
                 group_name=form_result['group_name'],
                 group_description=form_result['group_description'],
                 parent=form_result['parent_group_id'],
                 owner=request.authuser.user_id, # TODO: make editable
                 copy_permissions=form_result['group_copy_permissions']
+            )
             Session().commit()
             #TODO: in future action_logger(, '', '', '')
         except formencode.Invalid as errors:
             return htmlfill.render(
                 render('admin/repo_groups/repo_group_add.html'),
                 defaults=errors.value,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False)
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('Error occurred during creation of repository group %s') \
                     % request.POST.get('group_name'), category='error')
             parent_group_id = form_result['parent_group_id']
             #TODO: maybe we should get back to the main view, not the admin one
             raise HTTPFound(location=url('repos_groups', parent_group=parent_group_id))
         h.flash(_('Created repository group %s') % gr.group_name,
                 category='success')
         raise HTTPFound(location=url('repos_group_home', group_name=gr.group_name))
     def new(self):
         if HasPermissionAny('hg.admin')('group create'):
             #we're global admin, we're ok and we can create TOP level groups
             pass
         else:
             # we pass in parent group into creation form, thus we know
             # what would be the group, we can check perms here !
             group_id = safe_int(request.GET.get('parent_group'))
             group = RepoGroup.get(group_id) if group_id else None
             group_name = group.group_name if group else None
             if HasRepoGroupPermissionLevel('admin')(group_name, 'group create'):
                 pass
             else:
                 raise HTTPForbidden()
         self.__load_defaults()
         return render('admin/repo_groups/repo_group_add.html')
     @HasRepoGroupPermissionLevelDecorator('admin')
     def update(self, group_name):
         c.repo_group = RepoGroup.guess_instance(group_name)
         self.__load_defaults(extras=[c.repo_group.parent_group],
                              exclude=[c.repo_group])
         # TODO: kill allow_empty_group - it is only used for redundant form validation!
         if HasPermissionAny('hg.admin')('group edit'):
             #we're global admin, we're ok and we can create TOP level groups
             allow_empty_group = True
         elif not c.repo_group.parent_group:
             allow_empty_group = True
         else:
             allow_empty_group = False
         repo_group_form = RepoGroupForm(
             edit=True,
             old_data=c.repo_group.get_dict(),
             repo_groups=c.repo_groups,
             can_create_in_root=allow_empty_group,
         )()
         try:
             form_result = repo_group_form.to_python(dict(request.POST))
             new_gr = RepoGroupModel().update(group_name, form_result)
             Session().commit()
             h.flash(_('Updated repository group %s') \
                     % form_result['group_name'], category='success')
             # we now have new name !
             group_name = new_gr.group_name
             #TODO: in future action_logger(, '', '', '')
         except formencode.Invalid as errors:
             c.active = 'settings'
             return htmlfill.render(
                 render('admin/repo_groups/repo_group_edit.html'),
                 defaults=errors.value,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False)
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('Error occurred during update of repository group %s') \
                     % request.POST.get('group_name'), category='error')
         raise HTTPFound(location=url('edit_repo_group', group_name=group_name))
     @HasRepoGroupPermissionLevelDecorator('admin')
     def delete(self, group_name):
         gr = c.repo_group = RepoGroup.guess_instance(group_name)
         repos = gr.repositories.all()
         if repos:
             h.flash(_('This group contains %s repositories and cannot be '
                       'deleted') % len(repos), category='warning')
             raise HTTPFound(location=url('repos_groups'))
         children = gr.children.all()
         if children:
             h.flash(_('This group contains %s subgroups and cannot be deleted'
                       % (len(children))), category='warning')
             raise HTTPFound(location=url('repos_groups'))
         try:
             RepoGroupModel().delete(group_name)
             Session().commit()
             h.flash(_('Removed repository group %s') % group_name,
                     category='success')
             #TODO: in future action_logger(, '', '', '')
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('Error occurred during deletion of repository group %s')
                     % group_name, category='error')
         if gr.parent_group:
             raise HTTPFound(location=url('repos_group_home', group_name=gr.parent_group.group_name))
         raise HTTPFound(location=url('repos_groups'))
     def show_by_name(self, group_name):
         """
         This is a proxy that does a lookup group_name -> id, and shows
         the group by id view instead
         """
         group_name = group_name.rstrip('/')
         id_ = RepoGroup.get_by_group_name(group_name)
         if id_:
             return self.show(group_name)
         raise HTTPNotFound
     @HasRepoGroupPermissionLevelDecorator('read')
     def show(self, group_name):
         c.active = 'settings'
         c.group = c.repo_group = RepoGroup.guess_instance(group_name)
         groups = RepoGroup.query(sorted=True).filter_by(parent_group=c.group).all()
         c.groups = self.scm_model.get_repo_groups(groups)
         repos_list = Repository.query(sorted=True).filter_by(group=c.group).all()
         repos_data = RepoModel().get_repos_as_dict(repos_list=repos_list,
                                                    admin=False, short_name=True)
         # data used to render the grid
         c.data = repos_data
         return render('admin/repo_groups/repo_group_show.html')
     @HasRepoGroupPermissionLevelDecorator('admin')
     def edit(self, group_name):
         c.active = 'settings'
         c.repo_group = RepoGroup.guess_instance(group_name)
         self.__load_defaults(extras=[c.repo_group.parent_group],
                              exclude=[c.repo_group])
         defaults = self.__load_data(c.repo_group.group_id)
         return htmlfill.render(
             render('admin/repo_groups/repo_group_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False
+        )
     @HasRepoGroupPermissionLevelDecorator('admin')
     def edit_repo_group_advanced(self, group_name):
         c.active = 'advanced'
         c.repo_group = RepoGroup.guess_instance(group_name)
         return render('admin/repo_groups/repo_group_edit.html')
     @HasRepoGroupPermissionLevelDecorator('admin')
     def edit_repo_group_perms(self, group_name):
         c.active = 'perms'
         c.repo_group = RepoGroup.guess_instance(group_name)
         self.__load_defaults()
         defaults = self.__load_data(c.repo_group.group_id)
         return htmlfill.render(
             render('admin/repo_groups/repo_group_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False
+        )
     @HasRepoGroupPermissionLevelDecorator('admin')
     def update_perms(self, group_name):
         """
         Update permissions for given repository group
         :param group_name:
         """
         c.repo_group = RepoGroup.guess_instance(group_name)
         valid_recursive_choices = ['none', 'repos', 'groups', 'all']
         form_result = RepoGroupPermsForm(valid_recursive_choices)().to_python(request.POST)
         if not request.authuser.is_admin:
             if self._revoke_perms_on_yourself(form_result):
                 msg = _('Cannot revoke permission for yourself as admin')
                 h.flash(msg, category='warning')
                 raise HTTPFound(location=url('edit_repo_group_perms', group_name=group_name))
         recursive = form_result['recursive']
         # iterate over all members(if in recursive mode) of this groups and
         # set the permissions !
         # this can be potentially heavy operation
         RepoGroupModel()._update_permissions(c.repo_group,
                                              form_result['perms_new'],
                                              form_result['perms_updates'],
                                              recursive)
         #TODO: implement this
         #action_logger(request.authuser, 'admin_changed_repo_permissions',
         #              repo_name, request.ip_addr)
         Session().commit()
         h.flash(_('Repository group permissions updated'), category='success')
         raise HTTPFound(location=url('edit_repo_group_perms', group_name=group_name))
     @HasRepoGroupPermissionLevelDecorator('admin')
     def delete_perms(self, group_name):
         try:
             obj_type = request.POST.get('obj_type')
             obj_id = None
             if obj_type == 'user':
                 obj_id = safe_int(request.POST.get('user_id'))
             elif obj_type == 'user_group':
                 obj_id = safe_int(request.POST.get('user_group_id'))
             if not request.authuser.is_admin:
                 if obj_type == 'user' and request.authuser.user_id == obj_id:
                     msg = _('Cannot revoke permission for yourself as admin')
                     h.flash(msg, category='warning')
                     raise Exception('revoke admin permission on self')
             recursive = request.POST.get('recursive', 'none')
             if obj_type == 'user':
                 RepoGroupModel().delete_permission(repo_group=group_name,
                                                    obj=obj_id, obj_type='user',
                                                    recursive=recursive)
             elif obj_type == 'user_group':
                 RepoGroupModel().delete_permission(repo_group=group_name,
                                                    obj=obj_id,
                                                    obj_type='user_group',
                                                    recursive=recursive)
             Session().commit()
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('An error occurred during revoking of permission'),
                     category='error')
             raise HTTPInternalServerError()

kallithea/controllers/admin/repos.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/>.
 """
 kallithea.controllers.admin.repos
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Repositories controller for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 7, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 import formencode
 from formencode import htmlfill
 from pylons import request, tmpl_context as c
 from pylons.i18n.translation import _
 from tg import request, tmpl_context as c
 from tg.i18n import ugettext as _
 from sqlalchemy.sql.expression import func
 from webob.exc import HTTPFound, HTTPInternalServerError, HTTPForbidden, HTTPNotFound
 from kallithea.config.routing import url
 from kallithea.lib import helpers as h
 from kallithea.lib.auth import LoginRequired, \
     HasRepoPermissionLevelDecorator, NotAnonymous, HasPermissionAny
 from kallithea.lib.base import BaseRepoController, render, jsonify
 from kallithea.lib.utils import action_logger
 from kallithea.lib.vcs import RepositoryError
 from kallithea.model.meta import Session
 from kallithea.model.db import User, Repository, UserFollowing, RepoGroup, \
     Setting, RepositoryField
 from kallithea.model.forms import RepoForm, RepoFieldForm, RepoPermsForm
 from kallithea.model.scm import ScmModel, AvailableRepoGroupChoices, RepoList
 from kallithea.model.repo import RepoModel
 from kallithea.lib.exceptions import AttachedForksError
 from kallithea.lib.utils2 import safe_int
 log = logging.getLogger(__name__)
 class ReposController(BaseRepoController):
     """
     REST Controller styled on the Atom Publishing Protocol"""
     # To properly map this controller, ensure your config/routing.py
     # file has a resource setup:
     #     map.resource('repo', 'repos')
     @LoginRequired()
     def __before__(self):
         super(ReposController, self).__before__()
     def _load_repo(self):
         repo_obj = c.db_repo
         if repo_obj is None:
             h.not_mapped_error(c.repo_name)
             raise HTTPFound(location=url('repos'))
         return repo_obj
     def __load_defaults(self, repo=None):
         top_perms = ['hg.create.repository']
         if HasPermissionAny('hg.create.write_on_repogroup.true')():
             repo_group_perm_level = 'write'
         else:
             repo_group_perm_level = 'admin'
         extras = [] if repo is None else [repo.group]
         c.repo_groups = AvailableRepoGroupChoices(top_perms, repo_group_perm_level, extras)
         c.landing_revs_choices, c.landing_revs = ScmModel().get_repo_landing_revs(repo)
     def __load_data(self):
         """
         Load defaults settings for edit, and update
         """
         c.repo_info = self._load_repo()
         self.__load_defaults(c.repo_info)
         defaults = RepoModel()._get_defaults(c.repo_name)
         defaults['clone_uri'] = c.repo_info.clone_uri_hidden # don't show password
         return defaults
     def index(self, format='html'):
         _list = Repository.query(sorted=True).all()
         c.repos_list = RepoList(_list, perm_level='admin')
         repos_data = RepoModel().get_repos_as_dict(repos_list=c.repos_list,
                                                    admin=True,
                                                    super_user_actions=True)
         #data used to render the grid
         c.data = repos_data
         return render('admin/repos/repos.html')
     @NotAnonymous()
     def create(self):
         self.__load_defaults()
         form_result = {}
         try:
             # CanWriteGroup validators checks permissions of this POST
             form_result = RepoForm(repo_groups=c.repo_groups,
                                    landing_revs=c.landing_revs_choices)() \
                             .to_python(dict(request.POST))
             # create is done sometimes async on celery, db transaction
             # management is handled there.
             task = RepoModel().create(form_result, request.authuser.user_id)
             task_id = task.task_id
         except formencode.Invalid as errors:
             log.info(errors)
             return htmlfill.render(
                 render('admin/repos/repo_add.html'),
                 defaults=errors.value,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 force_defaults=False,
                 encoding="UTF-8")
         except Exception:
             log.error(traceback.format_exc())
             msg = (_('Error creating repository %s')
                    % form_result.get('repo_name'))
             h.flash(msg, category='error')
             raise HTTPFound(location=url('home'))
         raise HTTPFound(location=h.url('repo_creating_home',
                               repo_name=form_result['repo_name_full'],
                               task_id=task_id))
     @NotAnonymous()
     def create_repository(self):
         self.__load_defaults()
         if not c.repo_groups:
             raise HTTPForbidden
         parent_group = request.GET.get('parent_group')
         ## apply the defaults from defaults page
         defaults = Setting.get_default_repo_settings(strip_prefix=True)
         if parent_group:
             prg = RepoGroup.get(parent_group)
             if prg is None or not any(rgc[0] == prg.group_id
                                       for rgc in c.repo_groups):
                 raise HTTPForbidden
             defaults.update({'repo_group': parent_group})
         return htmlfill.render(
             render('admin/repos/repo_add.html'),
             defaults=defaults,
             errors={},
             prefix_error=False,
             encoding="UTF-8",
             force_defaults=False)
     @LoginRequired()
     @NotAnonymous()
     def repo_creating(self, repo_name):
         c.repo = repo_name
         c.task_id = request.GET.get('task_id')
         if not c.repo:
             raise HTTPNotFound()
         return render('admin/repos/repo_creating.html')
     @LoginRequired()
     @NotAnonymous()
     @jsonify
     def repo_check(self, repo_name):
         c.repo = repo_name
         task_id = request.GET.get('task_id')
         if task_id and task_id not in ['None']:
             from kallithea import CELERY_ON
             from kallithea.lib import celerypylons
             if CELERY_ON:
                 task = celerypylons.result.AsyncResult(task_id)
                 if task.failed():
                     raise HTTPInternalServerError(task.traceback)
         repo = Repository.get_by_repo_name(repo_name)
         if repo and repo.repo_state == Repository.STATE_CREATED:
             if repo.clone_uri:
                 h.flash(_('Created repository %s from %s')
                         % (repo.repo_name, repo.clone_uri_hidden), category='success')
             else:
                 repo_url = h.link_to(repo.repo_name,
                                      h.url('summary_home',
                                            repo_name=repo.repo_name))
                 fork = repo.fork
                 if fork is not None:
                     fork_name = fork.repo_name
                     h.flash(h.literal(_('Forked repository %s as %s')
                             % (fork_name, repo_url)), category='success')
                 else:
                     h.flash(h.literal(_('Created repository %s') % repo_url),
                             category='success')
             return {'result': True}
         return {'result': False}
     @HasRepoPermissionLevelDecorator('admin')
     def update(self, repo_name):
         c.repo_info = self._load_repo()
         self.__load_defaults(c.repo_info)
         c.active = 'settings'
         c.repo_fields = RepositoryField.query() \
             .filter(RepositoryField.repository == c.repo_info).all()
         repo_model = RepoModel()
         changed_name = repo_name
         repo = Repository.get_by_repo_name(repo_name)
         old_data = {
             'repo_name': repo_name,
             'repo_group': repo.group.get_dict() if repo.group else {},
             'repo_type': repo.repo_type,
+        }
         _form = RepoForm(edit=True, old_data=old_data,
                          repo_groups=c.repo_groups,
                          landing_revs=c.landing_revs_choices)()
         try:
             form_result = _form.to_python(dict(request.POST))
             repo = repo_model.update(repo_name, **form_result)
             ScmModel().mark_for_invalidation(repo_name)
             h.flash(_('Repository %s updated successfully') % repo_name,
                     category='success')
             changed_name = repo.repo_name
             action_logger(request.authuser, 'admin_updated_repo',
                 changed_name, request.ip_addr)
             Session().commit()
         except formencode.Invalid as errors:
             log.info(errors)
             defaults = self.__load_data()
             defaults.update(errors.value)
             c.users_array = repo_model.get_users_js()
             return htmlfill.render(
                 render('admin/repos/repo_edit.html'),
                 defaults=defaults,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False)
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('Error occurred during update of repository %s') \
                     % repo_name, category='error')
         raise HTTPFound(location=url('edit_repo', repo_name=changed_name))
     @HasRepoPermissionLevelDecorator('admin')
     def delete(self, repo_name):
         repo_model = RepoModel()
         repo = repo_model.get_by_repo_name(repo_name)
         if not repo:
             h.not_mapped_error(repo_name)
             raise HTTPFound(location=url('repos'))
         try:
             _forks = repo.forks.count()
             handle_forks = None
             if _forks and request.POST.get('forks'):
                 do = request.POST['forks']
                 if do == 'detach_forks':
                     handle_forks = 'detach'
                     h.flash(_('Detached %s forks') % _forks, category='success')
                 elif do == 'delete_forks':
                     handle_forks = 'delete'
                     h.flash(_('Deleted %s forks') % _forks, category='success')
             repo_model.delete(repo, forks=handle_forks)
             action_logger(request.authuser, 'admin_deleted_repo',
                 repo_name, request.ip_addr)
             ScmModel().mark_for_invalidation(repo_name)
             h.flash(_('Deleted repository %s') % repo_name, category='success')
             Session().commit()
         except AttachedForksError:
             h.flash(_('Cannot delete repository %s which still has forks')
                         % repo_name, category='warning')
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('An error occurred during deletion of %s') % repo_name,
                     category='error')
         if repo.group:
             raise HTTPFound(location=url('repos_group_home', group_name=repo.group.group_name))
         raise HTTPFound(location=url('repos'))
     @HasRepoPermissionLevelDecorator('admin')
     def edit(self, repo_name):
         defaults = self.__load_data()
         c.repo_fields = RepositoryField.query() \
             .filter(RepositoryField.repository == c.repo_info).all()
         repo_model = RepoModel()
         c.users_array = repo_model.get_users_js()
         c.active = 'settings'
         return htmlfill.render(
             render('admin/repos/repo_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @HasRepoPermissionLevelDecorator('admin')
     def edit_permissions(self, repo_name):
         c.repo_info = self._load_repo()
         repo_model = RepoModel()
         c.users_array = repo_model.get_users_js()
         c.user_groups_array = repo_model.get_user_groups_js()
         c.active = 'permissions'
         defaults = RepoModel()._get_defaults(repo_name)
         return htmlfill.render(
             render('admin/repos/repo_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     def edit_permissions_update(self, repo_name):
         form = RepoPermsForm()().to_python(request.POST)
         RepoModel()._update_permissions(repo_name, form['perms_new'],
                                         form['perms_updates'])
         #TODO: implement this
         #action_logger(request.authuser, 'admin_changed_repo_permissions',
         #              repo_name, request.ip_addr)
         Session().commit()
         h.flash(_('Repository permissions updated'), category='success')
         raise HTTPFound(location=url('edit_repo_perms', repo_name=repo_name))
     def edit_permissions_revoke(self, repo_name):
         try:
             obj_type = request.POST.get('obj_type')
             obj_id = None
             if obj_type == 'user':
                 obj_id = safe_int(request.POST.get('user_id'))
             elif obj_type == 'user_group':
                 obj_id = safe_int(request.POST.get('user_group_id'))
             if obj_type == 'user':
                 RepoModel().revoke_user_permission(repo=repo_name, user=obj_id)
             elif obj_type == 'user_group':
                 RepoModel().revoke_user_group_permission(
                     repo=repo_name, group_name=obj_id
+                )
             #TODO: implement this
             #action_logger(request.authuser, 'admin_revoked_repo_permissions',
             #              repo_name, request.ip_addr)
             Session().commit()
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('An error occurred during revoking of permission'),
                     category='error')
             raise HTTPInternalServerError()
     @HasRepoPermissionLevelDecorator('admin')
     def edit_fields(self, repo_name):
         c.repo_info = self._load_repo()
         c.repo_fields = RepositoryField.query() \
             .filter(RepositoryField.repository == c.repo_info).all()
         c.active = 'fields'
         if request.POST:
             raise HTTPFound(location=url('repo_edit_fields'))
         return render('admin/repos/repo_edit.html')
     @HasRepoPermissionLevelDecorator('admin')
     def create_repo_field(self, repo_name):
         try:
             form_result = RepoFieldForm()().to_python(dict(request.POST))
             new_field = RepositoryField()
             new_field.repository = Repository.get_by_repo_name(repo_name)
             new_field.field_key = form_result['new_field_key']
             new_field.field_type = form_result['new_field_type']  # python type
             new_field.field_value = form_result['new_field_value']  # set initial blank value
             new_field.field_desc = form_result['new_field_desc']
             new_field.field_label = form_result['new_field_label']
             Session().add(new_field)
             Session().commit()
         except Exception as e:
             log.error(traceback.format_exc())
             msg = _('An error occurred during creation of field')
             if isinstance(e, formencode.Invalid):
                 msg += ". " + e.msg
             h.flash(msg, category='error')
         raise HTTPFound(location=url('edit_repo_fields', repo_name=repo_name))
     @HasRepoPermissionLevelDecorator('admin')
     def delete_repo_field(self, repo_name, field_id):
         field = RepositoryField.get_or_404(field_id)
         try:
             Session().delete(field)
             Session().commit()
         except Exception as e:
             log.error(traceback.format_exc())
             msg = _('An error occurred during removal of field')
             h.flash(msg, category='error')
         raise HTTPFound(location=url('edit_repo_fields', repo_name=repo_name))
     @HasRepoPermissionLevelDecorator('admin')
     def edit_advanced(self, repo_name):
         c.repo_info = self._load_repo()
         c.default_user_id = User.get_default_user().user_id
         c.in_public_journal = UserFollowing.query() \
             .filter(UserFollowing.user_id == c.default_user_id) \
             .filter(UserFollowing.follows_repository == c.repo_info).scalar()
         _repos = Repository.query(sorted=True).all()
         read_access_repos = RepoList(_repos, perm_level='read')
         c.repos_list = [(None, _('-- Not a fork --'))]
         c.repos_list += [(x.repo_id, x.repo_name)
                          for x in read_access_repos
                          if x.repo_id != c.repo_info.repo_id]
         defaults = {
             'id_fork_of': c.repo_info.fork_id if c.repo_info.fork_id else ''
+        }
         c.active = 'advanced'
         if request.POST:
             raise HTTPFound(location=url('repo_edit_advanced'))
         return htmlfill.render(
             render('admin/repos/repo_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @HasRepoPermissionLevelDecorator('admin')
     def edit_advanced_journal(self, repo_name):
         """
         Sets this repository to be visible in public journal,
         in other words asking default user to follow this repo
         :param repo_name:
         """
         try:
             repo_id = Repository.get_by_repo_name(repo_name).repo_id
             user_id = User.get_default_user().user_id
             self.scm_model.toggle_following_repo(repo_id, user_id)
             h.flash(_('Updated repository visibility in public journal'),
                     category='success')
             Session().commit()
         except Exception:
             h.flash(_('An error occurred during setting this'
                       ' repository in public journal'),
                     category='error')
         raise HTTPFound(location=url('edit_repo_advanced', repo_name=repo_name))
     @HasRepoPermissionLevelDecorator('admin')
     def edit_advanced_fork(self, repo_name):
         """
         Mark given repository as a fork of another
         :param repo_name:
         """
         try:
             fork_id = request.POST.get('id_fork_of')
             repo = ScmModel().mark_as_fork(repo_name, fork_id,
                                            request.authuser.username)
             fork = repo.fork.repo_name if repo.fork else _('Nothing')
             Session().commit()
             h.flash(_('Marked repository %s as fork of %s') % (repo_name, fork),
                     category='success')
         except RepositoryError as e:
             log.error(traceback.format_exc())
             h.flash(str(e), category='error')
         except Exception as e:
             log.error(traceback.format_exc())
             h.flash(_('An error occurred during this operation'),
                     category='error')
         raise HTTPFound(location=url('edit_repo_advanced', repo_name=repo_name))
     @HasRepoPermissionLevelDecorator('admin')
     def edit_advanced_locking(self, repo_name):
         """
         Unlock repository when it is locked !
         :param repo_name:
         """
         try:
             repo = Repository.get_by_repo_name(repo_name)
             if request.POST.get('set_lock'):
                 Repository.lock(repo, request.authuser.user_id)
                 h.flash(_('Repository has been locked'), category='success')
             elif request.POST.get('set_unlock'):
                 Repository.unlock(repo)
                 h.flash(_('Repository has been unlocked'), category='success')
         except Exception as e:
             log.error(traceback.format_exc())
             h.flash(_('An error occurred during unlocking'),
                     category='error')
         raise HTTPFound(location=url('edit_repo_advanced', repo_name=repo_name))
     @HasRepoPermissionLevelDecorator('write')
     def toggle_locking(self, repo_name):
         try:
             repo = Repository.get_by_repo_name(repo_name)
             if repo.enable_locking:
                 if repo.locked[0]:
                     Repository.unlock(repo)
                     h.flash(_('Repository has been unlocked'), category='success')
                 else:
                     Repository.lock(repo, request.authuser.user_id)
                     h.flash(_('Repository has been locked'), category='success')
         except Exception as e:
             log.error(traceback.format_exc())
             h.flash(_('An error occurred during unlocking'),
                     category='error')
         raise HTTPFound(location=url('summary_home', repo_name=repo_name))
     @HasRepoPermissionLevelDecorator('admin')
     def edit_caches(self, repo_name):
         c.repo_info = self._load_repo()
         c.active = 'caches'
         if request.POST:
             try:
                 ScmModel().mark_for_invalidation(repo_name)
                 Session().commit()
                 h.flash(_('Cache invalidation successful'),
                         category='success')
             except Exception as e:
                 log.error(traceback.format_exc())
                 h.flash(_('An error occurred during cache invalidation'),
                         category='error')
             raise HTTPFound(location=url('edit_repo_caches', repo_name=c.repo_name))
         return render('admin/repos/repo_edit.html')
     @HasRepoPermissionLevelDecorator('admin')
     def edit_remote(self, repo_name):
         c.repo_info = self._load_repo()
         c.active = 'remote'
         if request.POST:
             try:
                 ScmModel().pull_changes(repo_name, request.authuser.username)
                 h.flash(_('Pulled from remote location'), category='success')
             except Exception as e:
                 log.error(traceback.format_exc())
                 h.flash(_('An error occurred during pull from remote location'),
                         category='error')
             raise HTTPFound(location=url('edit_repo_remote', repo_name=c.repo_name))
         return render('admin/repos/repo_edit.html')
     @HasRepoPermissionLevelDecorator('admin')
     def edit_statistics(self, repo_name):
         c.repo_info = self._load_repo()
         repo = c.repo_info.scm_instance
         if c.repo_info.stats:
             # this is on what revision we ended up so we add +1 for count
             last_rev = c.repo_info.stats.stat_on_revision + 1
         else:
             last_rev = 0
         c.stats_revision = last_rev
         c.repo_last_rev = repo.count() if repo.revisions else 0
         if last_rev == 0 or c.repo_last_rev == 0:
             c.stats_percentage = 0
         else:
             c.stats_percentage = '%.2f' % ((float((last_rev)) / c.repo_last_rev) * 100)
         c.active = 'statistics'
         if request.POST:
             try:
                 RepoModel().delete_stats(repo_name)
                 Session().commit()
             except Exception as e:
                 log.error(traceback.format_exc())
                 h.flash(_('An error occurred during deletion of repository stats'),
                         category='error')
             raise HTTPFound(location=url('edit_repo_statistics', repo_name=c.repo_name))
         return render('admin/repos/repo_edit.html')

kallithea/controllers/admin/settings.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/>.
 """
 kallithea.controllers.admin.settings
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 settings controller for Kallithea admin
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Jul 14, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 import formencode
 from formencode import htmlfill
 from pylons import request, tmpl_context as c, config
 from pylons.i18n.translation import _
 from tg import request, tmpl_context as c, config
 from tg.i18n import ugettext as _
 from webob.exc import HTTPFound
 from kallithea.config.routing import url
 from kallithea.lib import helpers as h
 from kallithea.lib.auth import LoginRequired, HasPermissionAnyDecorator
 from kallithea.lib.base import BaseController, render
 from kallithea.lib.celerylib import tasks
 from kallithea.lib.exceptions import HgsubversionImportError
 from kallithea.lib.utils import repo2db_mapper, set_app_settings
 from kallithea.model.db import Ui, Repository, Setting
 from kallithea.model.forms import ApplicationSettingsForm, \
     ApplicationUiSettingsForm, ApplicationVisualisationForm
 from kallithea.model.scm import ScmModel
 from kallithea.model.notification import EmailNotificationModel
 from kallithea.model.meta import Session
 from kallithea.lib.utils2 import str2bool, safe_unicode
 log = logging.getLogger(__name__)
 class SettingsController(BaseController):
     """REST Controller styled on the Atom Publishing Protocol"""
     # To properly map this controller, ensure your config/routing.py
     # file has a resource setup:
     #     map.resource('setting', 'settings', controller='admin/settings',
     #         path_prefix='/admin', name_prefix='admin_')
     @LoginRequired()
     def __before__(self):
         super(SettingsController, self).__before__()
     def _get_hg_ui_settings(self):
         ret = Ui.query().all()
         settings = {}
         for each in ret:
             k = each.ui_section + '_' + each.ui_key
             v = each.ui_value
             if k == 'paths_/':
                 k = 'paths_root_path'
             k = k.replace('.', '_')
             if each.ui_section in ['hooks', 'extensions']:
                 v = each.ui_active
             settings[k] = v
         return settings
     @HasPermissionAnyDecorator('hg.admin')
     def settings_vcs(self):
         c.active = 'vcs'
         if request.POST:
             application_form = ApplicationUiSettingsForm()()
             try:
                 form_result = application_form.to_python(dict(request.POST))
             except formencode.Invalid as errors:
                 return htmlfill.render(
                      render('admin/settings/settings.html'),
                      defaults=errors.value,
                      errors=errors.error_dict or {},
                      prefix_error=False,
                      encoding="UTF-8",
                      force_defaults=False)
             try:
                 if c.visual.allow_repo_location_change:
                     sett = Ui.get_by_key('paths', '/')
                     sett.ui_value = form_result['paths_root_path']
                 #HOOKS
                 sett = Ui.get_by_key('hooks', Ui.HOOK_UPDATE)
                 sett.ui_active = form_result['hooks_changegroup_update']
                 sett = Ui.get_by_key('hooks', Ui.HOOK_REPO_SIZE)
                 sett.ui_active = form_result['hooks_changegroup_repo_size']
                 sett = Ui.get_by_key('hooks', Ui.HOOK_PUSH)
                 sett.ui_active = form_result['hooks_changegroup_push_logger']
                 sett = Ui.get_by_key('hooks', Ui.HOOK_PULL)
                 sett.ui_active = form_result['hooks_outgoing_pull_logger']
                 ## EXTENSIONS
                 sett = Ui.get_or_create('extensions', 'largefiles')
                 sett.ui_active = form_result['extensions_largefiles']
                 sett = Ui.get_or_create('extensions', 'hgsubversion')
                 sett.ui_active = form_result['extensions_hgsubversion']
                 if sett.ui_active:
                     try:
                         import hgsubversion  # pragma: no cover
                     except ImportError:
                         raise HgsubversionImportError
 #                sett = Ui.get_or_create('extensions', 'hggit')
 #                sett.ui_active = form_result['extensions_hggit']
                 Session().commit()
                 h.flash(_('Updated VCS settings'), category='success')
             except HgsubversionImportError:
                 log.error(traceback.format_exc())
                 h.flash(_('Unable to activate hgsubversion support. '
                           'The "hgsubversion" library is missing'),
                         category='error')
             except Exception:
                 log.error(traceback.format_exc())
                 h.flash(_('Error occurred while updating '
                           'application settings'), category='error')
         defaults = Setting.get_app_settings()
         defaults.update(self._get_hg_ui_settings())
         return htmlfill.render(
             render('admin/settings/settings.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @HasPermissionAnyDecorator('hg.admin')
     def settings_mapping(self):
         c.active = 'mapping'
         if request.POST:
             rm_obsolete = request.POST.get('destroy', False)
             install_git_hooks = request.POST.get('hooks', False)
             overwrite_git_hooks = request.POST.get('hooks_overwrite', False);
             invalidate_cache = request.POST.get('invalidate', False)
             log.debug('rescanning repo location with destroy obsolete=%s, '
                       'install git hooks=%s and '
                       'overwrite git hooks=%s' % (rm_obsolete, install_git_hooks, overwrite_git_hooks))
             filesystem_repos = ScmModel().repo_scan()
             added, removed = repo2db_mapper(filesystem_repos, rm_obsolete,
                                             install_git_hooks=install_git_hooks,
                                             user=request.authuser.username,
                                             overwrite_git_hooks=overwrite_git_hooks)
             h.flash(h.literal(_('Repositories successfully rescanned. Added: %s. Removed: %s.') %
                 (', '.join(h.link_to(safe_unicode(repo_name), h.url('summary_home', repo_name=repo_name))
                  for repo_name in added) or '-',
                  ', '.join(h.escape(safe_unicode(repo_name)) for repo_name in removed) or '-')),
                 category='success')
             if invalidate_cache:
                 log.debug('invalidating all repositories cache')
                 i = 0
                 for repo in Repository.query():
                     try:
                         ScmModel().mark_for_invalidation(repo.repo_name)
                         i += 1
                     except VCSError as e:
                         log.warning('VCS error invalidating %s: %s', repo.repo_name, e)
                 h.flash(_('Invalidated %s repositories') % i, category='success')
             raise HTTPFound(location=url('admin_settings_mapping'))
         defaults = Setting.get_app_settings()
         defaults.update(self._get_hg_ui_settings())
         return htmlfill.render(
             render('admin/settings/settings.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @HasPermissionAnyDecorator('hg.admin')
     def settings_global(self):
         c.active = 'global'
         if request.POST:
             application_form = ApplicationSettingsForm()()
             try:
                 form_result = application_form.to_python(dict(request.POST))
             except formencode.Invalid as errors:
                 return htmlfill.render(
                     render('admin/settings/settings.html'),
                     defaults=errors.value,
                     errors=errors.error_dict or {},
                     prefix_error=False,
                     encoding="UTF-8",
                     force_defaults=False)
             try:
                 for setting in (
                     'title',
                     'realm',
                     'ga_code',
                     'captcha_public_key',
                     'captcha_private_key',
                 ):
                     Setting.create_or_update(setting, form_result[setting])
                 Session().commit()
                 set_app_settings(config)
                 h.flash(_('Updated application settings'), category='success')
             except Exception:
                 log.error(traceback.format_exc())
                 h.flash(_('Error occurred while updating '
                           'application settings'),
                           category='error')
             raise HTTPFound(location=url('admin_settings_global'))
         defaults = Setting.get_app_settings()
         defaults.update(self._get_hg_ui_settings())
         return htmlfill.render(
             render('admin/settings/settings.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @HasPermissionAnyDecorator('hg.admin')
     def settings_visual(self):
         c.active = 'visual'
         if request.POST:
             application_form = ApplicationVisualisationForm()()
             try:
                 form_result = application_form.to_python(dict(request.POST))
             except formencode.Invalid as errors:
                 return htmlfill.render(
                     render('admin/settings/settings.html'),
                     defaults=errors.value,
                     errors=errors.error_dict or {},
                     prefix_error=False,
                     encoding="UTF-8",
                     force_defaults=False)
             try:
                 settings = [
                     ('show_public_icon', 'show_public_icon', 'bool'),
                     ('show_private_icon', 'show_private_icon', 'bool'),
                     ('stylify_metatags', 'stylify_metatags', 'bool'),
                     ('repository_fields', 'repository_fields', 'bool'),
                     ('dashboard_items', 'dashboard_items', 'int'),
                     ('admin_grid_items', 'admin_grid_items', 'int'),
                     ('show_version', 'show_version', 'bool'),
                     ('use_gravatar', 'use_gravatar', 'bool'),
                     ('gravatar_url', 'gravatar_url', 'unicode'),
                     ('clone_uri_tmpl', 'clone_uri_tmpl', 'unicode'),
+                ]
                 for setting, form_key, type_ in settings:
                     Setting.create_or_update(setting, form_result[form_key], type_)
                 Session().commit()
                 set_app_settings(config)
                 h.flash(_('Updated visualisation settings'),
                         category='success')
             except Exception:
                 log.error(traceback.format_exc())
                 h.flash(_('Error occurred during updating '
                           'visualisation settings'),
                         category='error')
             raise HTTPFound(location=url('admin_settings_visual'))
         defaults = Setting.get_app_settings()
         defaults.update(self._get_hg_ui_settings())
         return htmlfill.render(
             render('admin/settings/settings.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @HasPermissionAnyDecorator('hg.admin')
     def settings_email(self):
         c.active = 'email'
         if request.POST:
             test_email = request.POST.get('test_email')
             test_email_subj = 'Kallithea test email'
             test_body = ('Kallithea Email test, '
                                'Kallithea version: %s' % c.kallithea_version)
             if not test_email:
                 h.flash(_('Please enter email address'), category='error')
                 raise HTTPFound(location=url('admin_settings_email'))
             test_email_txt_body = EmailNotificationModel() \
                 .get_email_tmpl(EmailNotificationModel.TYPE_DEFAULT,
                                 'txt', body=test_body)
             test_email_html_body = EmailNotificationModel() \
                 .get_email_tmpl(EmailNotificationModel.TYPE_DEFAULT,
                                 'html', body=test_body)
             recipients = [test_email] if test_email else None
             tasks.send_email(recipients, test_email_subj,
                              test_email_txt_body, test_email_html_body)
             h.flash(_('Send email task created'), category='success')
             raise HTTPFound(location=url('admin_settings_email'))
         defaults = Setting.get_app_settings()
         defaults.update(self._get_hg_ui_settings())
         import kallithea
         c.ini = kallithea.CONFIG
         return htmlfill.render(
             render('admin/settings/settings.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @HasPermissionAnyDecorator('hg.admin')
     def settings_hooks(self):
         c.active = 'hooks'
         if request.POST:
             if c.visual.allow_custom_hooks_settings:
                 ui_key = request.POST.get('new_hook_ui_key')
                 ui_value = request.POST.get('new_hook_ui_value')
                 hook_id = request.POST.get('hook_id')
                 try:
                     ui_key = ui_key and ui_key.strip()
                     if ui_value and ui_key:
                         Ui.create_or_update_hook(ui_key, ui_value)
                         h.flash(_('Added new hook'), category='success')
                     elif hook_id:
                         Ui.delete(hook_id)
                         Session().commit()
                     # check for edits
                     update = False
                     _d = request.POST.dict_of_lists()
                     for k, v in zip(_d.get('hook_ui_key', []),
                                     _d.get('hook_ui_value_new', [])):
                         Ui.create_or_update_hook(k, v)
                         update = True
                     if update:
                         h.flash(_('Updated hooks'), category='success')
                     Session().commit()
                 except Exception:
                     log.error(traceback.format_exc())
                     h.flash(_('Error occurred during hook creation'),
                             category='error')
                 raise HTTPFound(location=url('admin_settings_hooks'))
         defaults = Setting.get_app_settings()
         defaults.update(self._get_hg_ui_settings())
         c.hooks = Ui.get_builtin_hooks()
         c.custom_hooks = Ui.get_custom_hooks()
         return htmlfill.render(
             render('admin/settings/settings.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @HasPermissionAnyDecorator('hg.admin')
     def settings_search(self):
         c.active = 'search'
         if request.POST:
             repo_location = self._get_hg_ui_settings()['paths_root_path']
             full_index = request.POST.get('full_index', False)
             tasks.whoosh_index(repo_location, full_index)
             h.flash(_('Whoosh reindex task scheduled'), category='success')
             raise HTTPFound(location=url('admin_settings_search'))
         defaults = Setting.get_app_settings()
         defaults.update(self._get_hg_ui_settings())
         return htmlfill.render(
             render('admin/settings/settings.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @HasPermissionAnyDecorator('hg.admin')
     def settings_system(self):
         c.active = 'system'
         defaults = Setting.get_app_settings()
         defaults.update(self._get_hg_ui_settings())
         import kallithea
         c.ini = kallithea.CONFIG
         c.update_url = defaults.get('update_url')
         server_info = Setting.get_server_info()
         for key, val in server_info.iteritems():
             setattr(c, key, val)
         return htmlfill.render(
             render('admin/settings/settings.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @HasPermissionAnyDecorator('hg.admin')
     def settings_system_update(self):
         import json
         import urllib2
         from kallithea.lib.verlib import NormalizedVersion
         from kallithea import __version__
         defaults = Setting.get_app_settings()
         defaults.update(self._get_hg_ui_settings())
         _update_url = defaults.get('update_url', '')
         _update_url = "" # FIXME: disabled
         _err = lambda s: '<div class="alert alert-danger">%s</div>' % (s)
         try:
             import kallithea
             ver = kallithea.__version__
             log.debug('Checking for upgrade on `%s` server', _update_url)
             opener = urllib2.build_opener()
             opener.addheaders = [('User-agent', 'Kallithea-SCM/%s' % ver)]
             response = opener.open(_update_url)
             response_data = response.read()
             data = json.loads(response_data)
         except urllib2.URLError as e:
             log.error(traceback.format_exc())
             return _err('Failed to contact upgrade server: %r' % e)
         except ValueError as e:
             log.error(traceback.format_exc())
             return _err('Bad data sent from update server')
         latest = data['versions'][0]
         c.update_url = _update_url
         c.latest_data = latest
         c.latest_ver = latest['version']
         c.cur_ver = __version__
         c.should_upgrade = False
         if NormalizedVersion(c.latest_ver) > NormalizedVersion(c.cur_ver):
             c.should_upgrade = True
         c.important_notices = latest['general']
         return render('admin/settings/settings_system_update.html'),

kallithea/controllers/admin/user_groups.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/>.
 """
 kallithea.controllers.admin.user_groups
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 User Groups crud controller
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Jan 25, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 import formencode
 from formencode import htmlfill
 from pylons import request, tmpl_context as c, config
 from pylons.i18n.translation import _
 from tg import request, tmpl_context as c, config
 from tg.i18n import ugettext as _
 from webob.exc import HTTPFound
 from sqlalchemy.orm import joinedload
 from sqlalchemy.sql.expression import func
 from webob.exc import HTTPInternalServerError
 import kallithea
 from kallithea.config.routing import url
 from kallithea.lib import helpers as h
 from kallithea.lib.exceptions import UserGroupsAssignedException, \
     RepoGroupAssignmentError
 from kallithea.lib.utils2 import safe_unicode, safe_int
 from kallithea.lib.auth import LoginRequired, \
     HasUserGroupPermissionLevelDecorator, HasPermissionAnyDecorator
 from kallithea.lib.base import BaseController, render
 from kallithea.model.scm import UserGroupList
 from kallithea.model.user_group import UserGroupModel
 from kallithea.model.repo import RepoModel
 from kallithea.model.db import User, UserGroup, UserGroupToPerm, \
     UserGroupRepoToPerm, UserGroupRepoGroupToPerm
 from kallithea.model.forms import UserGroupForm, UserGroupPermsForm, \
     CustomDefaultPermissionsForm
 from kallithea.model.meta import Session
 from kallithea.lib.utils import action_logger
 log = logging.getLogger(__name__)
 class UserGroupsController(BaseController):
     """REST Controller styled on the Atom Publishing Protocol"""
     @LoginRequired()
     def __before__(self):
         super(UserGroupsController, self).__before__()
         c.available_permissions = config['available_permissions']
     def __load_data(self, user_group_id):
         c.group_members_obj = sorted((x.user for x in c.user_group.members),
                                      key=lambda u: u.username.lower())
         c.group_members = [(x.user_id, x.username) for x in c.group_members_obj]
         c.available_members = sorted(((x.user_id, x.username) for x in
                                       User.query().all()),
                                      key=lambda u: u[1].lower())
     def __load_defaults(self, user_group_id):
         """
         Load defaults settings for edit, and update
         :param user_group_id:
         """
         user_group = UserGroup.get_or_404(user_group_id)
         data = user_group.get_dict()
         return data
     def index(self, format='html'):
         _list = UserGroup.query() \
                         .order_by(func.lower(UserGroup.users_group_name)) \
                         .all()
         group_iter = UserGroupList(_list, perm_level='admin')
         user_groups_data = []
         total_records = len(group_iter)
         _tmpl_lookup = kallithea.CONFIG['pylons.app_globals'].mako_lookup
         template = _tmpl_lookup.get_template('data_table/_dt_elements.html')
         user_group_name = lambda user_group_id, user_group_name: (
             template.get_def("user_group_name")
             .render(user_group_id, user_group_name, _=_, h=h, c=c)
+        )
         user_group_actions = lambda user_group_id, user_group_name: (
             template.get_def("user_group_actions")
             .render(user_group_id, user_group_name, _=_, h=h, c=c)
+        )
         for user_gr in group_iter:
             user_groups_data.append({
                 "raw_name": user_gr.users_group_name,
                 "group_name": user_group_name(user_gr.users_group_id,
                                               user_gr.users_group_name),
                 "desc": h.escape(user_gr.user_group_description),
                 "members": len(user_gr.members),
                 "active": h.boolicon(user_gr.users_group_active),
                 "owner": h.person(user_gr.owner.username),
                 "action": user_group_actions(user_gr.users_group_id, user_gr.users_group_name)
             })
         c.data = {
             "totalRecords": total_records,
             "startIndex": 0,
             "sort": None,
             "dir": "asc",
             "records": user_groups_data
+        }
         return render('admin/user_groups/user_groups.html')
     @HasPermissionAnyDecorator('hg.admin', 'hg.usergroup.create.true')
     def create(self):
         users_group_form = UserGroupForm()()
         try:
             form_result = users_group_form.to_python(dict(request.POST))
             ug = UserGroupModel().create(name=form_result['users_group_name'],
                                          description=form_result['user_group_description'],
                                          owner=request.authuser.user_id,
                                          active=form_result['users_group_active'])
             gr = form_result['users_group_name']
             action_logger(request.authuser,
                           'admin_created_users_group:%s' % gr,
                           None, request.ip_addr)
             h.flash(h.literal(_('Created user group %s') % h.link_to(h.escape(gr), url('edit_users_group', id=ug.users_group_id))),
                 category='success')
             Session().commit()
         except formencode.Invalid as errors:
             return htmlfill.render(
                 render('admin/user_groups/user_group_add.html'),
                 defaults=errors.value,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False)
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('Error occurred during creation of user group %s') \
                     % request.POST.get('users_group_name'), category='error')
         raise HTTPFound(location=url('users_groups'))
     @HasPermissionAnyDecorator('hg.admin', 'hg.usergroup.create.true')
     def new(self, format='html'):
         return render('admin/user_groups/user_group_add.html')
     @HasUserGroupPermissionLevelDecorator('admin')
     def update(self, id):
         c.user_group = UserGroup.get_or_404(id)
         c.active = 'settings'
         self.__load_data(id)
         available_members = [safe_unicode(x[0]) for x in c.available_members]
         users_group_form = UserGroupForm(edit=True,
                                          old_data=c.user_group.get_dict(),
                                          available_members=available_members)()
         try:
             form_result = users_group_form.to_python(request.POST)
             UserGroupModel().update(c.user_group, form_result)
             gr = form_result['users_group_name']
             action_logger(request.authuser,
                           'admin_updated_users_group:%s' % gr,
                           None, request.ip_addr)
             h.flash(_('Updated user group %s') % gr, category='success')
             Session().commit()
         except formencode.Invalid as errors:
             ug_model = UserGroupModel()
             defaults = errors.value
             e = errors.error_dict or {}
             defaults.update({
                 'create_repo_perm': ug_model.has_perm(id,
                                                       'hg.create.repository'),
                 'fork_repo_perm': ug_model.has_perm(id,
                                                     'hg.fork.repository'),
             })
             return htmlfill.render(
                 render('admin/user_groups/user_group_edit.html'),
                 defaults=defaults,
                 errors=e,
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False)
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('Error occurred during update of user group %s') \
                     % request.POST.get('users_group_name'), category='error')
         raise HTTPFound(location=url('edit_users_group', id=id))
     @HasUserGroupPermissionLevelDecorator('admin')
     def delete(self, id):
         usr_gr = UserGroup.get_or_404(id)
         try:
             UserGroupModel().delete(usr_gr)
             Session().commit()
             h.flash(_('Successfully deleted user group'), category='success')
         except UserGroupsAssignedException as e:
             h.flash(e, category='error')
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('An error occurred during deletion of user group'),
                     category='error')
         raise HTTPFound(location=url('users_groups'))
     @HasUserGroupPermissionLevelDecorator('admin')
     def edit(self, id, format='html'):
         c.user_group = UserGroup.get_or_404(id)
         c.active = 'settings'
         self.__load_data(id)
         defaults = self.__load_defaults(id)
         return htmlfill.render(
             render('admin/user_groups/user_group_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False
+        )
     @HasUserGroupPermissionLevelDecorator('admin')
     def edit_perms(self, id):
         c.user_group = UserGroup.get_or_404(id)
         c.active = 'perms'
         repo_model = RepoModel()
         c.users_array = repo_model.get_users_js()
         c.user_groups_array = repo_model.get_user_groups_js()
         defaults = {}
         # fill user group users
         for p in c.user_group.user_user_group_to_perm:
             defaults.update({'u_perm_%s' % p.user.username:
                              p.permission.permission_name})
         for p in c.user_group.user_group_user_group_to_perm:
             defaults.update({'g_perm_%s' % p.user_group.users_group_name:
                              p.permission.permission_name})
         return htmlfill.render(
             render('admin/user_groups/user_group_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False
+        )
     @HasUserGroupPermissionLevelDecorator('admin')
     def update_perms(self, id):
         """
         grant permission for given usergroup
         :param id:
         """
         user_group = UserGroup.get_or_404(id)
         form = UserGroupPermsForm()().to_python(request.POST)
         # set the permissions !
         try:
             UserGroupModel()._update_permissions(user_group, form['perms_new'],
                                                  form['perms_updates'])
         except RepoGroupAssignmentError:
             h.flash(_('Target group cannot be the same'), category='error')
             raise HTTPFound(location=url('edit_user_group_perms', id=id))
         #TODO: implement this
         #action_logger(request.authuser, 'admin_changed_repo_permissions',
         #              repo_name, request.ip_addr)
         Session().commit()
         h.flash(_('User group permissions updated'), category='success')
         raise HTTPFound(location=url('edit_user_group_perms', id=id))
     @HasUserGroupPermissionLevelDecorator('admin')
     def delete_perms(self, id):
         try:
             obj_type = request.POST.get('obj_type')
             obj_id = None
             if obj_type == 'user':
                 obj_id = safe_int(request.POST.get('user_id'))
             elif obj_type == 'user_group':
                 obj_id = safe_int(request.POST.get('user_group_id'))
             if not request.authuser.is_admin:
                 if obj_type == 'user' and request.authuser.user_id == obj_id:
                     msg = _('Cannot revoke permission for yourself as admin')
                     h.flash(msg, category='warning')
                     raise Exception('revoke admin permission on self')
             if obj_type == 'user':
                 UserGroupModel().revoke_user_permission(user_group=id,
                                                         user=obj_id)
             elif obj_type == 'user_group':
                 UserGroupModel().revoke_user_group_permission(target_user_group=id,
                                                               user_group=obj_id)
             Session().commit()
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('An error occurred during revoking of permission'),
                     category='error')
             raise HTTPInternalServerError()
     @HasUserGroupPermissionLevelDecorator('admin')
     def edit_default_perms(self, id):
         c.user_group = UserGroup.get_or_404(id)
         c.active = 'default_perms'
         permissions = {
             'repositories': {},
             'repositories_groups': {}
+        }
         ugroup_repo_perms = UserGroupRepoToPerm.query() \
             .options(joinedload(UserGroupRepoToPerm.permission)) \
             .options(joinedload(UserGroupRepoToPerm.repository)) \
             .filter(UserGroupRepoToPerm.users_group_id == id) \
             .all()
         for gr in ugroup_repo_perms:
             permissions['repositories'][gr.repository.repo_name]  \
                 = gr.permission.permission_name
         ugroup_group_perms = UserGroupRepoGroupToPerm.query() \
             .options(joinedload(UserGroupRepoGroupToPerm.permission)) \
             .options(joinedload(UserGroupRepoGroupToPerm.group)) \
             .filter(UserGroupRepoGroupToPerm.users_group_id == id) \
             .all()
         for gr in ugroup_group_perms:
             permissions['repositories_groups'][gr.group.group_name] \
                 = gr.permission.permission_name
         c.permissions = permissions
         ug_model = UserGroupModel()
         defaults = c.user_group.get_dict()
         defaults.update({
             'create_repo_perm': ug_model.has_perm(c.user_group,
                                                   'hg.create.repository'),
             'create_user_group_perm': ug_model.has_perm(c.user_group,
                                                         'hg.usergroup.create.true'),
             'fork_repo_perm': ug_model.has_perm(c.user_group,
                                                 'hg.fork.repository'),
         })
         return htmlfill.render(
             render('admin/user_groups/user_group_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False
+        )
     @HasUserGroupPermissionLevelDecorator('admin')
     def update_default_perms(self, id):
         user_group = UserGroup.get_or_404(id)
         try:
             form = CustomDefaultPermissionsForm()()
             form_result = form.to_python(request.POST)
             inherit_perms = form_result['inherit_default_permissions']
             user_group.inherit_default_permissions = inherit_perms
             usergroup_model = UserGroupModel()
             defs = UserGroupToPerm.query() \
                 .filter(UserGroupToPerm.users_group == user_group) \
                 .all()
             for ug in defs:
                 Session().delete(ug)
             if form_result['create_repo_perm']:
                 usergroup_model.grant_perm(id, 'hg.create.repository')
             else:
                 usergroup_model.grant_perm(id, 'hg.create.none')
             if form_result['create_user_group_perm']:
                 usergroup_model.grant_perm(id, 'hg.usergroup.create.true')
             else:
                 usergroup_model.grant_perm(id, 'hg.usergroup.create.false')
             if form_result['fork_repo_perm']:
                 usergroup_model.grant_perm(id, 'hg.fork.repository')
             else:
                 usergroup_model.grant_perm(id, 'hg.fork.none')
             h.flash(_("Updated permissions"), category='success')
             Session().commit()
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('An error occurred during permissions saving'),
                     category='error')
         raise HTTPFound(location=url('edit_user_group_default_perms', id=id))
     @HasUserGroupPermissionLevelDecorator('admin')
     def edit_advanced(self, id):
         c.user_group = UserGroup.get_or_404(id)
         c.active = 'advanced'
         c.group_members_obj = sorted((x.user for x in c.user_group.members),
                                      key=lambda u: u.username.lower())
         return render('admin/user_groups/user_group_edit.html')
     @HasUserGroupPermissionLevelDecorator('admin')
     def edit_members(self, id):
         c.user_group = UserGroup.get_or_404(id)
         c.active = 'members'
         c.group_members_obj = sorted((x.user for x in c.user_group.members),
                                      key=lambda u: u.username.lower())
         c.group_members = [(x.user_id, x.username) for x in c.group_members_obj]
         return render('admin/user_groups/user_group_edit.html')

kallithea/controllers/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/>.
 """
 kallithea.controllers.admin.users
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Users crud controller
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 4, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 import formencode
 from formencode import htmlfill
 from pylons import request, tmpl_context as c, config
 from pylons.i18n.translation import _
 from tg import request, tmpl_context as c, config
 from tg.i18n import ugettext as _
 from sqlalchemy.sql.expression import func
 from webob.exc import HTTPFound, HTTPNotFound
 import kallithea
 from kallithea.config.routing import url
 from kallithea.lib.exceptions import DefaultUserException, \
     UserOwnsReposException, UserCreationError
 from kallithea.lib import helpers as h
 from kallithea.lib.auth import LoginRequired, HasPermissionAnyDecorator, \
     AuthUser
 from kallithea.lib import auth_modules
 from kallithea.lib.base import BaseController, render
 from kallithea.model.api_key import ApiKeyModel
 from kallithea.model.db import User, UserEmailMap, UserIpMap, UserToPerm
 from kallithea.model.forms import UserForm, CustomDefaultPermissionsForm
 from kallithea.model.user import UserModel
 from kallithea.model.meta import Session
 from kallithea.lib.utils import action_logger
 from kallithea.lib.utils2 import datetime_to_time, safe_int, generate_api_key
 log = logging.getLogger(__name__)
 class UsersController(BaseController):
     """REST Controller styled on the Atom Publishing Protocol"""
     @LoginRequired()
     @HasPermissionAnyDecorator('hg.admin')
     def __before__(self):
         super(UsersController, self).__before__()
         c.available_permissions = config['available_permissions']
     def index(self, format='html'):
         c.users_list = User.query().order_by(User.username) \
                         .filter_by(is_default_user=False) \
                         .order_by(func.lower(User.username)) \
                         .all()
         users_data = []
         total_records = len(c.users_list)
         _tmpl_lookup = kallithea.CONFIG['pylons.app_globals'].mako_lookup
         template = _tmpl_lookup.get_template('data_table/_dt_elements.html')
         grav_tmpl = '<div class="gravatar">%s</div>'
         username = lambda user_id, username: (
                 template.get_def("user_name")
                 .render(user_id, username, _=_, h=h, c=c))
         user_actions = lambda user_id, username: (
                 template.get_def("user_actions")
                 .render(user_id, username, _=_, h=h, c=c))
         for user in c.users_list:
             users_data.append({
                 "gravatar": grav_tmpl % h.gravatar(user.email, size=20),
                 "raw_name": user.username,
                 "username": username(user.user_id, user.username),
                 "firstname": h.escape(user.name),
                 "lastname": h.escape(user.lastname),
                 "last_login": h.fmt_date(user.last_login),
                 "last_login_raw": datetime_to_time(user.last_login),
                 "active": h.boolicon(user.active),
                 "admin": h.boolicon(user.admin),
                 "extern_type": user.extern_type,
                 "extern_name": user.extern_name,
                 "action": user_actions(user.user_id, user.username),
             })
         c.data = {
             "totalRecords": total_records,
             "startIndex": 0,
             "sort": None,
             "dir": "asc",
             "records": users_data
+        }
         return render('admin/users/users.html')
     def create(self):
         c.default_extern_type = User.DEFAULT_AUTH_TYPE
         c.default_extern_name = ''
         user_model = UserModel()
         user_form = UserForm()()
         try:
             form_result = user_form.to_python(dict(request.POST))
             user = user_model.create(form_result)
             action_logger(request.authuser, 'admin_created_user:%s' % user.username,
                           None, request.ip_addr)
             h.flash(_('Created user %s') % user.username,
                     category='success')
             Session().commit()
         except formencode.Invalid as errors:
             return htmlfill.render(
                 render('admin/users/user_add.html'),
                 defaults=errors.value,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False)
         except UserCreationError as e:
             h.flash(e, 'error')
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('Error occurred during creation of user %s') \
                     % request.POST.get('username'), category='error')
         raise HTTPFound(location=url('edit_user', id=user.user_id))
     def new(self, format='html'):
         c.default_extern_type = User.DEFAULT_AUTH_TYPE
         c.default_extern_name = ''
         return render('admin/users/user_add.html')
     def update(self, id):
         user_model = UserModel()
         user = user_model.get(id)
         _form = UserForm(edit=True, old_data={'user_id': id,
                                               'email': user.email})()
         form_result = {}
         try:
             form_result = _form.to_python(dict(request.POST))
             skip_attrs = ['extern_type', 'extern_name',
                          ] + auth_modules.get_managed_fields(user)
             user_model.update(id, form_result, skip_attrs=skip_attrs)
             usr = form_result['username']
             action_logger(request.authuser, 'admin_updated_user:%s' % usr,
                           None, request.ip_addr)
             h.flash(_('User updated successfully'), category='success')
             Session().commit()
         except formencode.Invalid as errors:
             defaults = errors.value
             e = errors.error_dict or {}
             defaults.update({
                 'create_repo_perm': user_model.has_perm(id,
                                                         'hg.create.repository'),
                 'fork_repo_perm': user_model.has_perm(id, 'hg.fork.repository'),
             })
             return htmlfill.render(
                 self._render_edit_profile(user),
                 defaults=defaults,
                 errors=e,
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False)
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('Error occurred during update of user %s') \
                     % form_result.get('username'), category='error')
         raise HTTPFound(location=url('edit_user', id=id))
     def delete(self, id):
         usr = User.get_or_404(id)
         try:
             UserModel().delete(usr)
             Session().commit()
             h.flash(_('Successfully deleted user'), category='success')
         except (UserOwnsReposException, DefaultUserException) as e:
             h.flash(e, category='warning')
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('An error occurred during deletion of user'),
                     category='error')
         raise HTTPFound(location=url('users'))
     def _get_user_or_raise_if_default(self, id):
         try:
             return User.get_or_404(id, allow_default=False)
         except DefaultUserException:
             h.flash(_("The default user cannot be edited"), category='warning')
             raise HTTPNotFound
     def _render_edit_profile(self, user):
         c.user = user
         c.active = 'profile'
         c.perm_user = AuthUser(dbuser=user)
         managed_fields = auth_modules.get_managed_fields(user)
         c.readonly = lambda n: 'readonly' if n in managed_fields else None
         return render('admin/users/user_edit.html')
     def edit(self, id, format='html'):
         user = self._get_user_or_raise_if_default(id)
         defaults = user.get_dict()
         return htmlfill.render(
             self._render_edit_profile(user),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     def edit_advanced(self, id):
         c.user = self._get_user_or_raise_if_default(id)
         c.active = 'advanced'
         c.perm_user = AuthUser(dbuser=c.user)
         umodel = UserModel()
         defaults = c.user.get_dict()
         defaults.update({
             'create_repo_perm': umodel.has_perm(c.user, 'hg.create.repository'),
             'create_user_group_perm': umodel.has_perm(c.user,
                                                       'hg.usergroup.create.true'),
             'fork_repo_perm': umodel.has_perm(c.user, 'hg.fork.repository'),
         })
         return htmlfill.render(
             render('admin/users/user_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     def edit_api_keys(self, id):
         c.user = self._get_user_or_raise_if_default(id)
         c.active = 'api_keys'
         show_expired = True
         c.lifetime_values = [
             (str(-1), _('Forever')),
             (str(5), _('5 minutes')),
             (str(60), _('1 hour')),
             (str(60 * 24), _('1 day')),
             (str(60 * 24 * 30), _('1 month')),
+        ]
         c.lifetime_options = [(c.lifetime_values, _("Lifetime"))]
         c.user_api_keys = ApiKeyModel().get_api_keys(c.user.user_id,
                                                      show_expired=show_expired)
         defaults = c.user.get_dict()
         return htmlfill.render(
             render('admin/users/user_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     def add_api_key(self, id):
         c.user = self._get_user_or_raise_if_default(id)
         lifetime = safe_int(request.POST.get('lifetime'), -1)
         description = request.POST.get('description')
         ApiKeyModel().create(c.user.user_id, description, lifetime)
         Session().commit()
         h.flash(_("API key successfully created"), category='success')
         raise HTTPFound(location=url('edit_user_api_keys', id=c.user.user_id))
     def delete_api_key(self, id):
         c.user = self._get_user_or_raise_if_default(id)
         api_key = request.POST.get('del_api_key')
         if request.POST.get('del_api_key_builtin'):
             c.user.api_key = generate_api_key()
             Session().commit()
             h.flash(_("API key successfully reset"), category='success')
         elif api_key:
             ApiKeyModel().delete(api_key, c.user.user_id)
             Session().commit()
             h.flash(_("API key successfully deleted"), category='success')
         raise HTTPFound(location=url('edit_user_api_keys', id=c.user.user_id))
     def update_account(self, id):
         pass
     def edit_perms(self, id):
         c.user = self._get_user_or_raise_if_default(id)
         c.active = 'perms'
         c.perm_user = AuthUser(dbuser=c.user)
         umodel = UserModel()
         defaults = c.user.get_dict()
         defaults.update({
             'create_repo_perm': umodel.has_perm(c.user, 'hg.create.repository'),
             'create_user_group_perm': umodel.has_perm(c.user,
                                                       'hg.usergroup.create.true'),
             'fork_repo_perm': umodel.has_perm(c.user, 'hg.fork.repository'),
         })
         return htmlfill.render(
             render('admin/users/user_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     def update_perms(self, id):
         user = self._get_user_or_raise_if_default(id)
         try:
             form = CustomDefaultPermissionsForm()()
             form_result = form.to_python(request.POST)
             inherit_perms = form_result['inherit_default_permissions']
             user.inherit_default_permissions = inherit_perms
             user_model = UserModel()
             defs = UserToPerm.query() \
                 .filter(UserToPerm.user == user) \
                 .all()
             for ug in defs:
                 Session().delete(ug)
             if form_result['create_repo_perm']:
                 user_model.grant_perm(id, 'hg.create.repository')
             else:
                 user_model.grant_perm(id, 'hg.create.none')
             if form_result['create_user_group_perm']:
                 user_model.grant_perm(id, 'hg.usergroup.create.true')
             else:
                 user_model.grant_perm(id, 'hg.usergroup.create.false')
             if form_result['fork_repo_perm']:
                 user_model.grant_perm(id, 'hg.fork.repository')
             else:
                 user_model.grant_perm(id, 'hg.fork.none')
             h.flash(_("Updated permissions"), category='success')
             Session().commit()
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('An error occurred during permissions saving'),
                     category='error')
         raise HTTPFound(location=url('edit_user_perms', id=id))
     def edit_emails(self, id):
         c.user = self._get_user_or_raise_if_default(id)
         c.active = 'emails'
         c.user_email_map = UserEmailMap.query() \
             .filter(UserEmailMap.user == c.user).all()
         defaults = c.user.get_dict()
         return htmlfill.render(
             render('admin/users/user_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     def add_email(self, id):
         user = self._get_user_or_raise_if_default(id)
         email = request.POST.get('new_email')
         user_model = UserModel()
         try:
             user_model.add_extra_email(id, email)
             Session().commit()
             h.flash(_("Added email %s to user") % email, category='success')
         except formencode.Invalid as error:
             msg = error.error_dict['email']
             h.flash(msg, category='error')
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('An error occurred during email saving'),
                     category='error')
         raise HTTPFound(location=url('edit_user_emails', id=id))
     def delete_email(self, id):
         user = self._get_user_or_raise_if_default(id)
         email_id = request.POST.get('del_email_id')
         user_model = UserModel()
         user_model.delete_extra_email(id, email_id)
         Session().commit()
         h.flash(_("Removed email from user"), category='success')
         raise HTTPFound(location=url('edit_user_emails', id=id))
     def edit_ips(self, id):
         c.user = self._get_user_or_raise_if_default(id)
         c.active = 'ips'
         c.user_ip_map = UserIpMap.query() \
             .filter(UserIpMap.user == c.user).all()
         c.inherit_default_ips = c.user.inherit_default_permissions
         c.default_user_ip_map = UserIpMap.query() \
             .filter(UserIpMap.user == User.get_default_user()).all()
         defaults = c.user.get_dict()
         return htmlfill.render(
             render('admin/users/user_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     def add_ip(self, id):
         ip = request.POST.get('new_ip')
         user_model = UserModel()
         try:
             user_model.add_extra_ip(id, ip)
             Session().commit()
             h.flash(_("Added IP address %s to user whitelist") % ip, category='success')
         except formencode.Invalid as error:
             msg = error.error_dict['ip']
             h.flash(msg, category='error')
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('An error occurred while adding IP address'),
                     category='error')
         if 'default_user' in request.POST:
             raise HTTPFound(location=url('admin_permissions_ips'))
         raise HTTPFound(location=url('edit_user_ips', id=id))
     def delete_ip(self, id):
         ip_id = request.POST.get('del_ip_id')
         user_model = UserModel()
         user_model.delete_extra_ip(id, ip_id)
         Session().commit()
         h.flash(_("Removed IP address from user whitelist"), category='success')
         if 'default_user' in request.POST:
             raise HTTPFound(location=url('admin_permissions_ips'))
         raise HTTPFound(location=url('edit_user_ips', id=id))

kallithea/controllers/api/__init__.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/>.
 """
 kallithea.controllers.api
 ~~~~~~~~~~~~~~~~~~~~~~~~~
 JSON RPC controller
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Aug 20, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import inspect
 import logging
 import types
 import traceback
 import time
 import itertools
 from paste.response import replace_header
 from pylons.controllers import WSGIController
 from pylons.controllers.util import Response
-from pylons import request
+from tg import request
 from webob.exc import HTTPError
 from kallithea.model.db import User
 from kallithea.model import meta
 from kallithea.lib.compat import json
 from kallithea.lib.auth import AuthUser
 from kallithea.lib.base import _get_ip_addr as _get_ip, _get_access_path
 from kallithea.lib.utils2 import safe_unicode, safe_str
 log = logging.getLogger('JSONRPC')
 class JSONRPCError(BaseException):
     def __init__(self, message):
         self.message = message
         super(JSONRPCError, self).__init__()
     def __str__(self):
         return safe_str(self.message)
 class JSONRPCErrorResponse(Response, Exception):
     """
     Generate a Response object with a JSON-RPC error body
     """
     def __init__(self, message=None, retid=None, code=None):
         Response.__init__(self,
                           body=json.dumps(dict(id=retid, result=None, error=message)),
                           status=code,
                           content_type='application/json')
 class JSONRPCController(WSGIController):
     """
      A WSGI-speaking JSON-RPC controller class
      See the specification:
      <http://json-rpc.org/wiki/specification>`.
      Valid controller return values should be json-serializable objects.
      Sub-classes should catch their exceptions and raise JSONRPCError
      if they want to pass meaningful errors to the client.
      """
     def _get_ip_addr(self, environ):
         return _get_ip(environ)
     def _get_method_args(self):
         """
         Return `self._rpc_args` to dispatched controller method
         chosen by __call__
         """
         return self._rpc_args
     def __call__(self, environ, start_response):
         """
         Parse the request body as JSON, look up the method on the
         controller and if it exists, dispatch to it.
         """
         try:
             return self._handle_request(environ, start_response)
         except JSONRPCErrorResponse as e:
             return e
         finally:
             meta.Session.remove()
     def _handle_request(self, environ, start_response):
         start = time.time()
         ip_addr = request.ip_addr = self._get_ip_addr(environ)
         self._req_id = None
         if 'CONTENT_LENGTH' not in environ:
             log.debug("No Content-Length")
             raise JSONRPCErrorResponse(retid=self._req_id,
                                        message="No Content-Length in request")
         else:
             length = environ['CONTENT_LENGTH'] or 0
             length = int(environ['CONTENT_LENGTH'])
             log.debug('Content-Length: %s', length)
         if length == 0:
             raise JSONRPCErrorResponse(retid=self._req_id,
                                        message="Content-Length is 0")
         raw_body = environ['wsgi.input'].read(length)
         try:
             json_body = json.loads(raw_body)
         except ValueError as e:
             # catch JSON errors Here
             raise JSONRPCErrorResponse(retid=self._req_id,
                                        message="JSON parse error ERR:%s RAW:%r"
                                                 % (e, raw_body))
         # check AUTH based on API key
         try:
             self._req_api_key = json_body['api_key']
             self._req_id = json_body['id']
             self._req_method = json_body['method']
             self._request_params = json_body['args']
             if not isinstance(self._request_params, dict):
                 self._request_params = {}
             log.debug('method: %s, params: %s',
                       self._req_method, self._request_params)
         except KeyError as e:
             raise JSONRPCErrorResponse(retid=self._req_id,
                                        message='Incorrect JSON query missing %s' % e)
         # check if we can find this session using api_key
         try:
             u = User.get_by_api_key(self._req_api_key)
             if u is None:
                 raise JSONRPCErrorResponse(retid=self._req_id,
                                            message='Invalid API key')
             auth_u = AuthUser(dbuser=u)
             if not AuthUser.check_ip_allowed(auth_u, ip_addr):
                 raise JSONRPCErrorResponse(retid=self._req_id,
                                            message='request from IP:%s not allowed' % (ip_addr,))
             else:
                 log.info('Access for IP:%s allowed', ip_addr)
         except Exception as e:
             raise JSONRPCErrorResponse(retid=self._req_id,
                                        message='Invalid API key')
         self._error = None
         try:
             self._func = self._find_method()
         except AttributeError as e:
             raise JSONRPCErrorResponse(retid=self._req_id,
                                        message=str(e))
         # now that we have a method, add self._req_params to
         # self.kargs and dispatch control to WGIController
         argspec = inspect.getargspec(self._func)
         arglist = argspec[0][1:]
         defaults = map(type, argspec[3] or [])
         default_empty = types.NotImplementedType
         # kw arguments required by this method
         func_kwargs = dict(itertools.izip_longest(reversed(arglist), reversed(defaults),
                                                   fillvalue=default_empty))
         # this is little trick to inject logged in user for
         # perms decorators to work they expect the controller class to have
         # authuser attribute set
         request.authuser = request.user = auth_u
         # This attribute will need to be first param of a method that uses
         # api_key, which is translated to instance of user at that name
         USER_SESSION_ATTR = 'apiuser'
         # get our arglist and check if we provided them as args
         for arg, default in func_kwargs.iteritems():
             if arg == USER_SESSION_ATTR:
                 # USER_SESSION_ATTR is something translated from API key and
                 # this is checked before so we don't need validate it
                 continue
             # skip the required param check if it's default value is
             # NotImplementedType (default_empty)
             if default == default_empty and arg not in self._request_params:
                 raise JSONRPCErrorResponse(
                     retid=self._req_id,
                     message='Missing non optional `%s` arg in JSON DATA' % arg,
+                )
         extra = set(self._request_params).difference(func_kwargs)
         if extra:
                 raise JSONRPCErrorResponse(
                     retid=self._req_id,
                     message='Unknown %s arg in JSON DATA' %
                             ', '.join('`%s`' % arg for arg in extra),
+                )
         self._rpc_args = {}
         self._rpc_args.update(self._request_params)
         self._rpc_args['action'] = self._req_method
         self._rpc_args['environ'] = environ
         self._rpc_args['start_response'] = start_response
         status = []
         headers = []
         exc_info = []
         def change_content(new_status, new_headers, new_exc_info=None):
             status.append(new_status)
             headers.extend(new_headers)
             exc_info.append(new_exc_info)
         output = WSGIController.__call__(self, environ, change_content)
         output = list(output) # expand iterator - just to ensure exact timing
         replace_header(headers, 'Content-Type', 'application/json')
         start_response(status[0], headers, exc_info[0])
         log.info('IP: %s Request to %s time: %.3fs' % (
             self._get_ip_addr(environ),
             safe_unicode(_get_access_path(environ)), time.time() - start)
+        )
         return output
     def _dispatch_call(self):
         """
         Implement dispatch interface specified by WSGIController
         """
         raw_response = ''
         try:
             raw_response = self._inspect_call(self._func)
             if isinstance(raw_response, HTTPError):
                 self._error = str(raw_response)
         except JSONRPCError as e:
             self._error = safe_str(e)
         except Exception as e:
             log.error('Encountered unhandled exception: %s',
                       traceback.format_exc(),)
             json_exc = JSONRPCError('Internal server error')
             self._error = safe_str(json_exc)
         if self._error is not None:
             raw_response = None
         response = dict(id=self._req_id, result=raw_response, error=self._error)
         try:
             return json.dumps(response)
         except TypeError as e:
             log.error('API FAILED. Error encoding response: %s', e)
             return json.dumps(
                 dict(
                     id=self._req_id,
                     result=None,
                     error="Error encoding response"
+                )
+            )
     def _find_method(self):
         """
         Return method named by `self._req_method` in controller if able
         """
         log.debug('Trying to find JSON-RPC method: %s', self._req_method)
         if self._req_method.startswith('_'):
             raise AttributeError("Method not allowed")
         try:
             func = getattr(self, self._req_method, None)
         except UnicodeEncodeError:
             raise AttributeError("Problem decoding unicode in requested "
                                  "method name.")
         if isinstance(func, types.MethodType):
             return func
         else:
             raise AttributeError("No such method: %s" % (self._req_method,))

kallithea/controllers/api/api.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/>.
 """
 kallithea.controllers.api.api
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 API controller for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Aug 20, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import time
 import traceback
 import logging
 from sqlalchemy import or_
-from pylons import request
+from tg import request
 from kallithea.controllers.api import JSONRPCController, JSONRPCError
 from kallithea.lib.auth import (
     PasswordGenerator, AuthUser, HasPermissionAnyDecorator,
     HasPermissionAnyDecorator, HasPermissionAny, HasRepoPermissionLevel,
     HasRepoGroupPermissionLevel, HasUserGroupPermissionLevel)
 from kallithea.lib.utils import map_groups, repo2db_mapper
 from kallithea.lib.utils2 import (
     str2bool, time_to_datetime, safe_int, Optional, OAttr)
 from kallithea.model.meta import Session
 from kallithea.model.repo_group import RepoGroupModel
 from kallithea.model.scm import ScmModel, UserGroupList
 from kallithea.model.repo import RepoModel
 from kallithea.model.user import UserModel
 from kallithea.model.user_group import UserGroupModel
 from kallithea.model.gist import GistModel
 from kallithea.model.db import (
     Repository, Setting, UserIpMap, Permission, User, Gist,
     RepoGroup, UserGroup)
 from kallithea.lib.compat import json
 from kallithea.lib.exceptions import (
     DefaultUserException, UserGroupsAssignedException)
 log = logging.getLogger(__name__)
 def store_update(updates, attr, name):
     """
     Stores param in updates dict if it's not instance of Optional
     allows easy updates of passed in params
     """
     if not isinstance(attr, Optional):
         updates[name] = attr
 def get_user_or_error(userid):
     """
     Get user by id or name or return JsonRPCError if not found
     :param userid:
     """
     user = UserModel().get_user(userid)
     if user is None:
         raise JSONRPCError("user `%s` does not exist" % (userid,))
     return user
 def get_repo_or_error(repoid):
     """
     Get repo by id or name or return JsonRPCError if not found
     :param repoid:
     """
     repo = RepoModel().get_repo(repoid)
     if repo is None:
         raise JSONRPCError('repository `%s` does not exist' % (repoid,))
     return repo
 def get_repo_group_or_error(repogroupid):
     """
     Get repo group by id or name or return JsonRPCError if not found
     :param repogroupid:
     """
     repo_group = RepoGroup.guess_instance(repogroupid)
     if repo_group is None:
         raise JSONRPCError(
             'repository group `%s` does not exist' % (repogroupid,))
     return repo_group
 def get_user_group_or_error(usergroupid):
     """
     Get user group by id or name or return JsonRPCError if not found
     :param usergroupid:
     """
     user_group = UserGroupModel().get_group(usergroupid)
     if user_group is None:
         raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
     return user_group
 def get_perm_or_error(permid, prefix=None):
     """
     Get permission by id or name or return JsonRPCError if not found
     :param permid:
     """
     perm = Permission.get_by_key(permid)
     if perm is None:
         raise JSONRPCError('permission `%s` does not exist' % (permid,))
     if prefix:
         if not perm.permission_name.startswith(prefix):
             raise JSONRPCError('permission `%s` is invalid, '
                                'should start with %s' % (permid, prefix))
     return perm
 def get_gist_or_error(gistid):
     """
     Get gist by id or gist_access_id or return JsonRPCError if not found
     :param gistid:
     """
     gist = GistModel().get_gist(gistid)
     if gist is None:
         raise JSONRPCError('gist `%s` does not exist' % (gistid,))
     return gist
 class ApiController(JSONRPCController):
     """
     API Controller
     The authenticated user can be found as request.authuser.
     Example function::
         def func(arg1, arg2,...):
             pass
     Each function should also **raise** JSONRPCError for any
     errors that happens.
     """
     @HasPermissionAnyDecorator('hg.admin')
     def test(self, args):
         return args
     @HasPermissionAnyDecorator('hg.admin')
     def pull(self, repoid):
         """
         Triggers a pull from remote location on given repo. Can be used to
         automatically keep remote repos up to date. This command can be executed
         only using api_key belonging to user with admin rights
         :param repoid: repository name or repository id
         :type repoid: str or int
         OUTPUT::
           id : <id_given_in_input>
           result : {
             "msg": "Pulled from `<repository name>`"
             "repository": "<repository name>"
+          }
           error :  null
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           error :  {
             "Unable to pull changes from `<reponame>`"
+          }
         """
         repo = get_repo_or_error(repoid)
         try:
             ScmModel().pull_changes(repo.repo_name,
                                     request.authuser.username)
             return dict(
                 msg='Pulled from `%s`' % repo.repo_name,
                 repository=repo.repo_name
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'Unable to pull changes from `%s`' % repo.repo_name
+            )
     @HasPermissionAnyDecorator('hg.admin')
     def rescan_repos(self, remove_obsolete=Optional(False)):
         """
         Triggers rescan repositories action. If remove_obsolete is set
         than also delete repos that are in database but not in the filesystem.
         aka "clean zombies". This command can be executed only using api_key
         belonging to user with admin rights.
         :param remove_obsolete: deletes repositories from
             database that are not found on the filesystem
         :type remove_obsolete: Optional(bool)
         OUTPUT::
           id : <id_given_in_input>
           result : {
             'added': [<added repository name>,...]
             'removed': [<removed repository name>,...]
+          }
           error :  null
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           error :  {
             'Error occurred during rescan repositories action'
+          }
         """
         try:
             rm_obsolete = Optional.extract(remove_obsolete)
             added, removed = repo2db_mapper(ScmModel().repo_scan(),
                                             remove_obsolete=rm_obsolete)
             return {'added': added, 'removed': removed}
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'Error occurred during rescan repositories action'
+            )
     def invalidate_cache(self, repoid):
         """
         Invalidate cache for repository.
         This command can be executed only using api_key belonging to user with admin
         rights or regular user that have write or admin or write access to repository.
         :param repoid: repository name or repository id
         :type repoid: str or int
         OUTPUT::
           id : <id_given_in_input>
           result : {
             'msg': Cache for repository `<repository name>` was invalidated,
             'repository': <repository name>
+          }
           error :  null
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           error :  {
             'Error occurred during cache invalidation action'
+          }
         """
         repo = get_repo_or_error(repoid)
         if not HasPermissionAny('hg.admin')():
             if not HasRepoPermissionLevel('write')(repo.repo_name):
                 raise JSONRPCError('repository `%s` does not exist' % (repoid,))
         try:
             ScmModel().mark_for_invalidation(repo.repo_name)
             return dict(
                 msg='Cache for repository `%s` was invalidated' % (repoid,),
                 repository=repo.repo_name
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'Error occurred during cache invalidation action'
+            )
     # permission check inside
     def lock(self, repoid, locked=Optional(None),
              userid=Optional(OAttr('apiuser'))):
         """
         Set locking state on given repository by given user. If userid param
         is skipped, then it is set to id of user who is calling this method.
         If locked param is skipped then function shows current lock state of
         given repo. This command can be executed only using api_key belonging
         to user with admin rights or regular user that have admin or write
         access to repository.
         :param repoid: repository name or repository id
         :type repoid: str or int
         :param locked: lock state to be set
         :type locked: Optional(bool)
         :param userid: set lock as user
         :type userid: Optional(str or int)
         OUTPUT::
           id : <id_given_in_input>
           result : {
             'repo': '<reponame>',
             'locked': <bool: lock state>,
             'locked_since': <int: lock timestamp>,
             'locked_by': <username of person who made the lock>,
             'lock_state_changed': <bool: True if lock state has been changed in this request>,
             'msg': 'Repo `<reponame>` locked by `<username>` on <timestamp>.'
             or
             'msg': 'Repo `<repository name>` not locked.'
             or
             'msg': 'User `<user name>` set lock state for repo `<repository name>` to `<new lock state>`'
+          }
           error :  null
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           error :  {
             'Error occurred locking repository `<reponame>`
+          }
         """
         repo = get_repo_or_error(repoid)
         if HasPermissionAny('hg.admin')():
             pass
         elif HasRepoPermissionLevel('write')(repo.repo_name):
             # make sure normal user does not pass someone else userid,
             # he is not allowed to do that
             if not isinstance(userid, Optional) and userid != request.authuser.user_id:
                 raise JSONRPCError(
                     'userid is not the same as your user'
+                )
         else:
             raise JSONRPCError('repository `%s` does not exist' % (repoid,))
         if isinstance(userid, Optional):
             userid = request.authuser.user_id
         user = get_user_or_error(userid)
         if isinstance(locked, Optional):
             lockobj = Repository.getlock(repo)
             if lockobj[0] is None:
                 _d = {
                     'repo': repo.repo_name,
                     'locked': False,
                     'locked_since': None,
                     'locked_by': None,
                     'lock_state_changed': False,
                     'msg': 'Repo `%s` not locked.' % repo.repo_name
+                }
                 return _d
             else:
                 userid, time_ = lockobj
                 lock_user = get_user_or_error(userid)
                 _d = {
                     'repo': repo.repo_name,
                     'locked': True,
                     'locked_since': time_,
                     'locked_by': lock_user.username,
                     'lock_state_changed': False,
                     'msg': ('Repo `%s` locked by `%s` on `%s`.'
                             % (repo.repo_name, lock_user.username,
                                json.dumps(time_to_datetime(time_))))
+                }
                 return _d
         # force locked state through a flag
         else:
             locked = str2bool(locked)
             try:
                 if locked:
                     lock_time = time.time()
                     Repository.lock(repo, user.user_id, lock_time)
                 else:
                     lock_time = None
                     Repository.unlock(repo)
                 _d = {
                     'repo': repo.repo_name,
                     'locked': locked,
                     'locked_since': lock_time,
                     'locked_by': user.username,
                     'lock_state_changed': True,
                     'msg': ('User `%s` set lock state for repo `%s` to `%s`'
                             % (user.username, repo.repo_name, locked))
+                }
                 return _d
             except Exception:
                 log.error(traceback.format_exc())
                 raise JSONRPCError(
                     'Error occurred locking repository `%s`' % repo.repo_name
+                )
     def get_locks(self, userid=Optional(OAttr('apiuser'))):
         """
         Get all repositories with locks for given userid, if
         this command is run by non-admin account userid is set to user
         who is calling this method, thus returning locks for himself.
         :param userid: User to get locks for
         :type userid: Optional(str or int)
         OUTPUT::
           id : <id_given_in_input>
           result : {
             [repo_object, repo_object,...]
+          }
           error :  null
         """
         if not HasPermissionAny('hg.admin')():
             # make sure normal user does not pass someone else userid,
             # he is not allowed to do that
             if not isinstance(userid, Optional) and userid != request.authuser.user_id:
                 raise JSONRPCError(
                     'userid is not the same as your user'
+                )
         ret = []
         if isinstance(userid, Optional):
             user = None
         else:
             user = get_user_or_error(userid)
         # show all locks
         for r in Repository.query():
             userid, time_ = r.locked
             if time_:
                 _api_data = r.get_api_data()
                 # if we use userfilter just show the locks for this user
                 if user is not None:
                     if safe_int(userid) == user.user_id:
                         ret.append(_api_data)
                 else:
                     ret.append(_api_data)
         return ret
     @HasPermissionAnyDecorator('hg.admin')
     def get_ip(self, userid=Optional(OAttr('apiuser'))):
         """
         Shows IP address as seen from Kallithea server, together with all
         defined IP addresses for given user. If userid is not passed data is
         returned for user who's calling this function.
         This command can be executed only using api_key belonging to user with
         admin rights.
         :param userid: username to show ips for
         :type userid: Optional(str or int)
         OUTPUT::
             id : <id_given_in_input>
             result : {
                          "server_ip_addr": "<ip_from_clien>",
                          "user_ips": [
+                                        {
                                            "ip_addr": "<ip_with_mask>",
                                            "ip_range": ["<start_ip>", "<end_ip>"],
                                         },
                                         ...
+                                     ]
+            }
         """
         if isinstance(userid, Optional):
             userid = request.authuser.user_id
         user = get_user_or_error(userid)
         ips = UserIpMap.query().filter(UserIpMap.user == user).all()
         return dict(
             server_ip_addr=request.ip_addr,
             user_ips=ips
+        )
     # alias for old
     show_ip = get_ip
     @HasPermissionAnyDecorator('hg.admin')
     def get_server_info(self):
         """
         return server info, including Kallithea version and installed packages
         OUTPUT::
           id : <id_given_in_input>
           result : {
             'modules': [<module name>,...]
             'py_version': <python version>,
             'platform': <platform type>,
             'kallithea_version': <kallithea version>
+          }
           error :  null
         """
         return Setting.get_server_info()
     def get_user(self, userid=Optional(OAttr('apiuser'))):
         """
         Gets a user by username or user_id, Returns empty result if user is
         not found. If userid param is skipped it is set to id of user who is
         calling this method. This command can be executed only using api_key
         belonging to user with admin rights, or regular users that cannot
         specify different userid than theirs
         :param userid: user to get data for
         :type userid: Optional(str or int)
         OUTPUT::
             id : <id_given_in_input>
             result: None if user does not exist or
+                    {
                         "user_id" :     "<user_id>",
                         "api_key" :     "<api_key>",
                         "api_keys":     "[<list of all API keys including additional ones>]"
                         "username" :    "<username>",
                         "firstname":    "<firstname>",
                         "lastname" :    "<lastname>",
                         "email" :       "<email>",
                         "emails":       "[<list of all emails including additional ones>]",
                         "ip_addresses": "[<ip_address_for_user>,...]",
                         "active" :      "<bool: user active>",
                         "admin" :       "<bool: user is admin>",
                         "extern_name" : "<extern_name>",
                         "extern_type" : "<extern type>
                         "last_login":   "<last_login>",
                         "permissions": {
                             "global": ["hg.create.repository",
                                        "repository.read",
                                        "hg.register.manual_activate"],
                             "repositories": {"repo1": "repository.none"},
                             "repositories_groups": {"Group1": "group.read"}
                          },
+                    }
             error:  null
         """
         if not HasPermissionAny('hg.admin')():
             # make sure normal user does not pass someone else userid,
             # he is not allowed to do that
             if not isinstance(userid, Optional) and userid != request.authuser.user_id:
                 raise JSONRPCError(
                     'userid is not the same as your user'
+                )
         if isinstance(userid, Optional):
             userid = request.authuser.user_id
         user = get_user_or_error(userid)
         data = user.get_api_data()
         data['permissions'] = AuthUser(user_id=user.user_id).permissions
         return data
     @HasPermissionAnyDecorator('hg.admin')
     def get_users(self):
         """
         Lists all existing users. This command can be executed only using api_key
         belonging to user with admin rights.
         OUTPUT::
             id : <id_given_in_input>
             result: [<user_object>, ...]
             error:  null
         """
         return [
             user.get_api_data()
             for user in User.query()
                 .order_by(User.username)
                 .filter_by(is_default_user=False)
+        ]
     @HasPermissionAnyDecorator('hg.admin')
     def create_user(self, username, email, password=Optional(''),
                     firstname=Optional(''), lastname=Optional(''),
                     active=Optional(True), admin=Optional(False),
                     extern_type=Optional(User.DEFAULT_AUTH_TYPE),
                     extern_name=Optional('')):
         """
         Creates new user. Returns new user object. This command can
         be executed only using api_key belonging to user with admin rights.
         :param username: new username
         :type username: str or int
         :param email: email
         :type email: str
         :param password: password
         :type password: Optional(str)
         :param firstname: firstname
         :type firstname: Optional(str)
         :param lastname: lastname
         :type lastname: Optional(str)
         :param active: active
         :type active: Optional(bool)
         :param admin: admin
         :type admin: Optional(bool)
         :param extern_name: name of extern
         :type extern_name: Optional(str)
         :param extern_type: extern_type
         :type extern_type: Optional(str)
         OUTPUT::
             id : <id_given_in_input>
             result: {
                       "msg" : "created new user `<username>`",
                       "user": <user_obj>
+                    }
             error:  null
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           error :  {
             "user `<username>` already exist"
             or
             "email `<email>` already exist"
             or
             "failed to create user `<username>`"
+          }
         """
         if User.get_by_username(username):
             raise JSONRPCError("user `%s` already exist" % (username,))
         if User.get_by_email(email):
             raise JSONRPCError("email `%s` already exist" % (email,))
         try:
             user = UserModel().create_or_update(
                 username=Optional.extract(username),
                 password=Optional.extract(password),
                 email=Optional.extract(email),
                 firstname=Optional.extract(firstname),
                 lastname=Optional.extract(lastname),
                 active=Optional.extract(active),
                 admin=Optional.extract(admin),
                 extern_type=Optional.extract(extern_type),
                 extern_name=Optional.extract(extern_name)
+            )
             Session().commit()
             return dict(
                 msg='created new user `%s`' % username,
                 user=user.get_api_data()
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to create user `%s`' % (username,))
     @HasPermissionAnyDecorator('hg.admin')
     def update_user(self, userid, username=Optional(None),
                     email=Optional(None), password=Optional(None),
                     firstname=Optional(None), lastname=Optional(None),
                     active=Optional(None), admin=Optional(None),
                     extern_type=Optional(None), extern_name=Optional(None)):
         """
         updates given user if such user exists. This command can
         be executed only using api_key belonging to user with admin rights.
         :param userid: userid to update
         :type userid: str or int
         :param username: new username
         :type username: str or int
         :param email: email
         :type email: str
         :param password: password
         :type password: Optional(str)
         :param firstname: firstname
         :type firstname: Optional(str)
         :param lastname: lastname
         :type lastname: Optional(str)
         :param active: active
         :type active: Optional(bool)
         :param admin: admin
         :type admin: Optional(bool)
         :param extern_name:
         :type extern_name: Optional(str)
         :param extern_type:
         :type extern_type: Optional(str)
         OUTPUT::
             id : <id_given_in_input>
             result: {
                       "msg" : "updated user ID:<userid> <username>",
                       "user": <user_object>,
+                    }
             error:  null
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           error :  {
             "failed to update user `<username>`"
+          }
         """
         user = get_user_or_error(userid)
         # only non optional arguments will be stored in updates
         updates = {}
         try:
             store_update(updates, username, 'username')
             store_update(updates, password, 'password')
             store_update(updates, email, 'email')
             store_update(updates, firstname, 'name')
             store_update(updates, lastname, 'lastname')
             store_update(updates, active, 'active')
             store_update(updates, admin, 'admin')
             store_update(updates, extern_name, 'extern_name')
             store_update(updates, extern_type, 'extern_type')
             user = UserModel().update_user(user, **updates)
             Session().commit()
             return dict(
                 msg='updated user ID:%s %s' % (user.user_id, user.username),
                 user=user.get_api_data()
+            )
         except DefaultUserException:
             log.error(traceback.format_exc())
             raise JSONRPCError('editing default user is forbidden')
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to update user `%s`' % (userid,))
     @HasPermissionAnyDecorator('hg.admin')
     def delete_user(self, userid):
         """
         deletes given user if such user exists. This command can
         be executed only using api_key belonging to user with admin rights.
         :param userid: user to delete
         :type userid: str or int
         OUTPUT::
             id : <id_given_in_input>
             result: {
                       "msg" : "deleted user ID:<userid> <username>",
                       "user": null
+                    }
             error:  null
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           error :  {
             "failed to delete user ID:<userid> <username>"
+          }
         """
         user = get_user_or_error(userid)
         try:
             UserModel().delete(userid)
             Session().commit()
             return dict(
                 msg='deleted user ID:%s %s' % (user.user_id, user.username),
                 user=None
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to delete user ID:%s %s'
                                % (user.user_id, user.username))
     # permission check inside
     def get_user_group(self, usergroupid):
         """
         Gets an existing user group. This command can be executed only using api_key
         belonging to user with admin rights or user who has at least

kallithea/controllers/changelog.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/>.
 """
 kallithea.controllers.changelog
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 changelog controller for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 21, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 from pylons import request, session, tmpl_context as c
 from pylons.i18n.translation import _
 from tg import request, session, tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPFound, HTTPNotFound, HTTPBadRequest
 import kallithea.lib.helpers as h
 from kallithea.config.routing import url
 from kallithea.lib.auth import LoginRequired, HasRepoPermissionLevelDecorator
 from kallithea.lib.base import BaseRepoController, render
 from kallithea.lib.graphmod import graph_data
 from kallithea.lib.page import RepoPage
 from kallithea.lib.vcs.exceptions import RepositoryError, ChangesetDoesNotExistError, \
     ChangesetError, NodeDoesNotExistError, EmptyRepositoryError
 from kallithea.lib.utils2 import safe_int, safe_str
 log = logging.getLogger(__name__)
 def _load_changelog_summary():
     # also used from summary ...
     p = safe_int(request.GET.get('page'), 1)
     size = safe_int(request.GET.get('size'), 10)
     def url_generator(**kw):
         return url('changelog_summary_home',
                    repo_name=c.db_repo.repo_name, size=size, **kw)
     collection = c.db_repo_scm_instance
     c.repo_changesets = RepoPage(collection, page=p,
                                  items_per_page=size,
                                  url=url_generator)
     page_revisions = [x.raw_id for x in list(c.repo_changesets)]
     c.comments = c.db_repo.get_comments(page_revisions)
     c.statuses = c.db_repo.statuses(page_revisions)
 class ChangelogController(BaseRepoController):
     def __before__(self):
         super(ChangelogController, self).__before__()
         c.affected_files_cut_off = 60
     @staticmethod
     def __get_cs(rev, repo):
         """
         Safe way to get changeset. If error occur fail with error message.
         :param rev: revision to fetch
         :param repo: repo instance
         """
         try:
             return c.db_repo_scm_instance.get_changeset(rev)
         except EmptyRepositoryError as e:
             h.flash(h.literal(_('There are no changesets yet')),
                     category='error')
         except RepositoryError as e:
             log.error(traceback.format_exc())
             h.flash(safe_str(e), category='error')
         raise HTTPBadRequest()
     @LoginRequired()
     @HasRepoPermissionLevelDecorator('read')
     def index(self, repo_name, revision=None, f_path=None):
         # Fix URL after page size form submission via GET
         # TODO: Somehow just don't send this extra junk in the GET URL
         if request.GET.get('set'):
             request.GET.pop('set', None)
             if revision is None:
                 raise HTTPFound(location=url('changelog_home', repo_name=repo_name, **request.GET))
             raise HTTPFound(location=url('changelog_file_home', repo_name=repo_name, revision=revision, f_path=f_path, **request.GET))
         limit = 2000
         default = 100
         if request.GET.get('size'):
             c.size = max(min(safe_int(request.GET.get('size')), limit), 1)
             session['changelog_size'] = c.size
             session.save()
         else:
             c.size = int(session.get('changelog_size', default))
         # min size must be 1
         c.size = max(c.size, 1)
         p = safe_int(request.GET.get('page'), 1)
         branch_name = request.GET.get('branch', None)
         if (branch_name and
             branch_name not in c.db_repo_scm_instance.branches and
             branch_name not in c.db_repo_scm_instance.closed_branches and
             not revision):
             raise HTTPFound(location=url('changelog_file_home', repo_name=c.repo_name,
                                     revision=branch_name, f_path=f_path or ''))
         if revision == 'tip':
             revision = None
         c.changelog_for_path = f_path
         try:
             if f_path:
                 log.debug('generating changelog for path %s', f_path)
                 # get the history for the file !
                 tip_cs = c.db_repo_scm_instance.get_changeset()
                 try:
                     collection = tip_cs.get_file_history(f_path)
                 except (NodeDoesNotExistError, ChangesetError):
                     #this node is not present at tip !
                     try:
                         cs = self.__get_cs(revision, repo_name)
                         collection = cs.get_file_history(f_path)
                     except RepositoryError as e:
                         h.flash(safe_str(e), category='warning')
                         raise HTTPFound(location=h.url('changelog_home', repo_name=repo_name))
                 collection = list(reversed(collection))
             else:
                 collection = c.db_repo_scm_instance.get_changesets(start=0, end=revision,
                                                         branch_name=branch_name)
             c.total_cs = len(collection)
             c.pagination = RepoPage(collection, page=p, item_count=c.total_cs,
                                     items_per_page=c.size, branch=branch_name,)
             page_revisions = [x.raw_id for x in c.pagination]
             c.comments = c.db_repo.get_comments(page_revisions)
             c.statuses = c.db_repo.statuses(page_revisions)
         except EmptyRepositoryError as e:
             h.flash(safe_str(e), category='warning')
             raise HTTPFound(location=url('summary_home', repo_name=c.repo_name))
         except (RepositoryError, ChangesetDoesNotExistError, Exception) as e:
             log.error(traceback.format_exc())
             h.flash(safe_str(e), category='error')
             raise HTTPFound(location=url('changelog_home', repo_name=c.repo_name))
         c.branch_name = branch_name
         c.branch_filters = [('', _('None'))] + \
             [(k, k) for k in c.db_repo_scm_instance.branches.keys()]
         if c.db_repo_scm_instance.closed_branches:
             prefix = _('(closed)') + ' '
             c.branch_filters += [('-', '-')] + \
                 [(k, prefix + k) for k in c.db_repo_scm_instance.closed_branches.keys()]
         revs = []
         if not f_path:
             revs = [x.revision for x in c.pagination]
         c.jsdata = graph_data(c.db_repo_scm_instance, revs)
         c.revision = revision # requested revision ref
         c.first_revision = c.pagination[0] # pagination is never empty here!
         return render('changelog/changelog.html')
     @LoginRequired()
     @HasRepoPermissionLevelDecorator('read')
     def changelog_details(self, cs):
         if request.environ.get('HTTP_X_PARTIAL_XHR'):
             c.cs = c.db_repo_scm_instance.get_changeset(cs)
             return render('changelog/changelog_details.html')
         raise HTTPNotFound()
     @LoginRequired()
     @HasRepoPermissionLevelDecorator('read')
     def changelog_summary(self, repo_name):
         if request.environ.get('HTTP_X_PARTIAL_XHR'):
             _load_changelog_summary()
             return render('changelog/changelog_summary_data.html')
         raise HTTPNotFound()

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

0 comments (0 inline, 0 general)