kallithea Changeset - 62822054a1e2

Changeset - 62822054a1e2

Parent rev.

Child rev.

[Not reviewed]

default

0 4 0

Mads Kiilerich - 6 years ago 2019-08-11 14:14:48
mads@kiilerich.com

Grafted from: 36e7c9d416d5

admin: drop dead code for system update notifications

If we want it, we probably want to do it in a different way.

4 files changed with 1 insertions and 57 deletions:

kallithea/config/routing.py

kallithea/controllers/admin/settings.py

kallithea/lib/db_manage.py

kallithea/templates/admin/settings/settings_system.html

0 comments (0 inline, 0 general)

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 routes import Mapper
 from tg import request
 # 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['paths']['controllers'],
                   always_scan=config['debug'])
     rmap.minimization = False
     rmap.explicit = False
     from kallithea.lib.utils import is_valid_repo, is_valid_repo_group
     def check_repo(environ, match_dict):
         """
         Check for valid repository for proper 404 handling.
         Also, a bit of side effect modifying match_dict ...
         """
         if match_dict.get('f_path'):
             # fix for multiple initial slashes that causes errors
             match_dict['f_path'] = match_dict['f_path'].lstrip('/')
         return is_valid_repo(match_dict['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()
     #==========================================================================
     # CUSTOM ROUTES HERE
     #==========================================================================
     # MAIN PAGE
     rmap.connect('home', '/', controller='home', action='index')
     rmap.connect('about', '/about', controller='home', action='about')
     rmap.redirect('/favicon.ico', '/images/favicon.ico')
     rmap.connect('repo_switcher_data', '/_repos', controller='home',
                  action='repo_switcher_data')
     rmap.connect('users_and_groups_data', '/_users_and_groups', controller='home',
                  action='users_and_groups_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_ssh_keys", "/users/{id}/edit/ssh_keys",
                   action="edit_ssh_keys", conditions=dict(method=["GET"]))
         m.connect("edit_user_ssh_keys", "/users/{id}/edit/ssh_keys",
                   action="ssh_keys_add", conditions=dict(method=["POST"]))
         m.connect("edit_user_ssh_keys_delete", "/users/{id}/edit/ssh_keys/delete",
                   action="ssh_keys_delete", 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"]))
         m.connect("my_account_ssh_keys", "/my_account/ssh_keys",
                   action="my_account_ssh_keys", conditions=dict(method=["GET"]))
         m.connect("my_account_ssh_keys", "/my_account/ssh_keys",
                   action="my_account_ssh_keys_add", conditions=dict(method=["POST"]))
         m.connect("my_account_ssh_keys_delete", "/my_account/ssh_keys/delete",
                   action="my_account_ssh_keys_delete", conditions=dict(method=["POST"]))
     # ADMIN GIST
     with rmap.submapper(path_prefix=ADMIN_PREFIX,
                         controller='admin/gists') as m:
         m.connect("gists", "/gists",
                   action="create", conditions=dict(method=["POST"]))
         m.connect("gists", "/gists",
                   action="index", conditions=dict(method=["GET"]))
         m.connect("new_gist", "/gists/new",
                   action="new", conditions=dict(method=["GET"]))
         m.connect("gist_delete", "/gists/{gist_id}/delete",
                   action="delete", conditions=dict(method=["POST"]))
         m.connect("edit_gist", "/gists/{gist_id}/edit",
                   action="edit", conditions=dict(method=["GET", "POST"]))
         m.connect("edit_gist_check_revision", "/gists/{gist_id}/edit/check_revision",
                   action="check_revision", conditions=dict(method=["POST"]))
         m.connect("gist", "/gists/{gist_id}",
                   action="show", conditions=dict(method=["GET"]))
         m.connect("gist_rev", "/gists/{gist_id}/{revision}",
                   revision="tip",
                   action="show", conditions=dict(method=["GET"]))
         m.connect("formatted_gist", "/gists/{gist_id}/{revision}/{format}",
                   revision="tip",
                   action="show", conditions=dict(method=["GET"]))
         m.connect("formatted_gist_file", "/gists/{gist_id}/{revision}/{format}/{f_path:.*}",
                   revision='tip',
                   action="show", conditions=dict(method=["GET"]))
     # ADMIN MAIN PAGES
     with rmap.submapper(path_prefix=ADMIN_PREFIX,
                         controller='admin/admin') as m:
         m.connect('admin_home', '', action='index')
         m.connect('admin_add_repo', '/add_repo/{new_repo:[a-z0-9. _-]*}',
                   action='add_repo')
     #==========================================================================
     # API V2
     #==========================================================================
     with rmap.submapper(path_prefix=ADMIN_PREFIX, controller='api/api',
                         action='_dispatch') as m:
         m.connect('api', '/api')
     # USER JOURNAL
     rmap.connect('journal', '%s/journal' % ADMIN_PREFIX,
                  controller='journal', action='index')
     rmap.connect('journal_rss', '%s/journal/rss' % ADMIN_PREFIX,
                  controller='journal', action='journal_rss')
     rmap.connect('journal_atom', '%s/journal/atom' % ADMIN_PREFIX,
                  controller='journal', action='journal_atom')
     rmap.connect('public_journal', '%s/public_journal' % ADMIN_PREFIX,
                  controller='journal', action="public_journal")
     rmap.connect('public_journal_rss', '%s/public_journal/rss' % ADMIN_PREFIX,
                  controller='journal', action="public_journal_rss")
     rmap.connect('public_journal_rss_old', '%s/public_journal_rss' % ADMIN_PREFIX,
                  controller='journal', action="public_journal_rss")
     rmap.connect('public_journal_atom',
                  '%s/public_journal/atom' % ADMIN_PREFIX, controller='journal',
                  action="public_journal_atom")
     rmap.connect('public_journal_atom_old',
                  '%s/public_journal_atom' % ADMIN_PREFIX, controller='journal',
                  action="public_journal_atom")
     rmap.connect('toggle_following', '%s/toggle_following' % ADMIN_PREFIX,
                  controller='journal', action='toggle_following',
                  conditions=dict(method=["POST"]))
     # SEARCH
     rmap.connect('search', '%s/search' % ADMIN_PREFIX, controller='search',)
     rmap.connect('search_repo_admin', '%s/search/{repo_name:.*}' % ADMIN_PREFIX,
                  controller='search',
                  conditions=dict(function=check_repo))
     rmap.connect('search_repo', '/{repo_name:.*?}/search',
                  controller='search',
                  conditions=dict(function=check_repo),
+                 )
     # LOGIN/LOGOUT/REGISTER/SIGN IN
     rmap.connect('session_csrf_secret_token', '%s/session_csrf_secret_token' % ADMIN_PREFIX, controller='login', action='session_csrf_secret_token')
     rmap.connect('login_home', '%s/login' % ADMIN_PREFIX, controller='login')
     rmap.connect('logout_home', '%s/logout' % ADMIN_PREFIX, controller='login',
                  action='logout')
     rmap.connect('register', '%s/register' % ADMIN_PREFIX, controller='login',
                  action='register')
     rmap.connect('reset_password', '%s/password_reset' % ADMIN_PREFIX,
                  controller='login', action='password_reset')
     rmap.connect('reset_password_confirmation',
                  '%s/password_reset_confirmation' % ADMIN_PREFIX,
                  controller='login', action='password_reset_confirmation')
     # FEEDS
     rmap.connect('rss_feed_home', '/{repo_name:.*?}/feed/rss',
                 controller='feed', action='rss',
                 conditions=dict(function=check_repo))
     rmap.connect('atom_feed_home', '/{repo_name:.*?}/feed/atom',
                 controller='feed', action='atom',
                 conditions=dict(function=check_repo))
     #==========================================================================
     # REPOSITORY ROUTES
     #==========================================================================
     rmap.connect('repo_creating_home', '/{repo_name:.*?}/repo_creating',
                 controller='admin/repos', action='repo_creating')
     rmap.connect('repo_check_home', '/{repo_name:.*?}/crepo_check',
                 controller='admin/repos', action='repo_check')
     rmap.connect('summary_home', '/{repo_name:.*?}',
                 controller='summary',
                 conditions=dict(function=check_repo))
     # must be here for proper group/repo catching
     rmap.connect('repos_group_home', '/{group_name:.*}',
                 controller='admin/repo_groups', action="show_by_name",
                 conditions=dict(function=check_group))
     rmap.connect('repo_stats_home', '/{repo_name:.*?}/statistics',
                 controller='summary', action='statistics',
                 conditions=dict(function=check_repo))
     rmap.connect('repo_size', '/{repo_name:.*?}/repo_size',
                 controller='summary', action='repo_size',
                 conditions=dict(function=check_repo))
     rmap.connect('repo_refs_data', '/{repo_name:.*?}/refs-data',
                  controller='home', action='repo_refs_data')
     rmap.connect('changeset_home', '/{repo_name:.*?}/changeset/{revision:.*}',
                 controller='changeset', revision='tip',
                 conditions=dict(function=check_repo))
     rmap.connect('changeset_children', '/{repo_name:.*?}/changeset_children/{revision}',
                 controller='changeset', revision='tip', action="changeset_children",
                 conditions=dict(function=check_repo))
     rmap.connect('changeset_parents', '/{repo_name:.*?}/changeset_parents/{revision}',
                 controller='changeset', revision='tip', action="changeset_parents",
                 conditions=dict(function=check_repo))
     # repo edit options
     rmap.connect("edit_repo", "/{repo_name:.*?}/settings",
                  controller='admin/repos', action="edit",
                  conditions=dict(method=["GET"], function=check_repo))
     rmap.connect("edit_repo_perms", "/{repo_name:.*?}/settings/permissions",
                  controller='admin/repos', action="edit_permissions",
                  conditions=dict(method=["GET"], function=check_repo))
     rmap.connect("edit_repo_perms_update", "/{repo_name:.*?}/settings/permissions",
                  controller='admin/repos', action="edit_permissions_update",
                  conditions=dict(method=["POST"], function=check_repo))
     rmap.connect("edit_repo_perms_revoke", "/{repo_name:.*?}/settings/permissions/delete",
                  controller='admin/repos', action="edit_permissions_revoke",
                  conditions=dict(method=["POST"], function=check_repo))
     rmap.connect("edit_repo_fields", "/{repo_name:.*?}/settings/fields",
                  controller='admin/repos', action="edit_fields",
                  conditions=dict(method=["GET"], function=check_repo))
     rmap.connect('create_repo_fields', "/{repo_name:.*?}/settings/fields/new",
                  controller='admin/repos', action="create_repo_field",
                  conditions=dict(method=["POST"], function=check_repo))
     rmap.connect('delete_repo_fields', "/{repo_name:.*?}/settings/fields/{field_id}/delete",
                  controller='admin/repos', action="delete_repo_field",
                  conditions=dict(method=["POST"], function=check_repo))
     rmap.connect("edit_repo_advanced", "/{repo_name:.*?}/settings/advanced",
                  controller='admin/repos', action="edit_advanced",
                  conditions=dict(method=["GET"], function=check_repo))
     rmap.connect("edit_repo_advanced_journal", "/{repo_name:.*?}/settings/advanced/journal",
                  controller='admin/repos', action="edit_advanced_journal",
                  conditions=dict(method=["POST"], function=check_repo))
     rmap.connect("edit_repo_advanced_fork", "/{repo_name:.*?}/settings/advanced/fork",
                  controller='admin/repos', action="edit_advanced_fork",
                  conditions=dict(method=["POST"], function=check_repo))
     rmap.connect("edit_repo_caches", "/{repo_name:.*?}/settings/caches",
                  controller='admin/repos', action="edit_caches",
                  conditions=dict(method=["GET"], function=check_repo))
     rmap.connect("update_repo_caches", "/{repo_name:.*?}/settings/caches",
                  controller='admin/repos', action="edit_caches",
                  conditions=dict(method=["POST"], function=check_repo))
     rmap.connect("edit_repo_remote", "/{repo_name:.*?}/settings/remote",
                  controller='admin/repos', action="edit_remote",
                  conditions=dict(method=["GET"], function=check_repo))
     rmap.connect("edit_repo_remote_update", "/{repo_name:.*?}/settings/remote",
                  controller='admin/repos', action="edit_remote",
                  conditions=dict(method=["POST"], function=check_repo))
     rmap.connect("edit_repo_statistics", "/{repo_name:.*?}/settings/statistics",
                  controller='admin/repos', action="edit_statistics",
                  conditions=dict(method=["GET"], function=check_repo))
     rmap.connect("edit_repo_statistics_update", "/{repo_name:.*?}/settings/statistics",
                  controller='admin/repos', action="edit_statistics",
                  conditions=dict(method=["POST"], function=check_repo))
     # still working url for backward compat.
     rmap.connect('raw_changeset_home_depraced',
                  '/{repo_name:.*?}/raw-changeset/{revision}',
                  controller='changeset', action='changeset_raw',
                  revision='tip', conditions=dict(function=check_repo))
     ## new URLs
     rmap.connect('changeset_raw_home',
                  '/{repo_name:.*?}/changeset-diff/{revision}',
                  controller='changeset', action='changeset_raw',
                  revision='tip', conditions=dict(function=check_repo))
     rmap.connect('changeset_patch_home',
                  '/{repo_name:.*?}/changeset-patch/{revision}',
                  controller='changeset', action='changeset_patch',
                  revision='tip', conditions=dict(function=check_repo))
     rmap.connect('changeset_download_home',
                  '/{repo_name:.*?}/changeset-download/{revision}',
                  controller='changeset', action='changeset_download',
                  revision='tip', conditions=dict(function=check_repo))
     rmap.connect('changeset_comment',
                  '/{repo_name:.*?}/changeset-comment/{revision}',
                 controller='changeset', revision='tip', action='comment',
                 conditions=dict(function=check_repo))
     rmap.connect('changeset_comment_delete',
                  '/{repo_name:.*?}/changeset-comment/{comment_id}/delete',
                 controller='changeset', action='delete_comment',
                 conditions=dict(function=check_repo, method=["POST"]))
     rmap.connect('changeset_info', '/changeset_info/{repo_name:.*?}/{revision}',
                  controller='changeset', action='changeset_info')
     rmap.connect('compare_home',
                  '/{repo_name:.*?}/compare',
                  controller='compare', action='index',
                  conditions=dict(function=check_repo))
     rmap.connect('compare_url',
                  '/{repo_name:.*?}/compare/{org_ref_type}@{org_ref_name:.*?}...{other_ref_type}@{other_ref_name:.*?}',
                  controller='compare', action='compare',
                  conditions=dict(function=check_repo),
                  requirements=dict(
                             org_ref_type='(branch|book|tag|rev|__other_ref_type__)',
                             other_ref_type='(branch|book|tag|rev|__org_ref_type__)')
+                 )
     rmap.connect('pullrequest_home',
                  '/{repo_name:.*?}/pull-request/new', controller='pullrequests',
                  action='index', conditions=dict(function=check_repo,
                                                  method=["GET"]))
     rmap.connect('pullrequest_repo_info',
                  '/{repo_name:.*?}/pull-request-repo-info',
                  controller='pullrequests', action='repo_info',
                  conditions=dict(function=check_repo, method=["GET"]))
     rmap.connect('pullrequest',
                  '/{repo_name:.*?}/pull-request/new', controller='pullrequests',
                  action='create', conditions=dict(function=check_repo,
                                                   method=["POST"]))
     rmap.connect('pullrequest_show',
                  '/{repo_name:.*?}/pull-request/{pull_request_id:\\d+}{extra:(/.*)?}', extra='',
                  controller='pullrequests',
                  action='show', conditions=dict(function=check_repo,
                                                 method=["GET"]))
     rmap.connect('pullrequest_post',
                  '/{repo_name:.*?}/pull-request/{pull_request_id}',
                  controller='pullrequests',
                  action='post', conditions=dict(function=check_repo,
                                                 method=["POST"]))
     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_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))

kallithea/controllers/admin/settings.py

➞

Show inline comments

@@ @@ -41,437 +41,393 @@ from kallithea.lib.auth import HasPermis @@
 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.lib.utils2 import safe_unicode
 from kallithea.lib.vcs import VCSError
 from kallithea.model.db import Repository, Setting, Ui
 from kallithea.model.forms import ApplicationSettingsForm, ApplicationUiSettingsForm, ApplicationVisualisationForm
 from kallithea.model.meta import Session
 from kallithea.model.notification import EmailNotificationModel
 from kallithea.model.scm import ScmModel
 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(allow_default_user=True)
     def _before(self, *args, **kwargs):
         super(SettingsController, self)._before(*args, **kwargs)
     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']
                 ## 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)
             added_msg = h.HTML(', ').join(
                 h.link_to(safe_unicode(repo_name), h.url('summary_home', repo_name=repo_name)) for repo_name in added
             ) or '-'
             removed_msg = h.HTML(', ').join(
                 safe_unicode(repo_name) for repo_name in removed
             ) or '-'
             h.flash(h.HTML(_('Repositories successfully rescanned. Added: %s. Removed: %s.')) %
                     (added_msg, removed_msg), 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_metalabels', 'stylify_metalabels', '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'),
                     ('clone_ssh_tmpl', 'clone_ssh_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_key in (x.ui_key for x in Ui.get_custom_hooks()):
                         h.flash(_('Hook already exists'), category='error')
                     elif ui_key in (x.ui_key for x in Ui.get_builtin_hooks()):
                         h.flash(_('Builtin hooks are read-only. Please use another hook name.'), category='error')
                     elif 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, ov in zip(_d.get('hook_ui_key', []),
                                         _d.get('hook_ui_value_new', []),
                                         _d.get('hook_ui_value', [])):
                         if v != ov:
                             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/lib/db_manage.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.lib.db_manage
 ~~~~~~~~~~~~~~~~~~~~~~~
 Database creation, and setup module for Kallithea. Used for creation
 of database as well as for migration operations
 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 10, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import os
 import sys
 import uuid
 import alembic.command
 import alembic.config
 import sqlalchemy
 from sqlalchemy.engine import create_engine
 from kallithea.model.base import init_model
 from kallithea.model.db import Permission, RepoGroup, Repository, Setting, Ui, User, UserRepoGroupToPerm, UserToPerm
 #from kallithea.model import meta
 from kallithea.model.meta import Base, Session
 from kallithea.model.permission import PermissionModel
 from kallithea.model.repo_group import RepoGroupModel
 from kallithea.model.user import UserModel
 log = logging.getLogger(__name__)
 class DbManage(object):
     def __init__(self, dbconf, root, tests=False, SESSION=None, cli_args=None):
         self.dbname = dbconf.split('/')[-1]
         self.tests = tests
         self.root = root
         self.dburi = dbconf
         self.db_exists = False
         self.cli_args = cli_args or {}
         self.init_db(SESSION=SESSION)
     def _ask_ok(self, msg):
         """Invoke ask_ok unless the force_ask option provides the answer"""
         force_ask = self.cli_args.get('force_ask')
         if force_ask is not None:
             return force_ask
         from kallithea.lib.utils2 import ask_ok
         return ask_ok(msg)
     def init_db(self, SESSION=None):
         if SESSION:
             self.sa = SESSION
         else:
             # init new sessions
             engine = create_engine(self.dburi)
             init_model(engine)
             self.sa = Session()
     def create_tables(self, override=False):
         """
         Create a auth database
         """
         log.info("Any existing database is going to be destroyed")
         if self.tests:
             destroy = True
         else:
             destroy = self._ask_ok('Are you sure to destroy old database ? [y/n]')
         if not destroy:
             print 'Nothing done.'
             sys.exit(0)
         if destroy:
             # drop and re-create old schemas
             url = sqlalchemy.engine.url.make_url(self.dburi)
             database = url.database
             # Some databases enforce foreign key constraints and Base.metadata.drop_all() doesn't work
             if url.drivername == 'mysql':
                 url.database = None  # don't connect to the database (it might not exist)
                 engine = sqlalchemy.create_engine(url)
                 with engine.connect() as conn:
                     conn.execute('DROP DATABASE IF EXISTS ' + database)
                     conn.execute('CREATE DATABASE ' + database)
             elif url.drivername == 'postgresql':
                 from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
                 url.database = 'postgres'  # connect to the system database (as the real one might not exist)
                 engine = sqlalchemy.create_engine(url)
                 with engine.connect() as conn:
                     conn.connection.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
                     conn.execute('DROP DATABASE IF EXISTS ' + database)
                     conn.execute('CREATE DATABASE ' + database)
             else:
                 # known to work on SQLite - possibly not on other databases with strong referential integrity
                 Base.metadata.drop_all()
         checkfirst = not override
         Base.metadata.create_all(checkfirst=checkfirst)
         # Create an Alembic configuration and generate the version table,
         # "stamping" it with the most recent Alembic migration revision, to
         # tell Alembic that all the schema upgrades are already in effect.
         alembic_cfg = alembic.config.Config()
         alembic_cfg.set_main_option('script_location', 'kallithea:alembic')
         alembic_cfg.set_main_option('sqlalchemy.url', self.dburi)
         # This command will give an error in an Alembic multi-head scenario,
         # but in practice, such a scenario should not come up during database
         # creation, even during development.
         alembic.command.stamp(alembic_cfg, 'head')
         log.info('Created tables for %s', self.dbname)
     def fix_repo_paths(self):
         """
         Fixes a old kallithea version path into new one without a '*'
         """
         paths = Ui.query() \
                 .filter(Ui.ui_key == '/') \
                 .scalar()
         paths.ui_value = paths.ui_value.replace('*', '')
         self.sa.commit()
     def fix_default_user(self):
         """
         Fixes a old default user with some 'nicer' default values,
         used mostly for anonymous access
         """
         def_user = User.query().filter_by(is_default_user=True).one()
         def_user.name = 'Anonymous'
         def_user.lastname = 'User'
         def_user.email = 'anonymous@kallithea-scm.org'
         self.sa.commit()
     def fix_settings(self):
         """
         Fixes kallithea settings adds ga_code key for google analytics
         """
         hgsettings3 = Setting('ga_code', '')
         self.sa.add(hgsettings3)
         self.sa.commit()
     def admin_prompt(self, second=False):
         if not self.tests:
             import getpass
             username = self.cli_args.get('username')
             password = self.cli_args.get('password')
             email = self.cli_args.get('email')
             def get_password():
                 password = getpass.getpass('Specify admin password '
                                            '(min 6 chars):')
                 confirm = getpass.getpass('Confirm password:')
                 if password != confirm:
                     log.error('passwords mismatch')
                     return False
                 if len(password) < 6:
                     log.error('password is to short use at least 6 characters')
                     return False
                 return password
             if username is None:
                 username = raw_input('Specify admin username:')
             if password is None:
                 password = get_password()
                 if not password:
                     # second try
                     password = get_password()
                     if not password:
                         sys.exit()
             if email is None:
                 email = raw_input('Specify admin email:')
             self.create_user(username, password, email, True)
         else:
             log.info('creating admin and regular test users')
             from kallithea.tests.base import TEST_USER_ADMIN_LOGIN, \
                 TEST_USER_ADMIN_PASS, TEST_USER_ADMIN_EMAIL, \
                 TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS, \
                 TEST_USER_REGULAR_EMAIL, TEST_USER_REGULAR2_LOGIN, \
                 TEST_USER_REGULAR2_PASS, TEST_USER_REGULAR2_EMAIL
             self.create_user(TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS,
                              TEST_USER_ADMIN_EMAIL, True)
             self.create_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS,
                              TEST_USER_REGULAR_EMAIL, False)
             self.create_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS,
                              TEST_USER_REGULAR2_EMAIL, False)
     def create_auth_plugin_options(self, skip_existing=False):
         """
         Create default auth plugin settings, and make it active
         :param skip_existing:
         """
         for k, v, t in [('auth_plugins', 'kallithea.lib.auth_modules.auth_internal', 'list'),
                         ('auth_internal_enabled', 'True', 'bool')]:
             if skip_existing and Setting.get_by_name(k) is not None:
                 log.debug('Skipping option %s', k)
                 continue
             setting = Setting(k, v, t)
             self.sa.add(setting)
     def create_default_options(self, skip_existing=False):
         """Creates default settings"""
         for k, v, t in [
             ('default_repo_enable_downloads', False, 'bool'),
             ('default_repo_enable_statistics', False, 'bool'),
             ('default_repo_private', False, 'bool'),
             ('default_repo_type', 'hg', 'unicode')
         ]:
             if skip_existing and Setting.get_by_name(k) is not None:
                 log.debug('Skipping option %s', k)
                 continue
             setting = Setting(k, v, t)
             self.sa.add(setting)
     def fixup_groups(self):
         def_usr = User.get_default_user()
         for g in RepoGroup.query().all():
             g.group_name = g.get_new_name(g.name)
             # get default perm
             default = UserRepoGroupToPerm.query() \
                 .filter(UserRepoGroupToPerm.group == g) \
                 .filter(UserRepoGroupToPerm.user == def_usr) \
                 .scalar()
             if default is None:
                 log.debug('missing default permission for group %s adding', g)
                 RepoGroupModel()._create_default_perms(g)
     def reset_permissions(self, username):
         """
         Resets permissions to default state, useful when old systems had
         bad permissions, we must clean them up
         :param username:
         """
         default_user = User.get_by_username(username)
         if not default_user:
             return
         u2p = UserToPerm.query() \
             .filter(UserToPerm.user == default_user).all()
         fixed = False
         if len(u2p) != len(Permission.DEFAULT_USER_PERMISSIONS):
             for p in u2p:
                 Session().delete(p)
             fixed = True
             self.populate_default_permissions()
         return fixed
     def update_repo_info(self):
         for repo in Repository.query():
             repo.update_changeset_cache()
     def prompt_repo_root_path(self, test_repo_path='', retries=3):
         _path = self.cli_args.get('repos_location')
         if retries == 3:
             log.info('Setting up repositories config')
         if _path is not None:
             path = _path
         elif not self.tests and not test_repo_path:
             path = raw_input(
                  'Enter a valid absolute path to store repositories. '
                  'All repositories in that path will be added automatically:'
+            )
         else:
             path = test_repo_path
         path_ok = True
         # check proper dir
         if not os.path.isdir(path):
             path_ok = False
             log.error('Given path %s is not a valid directory', path)
         elif not os.path.isabs(path):
             path_ok = False
             log.error('Given path %s is not an absolute path', path)
         # check if path is at least readable.
         if not os.access(path, os.R_OK):
             path_ok = False
             log.error('Given path %s is not readable', path)
         # check write access, warn user about non writeable paths
         elif not os.access(path, os.W_OK) and path_ok:
             log.warning('No write permission to given path %s', path)
             if not self._ask_ok('Given path %s is not writeable, do you want to '
                           'continue with read only mode ? [y/n]' % (path,)):
                 log.error('Canceled by user')
                 sys.exit(-1)
         if retries == 0:
             sys.exit('max retries reached')
         if not path_ok:
             if _path is not None:
                 sys.exit('Invalid repo path: %s' % _path)
             retries -= 1
             return self.prompt_repo_root_path(test_repo_path, retries) # recursing!!!
         real_path = os.path.normpath(os.path.realpath(path))
         if real_path != os.path.normpath(path):
             log.warning('Using normalized path %s instead of %s', real_path, path)
         return real_path
     def create_settings(self, repo_root_path):
         ui_config = [
             ('paths', '/', repo_root_path, True),
             #('phases', 'publish', 'false', False)
             ('hooks', Ui.HOOK_UPDATE, 'hg update >&2', False),
             ('hooks', Ui.HOOK_REPO_SIZE, 'python:kallithea.lib.hooks.repo_size', True),
             ('extensions', 'largefiles', '', True),
             ('largefiles', 'usercache', os.path.join(repo_root_path, '.cache', 'largefiles'), True),
             ('extensions', 'hgsubversion', '', False),
             ('extensions', 'hggit', '', False),
+        ]
         for ui_section, ui_key, ui_value, ui_active in ui_config:
             ui_conf = Ui(
                 ui_section=ui_section,
                 ui_key=ui_key,
                 ui_value=ui_value,
                 ui_active=ui_active)
             self.sa.add(ui_conf)
         settings = [
             ('realm', 'Kallithea', 'unicode'),
             ('title', '', 'unicode'),
             ('ga_code', '', 'unicode'),
             ('show_public_icon', True, 'bool'),
             ('show_private_icon', True, 'bool'),
             ('stylify_metalabels', False, 'bool'),
             ('dashboard_items', 100, 'int'), # TODO: call it page_size
             ('admin_grid_items', 25, 'int'),
             ('show_version', True, 'bool'),
             ('use_gravatar', True, 'bool'),
             ('gravatar_url', User.DEFAULT_GRAVATAR_URL, 'unicode'),
             ('clone_uri_tmpl', Repository.DEFAULT_CLONE_URI, 'unicode'),
             ('clone_ssh_tmpl', Repository.DEFAULT_CLONE_SSH, 'unicode'),
             ('update_url', Setting.DEFAULT_UPDATE_URL, 'unicode'),
+        ]
         for key, val, type_ in settings:
             sett = Setting(key, val, type_)
             self.sa.add(sett)
         self.create_auth_plugin_options()
         self.create_default_options()
         log.info('Populated Ui and Settings defaults')
     def create_user(self, username, password, email='', admin=False):
         log.info('creating user %s', username)
         UserModel().create_or_update(username, password, email,
                                      firstname=u'Kallithea', lastname=u'Admin',
                                      active=True, admin=admin,
                                      extern_type=User.DEFAULT_AUTH_TYPE)
     def create_default_user(self):
         log.info('creating default user')
         # create default user for handling default permissions.
         user = UserModel().create_or_update(username=User.DEFAULT_USER,
                                             password=str(uuid.uuid1())[:20],
                                             email='anonymous@kallithea-scm.org',
                                             firstname=u'Anonymous',
                                             lastname=u'User')
         # based on configuration options activate/deactivate this user which
         # controls anonymous access
         if self.cli_args.get('public_access') is False:
             log.info('Public access disabled')
             user.active = False
             Session().commit()
     def create_permissions(self):
         """
         Creates all permissions defined in the system
         """
         # module.(access|create|change|delete)_[name]
         # module.(none|read|write|admin)
         log.info('creating permissions')
         PermissionModel().create_permissions()
     def populate_default_permissions(self):
         """
         Populate default permissions. It will create only the default
         permissions that are missing, and not alter already defined ones
         """
         log.info('creating default user permissions')
         PermissionModel().create_default_permissions(user=User.DEFAULT_USER)

kallithea/templates/admin/settings/settings_system.html

➞

Show inline comments

 <div id="update_notice" style="display: none">
     <div>${_('Checking for updates...')}</div>
 </div>
 <%
  elems = [
-    (_('Kallithea version'), h.literal('%s <b><span id="check_for_update" style="display:none">%s</span></b>') % (c.kallithea_version, _('Check for updates')), ''),
     (_('Kallithea version'), c.kallithea_version, ''),
     (_('Kallithea configuration file'), c.ini['__file__'], ''),
     (_('Python version'), c.py_version, ''),
     (_('Platform'), c.platform, ''),
     (_('Git version'), c.git_version, ''),
     (_('Git path'), c.ini.get('git_path'), ''),
     (_('Upgrade info endpoint'), h.literal('%s <br/><span class="text-muted">%s.</span>') % (c.update_url, _('Note: please make sure this server can access this URL')), ''),
+ ]
 %>
 <dl class="dl-horizontal">
 %for dt, dd, tt in elems:
   <dt title="${dt}">${dt}:</dt>
   <dd title="${tt}">${dd}</dd>
 %endfor
 </dl>
 <h4>${_('Python Packages')}</h4>
 <table class="table">
   <tbody>
       %for key, value in c.modules:
           <tr>
               <td>${key}</td>
               <td>${value}</td>
           </tr>
       %endfor
   </tbody>
 </table>
 <script>
     $('#check_for_update').click(function(e){
         var $update_notice = $('#update_notice');
         $update_notice.show();
         asynchtml(${h.js(h.url('admin_settings_system_update'))}, $update_notice);
     });
 </script>

0 comments (0 inline, 0 general)