Changeset - 116151b6bfb2
[Not reviewed]
default
0 8 0
Mads Kiilerich (mads) - 5 years ago 2020-12-30 00:14:57
mads@kiilerich.com
Grafted from: 84df601909c0
celery: drop tracking of task_id - we use ignore_result=True and will never get anything back

There is thus no need for configuration of celery.result_backend .

The alternative would be to fix it. That could give better error reporting from
failing repo creations, but would require quite a bit of additional changes
before it actually works reliably.
8 files changed with 3 insertions and 36 deletions:
0 comments (0 inline, 0 general)
development.ini
Show inline comments
 
###################################################################################
 
###################################################################################
 
## Kallithea config file generated with kallithea-cli                            ##
 
##                                                                               ##
 
## The %(here)s variable will generally be replaced with the parent directory of ##
 
## this file. Other use of % must be escaped as %% .                             ##
 
###################################################################################
 
###################################################################################
 

	
 
[DEFAULT]
 

	
 
################################################################################
 
## Email settings                                                             ##
 
##                                                                            ##
 
## Refer to the documentation ("Email settings") for more details.            ##
 
##                                                                            ##
 
## It is recommended to use a valid sender address that passes access         ##
 
## validation and spam filtering in mail servers.                             ##
 
################################################################################
 

	
 
## 'From' header for application emails. You can optionally add a name.
 
## Default:
 
#app_email_from = Kallithea
 
## Examples:
 
#app_email_from = Kallithea <kallithea-noreply@example.com>
 
#app_email_from = kallithea-noreply@example.com
 

	
 
## Subject prefix for application emails.
 
## A space between this prefix and the real subject is automatically added.
 
## Default:
 
#email_prefix =
 
## Example:
 
#email_prefix = [Kallithea]
 

	
 
## Recipients for error emails and fallback recipients of application mails.
 
## Multiple addresses can be specified, comma-separated.
 
## Only addresses are allowed, do not add any name part.
 
## Default:
 
#email_to =
 
## Examples:
 
#email_to = admin@example.com
 
#email_to = admin@example.com,another_admin@example.com
 
email_to =
 

	
 
## 'From' header for error emails. You can optionally add a name.
 
## Default: (none)
 
## Examples:
 
#error_email_from = Kallithea Errors <kallithea-noreply@example.com>
 
#error_email_from = kallithea_errors@example.com
 
error_email_from =
 

	
 
## SMTP server settings
 
## If specifying credentials, make sure to use secure connections.
 
## Default: Send unencrypted unauthenticated mails to the specified smtp_server.
 
## For "SSL", use smtp_use_ssl = true and smtp_port = 465.
 
## For "STARTTLS", use smtp_use_tls = true and smtp_port = 587.
 
smtp_server =
 
smtp_username =
 
smtp_password =
 
smtp_port =
 
smtp_use_ssl = false
 
smtp_use_tls = false
 

	
 
## Entry point for 'gearbox serve'
 
[server:main]
 
#host = 127.0.0.1
 
host = 0.0.0.0
 
port = 5000
 

	
 
## Gearbox serve uses the Waitress web server ##
 
use = egg:waitress#main
 
## avoid multi threading
 
threads = 1
 
## allow push of repos bigger than the default of 1 GB
 
max_request_body_size = 107374182400
 
## use poll instead of select, fixes fd limits, may not work on old
 
## windows systems.
 
#asyncore_use_poll = True
 

	
 
## middleware for hosting the WSGI application under a URL prefix
 
#[filter:proxy-prefix]
 
#use = egg:PasteDeploy#prefix
 
#prefix = /<your-prefix>
 

	
 
[app:main]
 
use = egg:kallithea
 
## enable proxy prefix middleware
 
#filter-with = proxy-prefix
 

	
 
full_stack = true
 
static_files = true
 

	
 
## Internationalization (see setup documentation for details)
 
## By default, the languages requested by the browser are used if available, with English as default.
 
## Set i18n.enabled=false to disable automatic language choice.
 
#i18n.enabled = true
 
## To Force a language, set i18n.enabled=false and specify the language in i18n.lang.
 
## Valid values are the names of subdirectories in kallithea/i18n with a LC_MESSAGES/kallithea.mo
 
#i18n.lang = en
 

	
 
cache_dir = %(here)s/data
 
index_dir = %(here)s/data/index
 

	
 
## uncomment and set this path to use archive download cache
 
archive_cache_dir = %(here)s/data/tarballcache
 

	
 
## change this to unique ID for security
 
#app_instance_uuid = VERY-SECRET
 
app_instance_uuid = development-not-secret
 

	
 
## cut off limit for large diffs (size in bytes)
 
cut_off_limit = 256000
 

	
 
## force https in Kallithea, fixes https redirects, assumes it's always https
 
force_https = false
 

	
 
## use Strict-Transport-Security headers
 
use_htsts = false
 

	
 
## number of commits stats will parse on each iteration
 
commit_parse_limit = 25
 

	
 
## Path to Python executable to be used for git hooks.
 
## This value will be written inside the git hook scripts as the text
 
## after '#!' (shebang). When empty or not defined, the value of
 
## 'sys.executable' at the time of installation of the git hooks is
 
## used, which is correct in many cases but for example not when using uwsgi.
 
## If you change this setting, you should reinstall the Git hooks via
 
## Admin > Settings > Remap and Rescan.
 
#git_hook_interpreter = /srv/kallithea/venv/bin/python3
 

	
 
## path to git executable
 
git_path = git
 

	
 
## git rev filter option, --all is the default filter, if you need to
 
## hide all refs in changelog switch this to --branches --tags
 
#git_rev_filter = --branches --tags
 

	
 
## RSS feed options
 
rss_cut_off_limit = 256000
 
rss_items_per_page = 10
 
rss_include_diff = false
 

	
 
## options for showing and identifying changesets
 
show_sha_length = 12
 
show_revision_number = false
 

	
 
## Canonical URL to use when creating full URLs in UI and texts.
 
## Useful when the site is available under different names or protocols.
 
## Defaults to what is provided in the WSGI environment.
 
#canonical_url = https://kallithea.example.com/repos
 

	
 
## gist URL alias, used to create nicer urls for gist. This should be an
 
## url that does rewrites to _admin/gists/<gistid>.
 
## example: http://gist.example.com/{gistid}. Empty means use the internal
 
## Kallithea url, ie. http[s]://kallithea.example.com/_admin/gists/<gistid>
 
gist_alias_url =
 

	
 
## default encoding used to convert from and to unicode
 
## can be also a comma separated list of encoding in case of mixed encodings
 
default_encoding = utf-8
 

	
 
## Set Mercurial encoding, similar to setting HGENCODING before launching Kallithea
 
hgencoding = utf-8
 

	
 
## issue tracker for Kallithea (leave blank to disable, absent for default)
 
#bugtracker = https://bitbucket.org/conservancy/kallithea/issues
 

	
 
## issue tracking mapping for commit messages, comments, PR descriptions, ...
 
## Refer to the documentation ("Integration with issue trackers") for more details.
 

	
 
## regular expression to match issue references
 
## This pattern may/should contain parenthesized groups, that can
 
## be referred to in issue_server_link or issue_sub using Python backreferences
 
## (e.g. \1, \2, ...). You can also create named groups with '(?P<groupname>)'.
 
## To require mandatory whitespace before the issue pattern, use:
 
## (?:^|(?<=\s)) before the actual pattern, and for mandatory whitespace
 
## behind the issue pattern, use (?:$|(?=\s)) after the actual pattern.
 

	
 
issue_pat = #(\d+)
 

	
 
## server url to the issue
 
## This pattern may/should contain backreferences to parenthesized groups in issue_pat.
 
## A backreference can be \1, \2, ... or \g<groupname> if you specified a named group
 
## called 'groupname' in issue_pat.
 
## The special token {repo} is replaced with the full repository name
 
## including repository groups, while {repo_name} is replaced with just
 
## the name of the repository.
 

	
 
issue_server_link = https://issues.example.com/{repo}/issue/\1
 

	
 
## substitution pattern to use as the link text
 
## If issue_sub is empty, the text matched by issue_pat is retained verbatim
 
## for the link text. Otherwise, the link text is that of issue_sub, with any
 
## backreferences to groups in issue_pat replaced.
 

	
 
issue_sub =
 

	
 
## issue_pat, issue_server_link and issue_sub can have suffixes to specify
 
## multiple patterns, to other issues server, wiki or others
 
## below an example how to create a wiki pattern
 
## wiki-some-id -> https://wiki.example.com/some-id
 

	
 
#issue_pat_wiki = wiki-(\S+)
 
#issue_server_link_wiki = https://wiki.example.com/\1
 
#issue_sub_wiki = WIKI-\1
 

	
 
## alternative return HTTP header for failed authentication. Default HTTP
 
## response is 401 HTTPUnauthorized. Currently Mercurial clients have trouble with
 
## handling that. Set this variable to 403 to return HTTPForbidden
 
auth_ret_code =
 

	
 
## allows to change the repository location in settings page
 
allow_repo_location_change = True
 

	
 
## allows to setup custom hooks in settings page
 
allow_custom_hooks_settings = True
 

	
 
## extra extensions for indexing, space separated and without the leading '.'.
 
#index.extensions =
 
#    gemfile
 
#    lock
 

	
 
## extra filenames for indexing, space separated
 
#index.filenames =
 
#    .dockerignore
 
#    .editorconfig
 
#    INSTALL
 
#    CHANGELOG
 

	
 
####################################
 
##            SSH CONFIG          ##
 
####################################
 

	
 
## SSH is disabled by default, until an Administrator decides to enable it.
 
ssh_enabled = false
 

	
 
## File where users' SSH keys will be stored *if* ssh_enabled is true.
 
#ssh_authorized_keys = /home/kallithea/.ssh/authorized_keys
 

	
 
## Path to be used in ssh_authorized_keys file to invoke kallithea-cli with ssh-serve.
 
#kallithea_cli_path = /srv/kallithea/venv/bin/kallithea-cli
 

	
 
## Locale to be used in the ssh-serve command.
 
## This is needed because an SSH client may try to use its own locale
 
## settings, which may not be available on the server.
 
## See `locale -a` for valid values on this system.
 
#ssh_locale = C.UTF-8
 

	
 
####################################
 
##         CELERY CONFIG          ##
 
####################################
 

	
 
## Note: Celery doesn't support Windows.
 
use_celery = false
 

	
 
## Celery config settings from https://docs.celeryproject.org/en/4.4.0/userguide/configuration.html prefixed with 'celery.'.
 

	
 
## Example: use the message queue on the local virtual host 'kallitheavhost' as the RabbitMQ user 'kallithea':
 
celery.broker_url = amqp://kallithea:thepassword@localhost:5672/kallitheavhost
 

	
 
celery.result_backend = db+sqlite:///celery-results.db
 

	
 
#celery.amqp.task.result.expires = 18000
 

	
 
celery.worker_concurrency = 2
 
celery.worker_max_tasks_per_child = 100
 

	
 
####################################
 
##          BEAKER CACHE          ##
 
####################################
 

	
 
beaker.cache.data_dir = %(here)s/data/cache/data
 
beaker.cache.lock_dir = %(here)s/data/cache/lock
 

	
 
beaker.cache.regions = long_term,long_term_file
 

	
 
beaker.cache.long_term.type = memory
 
beaker.cache.long_term.expire = 36000
 
beaker.cache.long_term.key_length = 256
 

	
 
beaker.cache.long_term_file.type = file
 
beaker.cache.long_term_file.expire = 604800
 
beaker.cache.long_term_file.key_length = 256
 

	
 
####################################
 
##        BEAKER SESSION          ##
 
####################################
 

	
 
## Name of session cookie. Should be unique for a given host and path, even when running
 
## on different ports. Otherwise, cookie sessions will be shared and messed up.
 
session.key = kallithea
 
## Sessions should always only be accessible by the browser, not directly by JavaScript.
 
session.httponly = true
 
## Session lifetime. 2592000 seconds is 30 days.
 
session.timeout = 2592000
 

	
 
## Server secret used with HMAC to ensure integrity of cookies.
 
#session.secret = VERY-SECRET
 
session.secret = development-not-secret
 
## Further, encrypt the data with AES.
 
#session.encrypt_key = <key_for_encryption>
 
#session.validate_key = <validation_key>
 

	
 
## Type of storage used for the session, current types are
 
## dbm, file, memcached, database, and memory.
 

	
 
## File system storage of session data. (default)
 
#session.type = file
 

	
 
## Cookie only, store all session data inside the cookie. Requires secure secrets.
 
#session.type = cookie
 

	
 
## Database storage of session data.
 
#session.type = ext:database
 
#session.sa.url = postgresql://postgres:qwe@localhost/kallithea
 
#session.table_name = db_session
 

	
 
####################################
 
##        ERROR HANDLING          ##
 
####################################
 

	
 
## Show a nice error page for application HTTP errors and exceptions (default true)
 
#errorpage.enabled = true
 

	
 
## Enable Backlash client-side interactive debugger (default false)
 
## WARNING: *THIS MUST BE false IN PRODUCTION ENVIRONMENTS!!!*
 
## This debug mode will allow all visitors to execute malicious code.
 
#debug = false
 
debug = true
 

	
 
## Enable Backlash server-side error reporting (unless debug mode handles it client-side) (default true)
 
#trace_errors.enable = true
 
## Errors will be reported by mail if trace_errors.error_email is set.
 

	
 
## Propagate email settings to ErrorReporter of TurboGears2
 
## You do not normally need to change these lines
 
get trace_errors.smtp_server = smtp_server
 
get trace_errors.smtp_port = smtp_port
 
get trace_errors.from_address = error_email_from
 
get trace_errors.error_email = email_to
 
get trace_errors.smtp_username = smtp_username
 
get trace_errors.smtp_password = smtp_password
 
get trace_errors.smtp_use_tls = smtp_use_tls
 

	
 
##################################
 
##        LOGVIEW CONFIG        ##
 
##################################
 

	
 
logview.sqlalchemy = #faa
 
logview.pylons.templating = #bfb
 
logview.pylons.util = #eee
 

	
 
#########################
 
##      DB CONFIG      ##
 
#########################
 

	
 
sqlalchemy.url = sqlite:///%(here)s/kallithea.db?timeout=60
 
#sqlalchemy.url = postgresql://kallithea:password@localhost/kallithea
 
#sqlalchemy.url = mysql://kallithea:password@localhost/kallithea?charset=utf8mb4
 
## Note: the mysql:// prefix should also be used for MariaDB
 

	
 
sqlalchemy.pool_recycle = 3600
 

	
 
################################
 
##   ALEMBIC CONFIGURATION    ##
 
################################
 

	
 
[alembic]
 
script_location = kallithea:alembic
 

	
 
################################
 
##   LOGGING CONFIGURATION    ##
 
################################
 

	
 
[loggers]
 
keys = root, routes, kallithea, sqlalchemy, tg, gearbox, beaker, templates, whoosh_indexer, werkzeug, backlash
 

	
 
[handlers]
 
keys = console, console_color, console_color_sql, null
 

	
 
[formatters]
 
keys = generic, color_formatter, color_formatter_sql
 

	
 
#############
 
## LOGGERS ##
 
#############
 

	
 
[logger_root]
 
level = NOTSET
 
#handlers = console
 
## For coloring based on log level:
 
handlers = console_color
 

	
 
[logger_routes]
 
#level = WARN
 
level = DEBUG
 
handlers =
 
qualname = routes.middleware
 
## "level = DEBUG" logs the route matched and routing variables.
 

	
 
[logger_beaker]
 
#level = WARN
 
level = DEBUG
 
handlers =
 
qualname = beaker.container
 

	
 
[logger_templates]
 
#level = WARN
 
level = INFO
 
handlers =
 
qualname = pylons.templating
 

	
 
[logger_kallithea]
 
#level = WARN
 
level = DEBUG
 
handlers =
 
qualname = kallithea
 

	
 
[logger_tg]
 
#level = WARN
 
level = DEBUG
 
handlers =
 
qualname = tg
 

	
 
[logger_gearbox]
 
#level = WARN
 
level = DEBUG
 
handlers =
 
qualname = gearbox
 

	
 
[logger_sqlalchemy]
 
level = WARN
 
handlers =
 
qualname = sqlalchemy.engine
 
## For coloring based on log level and pretty printing of SQL:
 
#level = INFO
 
#handlers = console_color_sql
 
#propagate = 0
 

	
 
[logger_whoosh_indexer]
 
#level = WARN
 
level = DEBUG
 
handlers =
 
qualname = whoosh_indexer
 

	
 
[logger_werkzeug]
 
level = WARN
 
handlers =
 
qualname = werkzeug
 

	
 
[logger_backlash]
 
level = WARN
 
handlers =
 
qualname = backlash
 

	
 
##############
 
## HANDLERS ##
 
##############
 

	
 
[handler_console]
 
class = StreamHandler
 
args = (sys.stderr,)
 
formatter = generic
 

	
 
[handler_console_color]
 
## ANSI color coding based on log level
 
class = StreamHandler
 
args = (sys.stderr,)
 
formatter = color_formatter
 

	
 
[handler_console_color_sql]
 
## ANSI color coding and pretty printing of SQL statements
 
class = StreamHandler
 
args = (sys.stderr,)
 
formatter = color_formatter_sql
 

	
 
[handler_null]
 
class = NullHandler
 
args = ()
 

	
 
################
 
## FORMATTERS ##
 
################
 

	
 
[formatter_generic]
 
format = %(asctime)s.%(msecs)03d %(levelname)-5.5s [%(name)s] %(message)s
 
datefmt = %Y-%m-%d %H:%M:%S
 

	
 
[formatter_color_formatter]
 
class = kallithea.lib.colored_formatter.ColorFormatter
 
format = %(asctime)s.%(msecs)03d %(levelname)-5.5s [%(name)s] %(message)s
 
datefmt = %Y-%m-%d %H:%M:%S
 

	
 
[formatter_color_formatter_sql]
 
class = kallithea.lib.colored_formatter.ColorFormatterSql
 
format = %(asctime)s.%(msecs)03d %(levelname)-5.5s [%(name)s] %(message)s
 
datefmt = %Y-%m-%d %H:%M:%S
 

	
 
#################
 
## SSH LOGGING ##
 
#################
 

	
 
## The default loggers use 'handler_console' that uses StreamHandler with
 
## destination 'sys.stderr'. In the context of the SSH server process, these log
 
## messages would be sent to the client, which is normally not what you want.
 
## By default, when running ssh-serve, just use NullHandler and disable logging
 
## completely. For other logging options, see:
 
## https://docs.python.org/2/library/logging.handlers.html
 

	
 
[ssh_serve:logger_root]
 
level = CRITICAL
 
handlers = null
 

	
 
## Note: If logging is configured with other handlers, they might need similar
 
## muting for ssh-serve too.
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 tg import request
 
from tg import tmpl_context as c
 
from tg.i18n import ugettext as _
 
from webob.exc import HTTPForbidden, HTTPFound, HTTPInternalServerError, HTTPNotFound
 

	
 
import kallithea
 
from kallithea.controllers import base
 
from kallithea.lib import webutils
 
from kallithea.lib.auth import HasRepoPermissionLevelDecorator, LoginRequired, NotAnonymous
 
from kallithea.lib.exceptions import AttachedForksError
 
from kallithea.lib.utils2 import safe_int
 
from kallithea.lib.vcs import RepositoryError
 
from kallithea.lib.webutils import url
 
from kallithea.model import db, meta, userlog
 
from kallithea.model.forms import RepoFieldForm, RepoForm, RepoPermsForm
 
from kallithea.model.repo import RepoModel
 
from kallithea.model.scm import AvailableRepoGroupChoices, RepoList, ScmModel
 

	
 

	
 
log = logging.getLogger(__name__)
 

	
 

	
 
class ReposController(base.BaseRepoController):
 

	
 
    @LoginRequired(allow_default_user=True)
 
    def _before(self, *args, **kwargs):
 
        super(ReposController, self)._before(*args, **kwargs)
 

	
 
    def _load_repo(self):
 
        repo_obj = c.db_repo
 

	
 
        if repo_obj is None:
 
            raise HTTPNotFound()
 

	
 
        return repo_obj
 

	
 
    def __load_defaults(self, repo=None):
 
        extras = [] if repo is None else [repo.group]
 

	
 
        c.repo_groups = AvailableRepoGroupChoices('write', 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
 
        defaults['permanent_url'] = c.repo_info.clone_url(clone_uri_tmpl=c.clone_uri_tmpl, with_id=True)
 

	
 
        return defaults
 

	
 
    def index(self, format='html'):
 
        repos_list = RepoList(db.Repository.query(sorted=True).all(), perm_level='admin')
 
        # the repo list will be filtered to only show repos where the user has read permissions
 
        repos_data = RepoModel().get_repos_as_dict(repos_list, admin=True)
 
        # data used to render the grid
 
        c.data = repos_data
 

	
 
        return base.render('admin/repos/repos.html')
 

	
 
    @NotAnonymous()
 
    def create(self):
 
        self.__load_defaults()
 
        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))
 
        except formencode.Invalid as errors:
 
            log.info(errors)
 
            return htmlfill.render(
 
                base.render('admin/repos/repo_add.html'),
 
                defaults=errors.value,
 
                errors=errors.error_dict or {},
 
                prefix_error=False,
 
                force_defaults=False,
 
                encoding="UTF-8")
 

	
 
        try:
 
            # 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 Exception:
 
            log.error(traceback.format_exc())
 
            msg = (_('Error creating repository %s')
 
                   % form_result.get('repo_name'))
 
            webutils.flash(msg, category='error')
 
            raise HTTPFound(location=url('home'))
 

	
 
        raise HTTPFound(location=webutils.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 = db.Setting.get_default_repo_settings(strip_prefix=True)
 
        if parent_group:
 
            prg = db.RepoGroup.get(parent_group)
 
            if prg is None or not any(rgc[0] == prg.group_id
 
                                      for rgc in c.repo_groups):
 
                raise HTTPForbidden
 
        else:
 
            parent_group = '-1'
 
        defaults.update({'repo_group': parent_group})
 

	
 
        return htmlfill.render(
 
            base.render('admin/repos/repo_add.html'),
 
            defaults=defaults,
 
            errors={},
 
            prefix_error=False,
 
            encoding="UTF-8",
 
            force_defaults=False)
 

	
 
    @LoginRequired()
 
    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 base.render('admin/repos/repo_creating.html')
 

	
 
    @LoginRequired()
 
    @base.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']:
 
            if kallithea.CELERY_APP:
 
                task_result = kallithea.CELERY_APP.AsyncResult(task_id)
 
                if task_result.failed():
 
                    raise HTTPInternalServerError(task_result.traceback)
 

	
 
        repo = db.Repository.get_by_repo_name(repo_name)
 
        if repo and repo.repo_state == db.Repository.STATE_CREATED:
 
            if repo.clone_uri:
 
                webutils.flash(_('Created repository %s from %s')
 
                        % (repo.repo_name, repo.clone_uri_hidden), category='success')
 
            else:
 
                repo_url = webutils.link_to(repo.repo_name,
 
                                     webutils.url('summary_home',
 
                                           repo_name=repo.repo_name))
 
                fork = repo.fork
 
                if fork is not None:
 
                    fork_name = fork.repo_name
 
                    webutils.flash(webutils.HTML(_('Forked repository %s as %s'))
 
                            % (fork_name, repo_url), category='success')
 
                else:
 
                    webutils.flash(webutils.HTML(_('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 = db.RepositoryField.query() \
 
            .filter(db.RepositoryField.repository == c.repo_info).all()
 

	
 
        repo_model = RepoModel()
 
        changed_name = repo_name
 
        repo = db.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)
 
            webutils.flash(_('Repository %s updated successfully') % repo_name,
 
                    category='success')
 
            changed_name = repo.repo_name
 
            userlog.action_logger(request.authuser, 'admin_updated_repo',
 
                changed_name, request.ip_addr)
 
            meta.Session().commit()
 
        except formencode.Invalid as errors:
 
            log.info(errors)
 
            defaults = self.__load_data()
 
            defaults.update(errors.value)
 
            return htmlfill.render(
 
                base.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())
 
            webutils.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:
 
            raise HTTPNotFound()
 
        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'
 
                    webutils.flash(_('Detached %s forks') % _forks, category='success')
 
                elif do == 'delete_forks':
 
                    handle_forks = 'delete'
 
                    webutils.flash(_('Deleted %s forks') % _forks, category='success')
 
            repo_model.delete(repo, forks=handle_forks)
 
            userlog.action_logger(request.authuser, 'admin_deleted_repo',
 
                repo_name, request.ip_addr)
 
            ScmModel().mark_for_invalidation(repo_name)
 
            webutils.flash(_('Deleted repository %s') % repo_name, category='success')
 
            meta.Session().commit()
 
        except AttachedForksError:
 
            webutils.flash(_('Cannot delete repository %s which still has forks')
 
                        % repo_name, category='warning')
 

	
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            webutils.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 = db.RepositoryField.query() \
 
            .filter(db.RepositoryField.repository == c.repo_info).all()
 
        c.active = 'settings'
 
        return htmlfill.render(
 
            base.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()
 
        c.active = 'permissions'
 
        defaults = RepoModel()._get_defaults(repo_name)
 

	
 
        return htmlfill.render(
 
            base.render('admin/repos/repo_edit.html'),
 
            defaults=defaults,
 
            encoding="UTF-8",
 
            force_defaults=False)
 

	
 
    @HasRepoPermissionLevelDecorator('admin')
 
    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)
 
        meta.Session().commit()
 
        webutils.flash(_('Repository permissions updated'), category='success')
 
        raise HTTPFound(location=url('edit_repo_perms', repo_name=repo_name))
 

	
 
    @HasRepoPermissionLevelDecorator('admin')
 
    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'))
 
            else:
 
                assert False
 

	
 
            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
 
                )
 
            else:
 
                assert False
 
            # TODO: implement this
 
            #action_logger(request.authuser, 'admin_revoked_repo_permissions',
 
            #              repo_name, request.ip_addr)
 
            meta.Session().commit()
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            webutils.flash(_('An error occurred during revoking of permission'),
 
                    category='error')
 
            raise HTTPInternalServerError()
 
        return []
 

	
 
    @HasRepoPermissionLevelDecorator('admin')
 
    def edit_fields(self, repo_name):
 
        c.repo_info = self._load_repo()
 
        c.repo_fields = db.RepositoryField.query() \
 
            .filter(db.RepositoryField.repository == c.repo_info).all()
 
        c.active = 'fields'
 
        if request.POST:
 

	
 
            raise HTTPFound(location=url('repo_edit_fields'))
 
        return base.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 = db.RepositoryField()
 
            new_field.repository = db.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']
 
            meta.Session().add(new_field)
 
            meta.Session().commit()
 
        except formencode.Invalid as e:
 
            webutils.flash(_('Field validation error: %s') % e.msg, category='error')
 
        except Exception as e:
 
            log.error(traceback.format_exc())
 
            webutils.flash(_('An error occurred during creation of field: %r') % e, 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 = db.RepositoryField.get_or_404(field_id)
 
        try:
 
            meta.Session().delete(field)
 
            meta.Session().commit()
 
        except Exception as e:
 
            log.error(traceback.format_exc())
 
            msg = _('An error occurred during removal of field')
 
            webutils.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 = kallithea.DEFAULT_USER_ID
 
        c.in_public_journal = db.UserFollowing.query() \
 
            .filter(db.UserFollowing.user_id == c.default_user_id) \
 
            .filter(db.UserFollowing.follows_repository == c.repo_info).scalar()
 

	
 
        _repos = db.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
 
                         and x.repo_type == c.repo_info.repo_type]
 

	
 
        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(
 
            base.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 = db.Repository.get_by_repo_name(repo_name).repo_id
 
            user_id = kallithea.DEFAULT_USER_ID
 
            self.scm_model.toggle_following_repo(repo_id, user_id)
 
            webutils.flash(_('Updated repository visibility in public journal'),
 
                    category='success')
 
            meta.Session().commit()
 
        except Exception:
 
            webutils.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')
 
            meta.Session().commit()
 
            webutils.flash(_('Marked repository %s as fork of %s') % (repo_name, fork),
 
                    category='success')
 
        except RepositoryError as e:
 
            log.error(traceback.format_exc())
 
            webutils.flash(e, category='error')
 
        except Exception as e:
 
            log.error(traceback.format_exc())
 
            webutils.flash(_('An error occurred during this operation'),
 
                    category='error')
 

	
 
        raise HTTPFound(location=url('edit_repo_advanced', repo_name=repo_name))
 

	
 
    @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, request.ip_addr)
 
                webutils.flash(_('Pulled from remote location'), category='success')
 
            except Exception as e:
 
                log.error(traceback.format_exc())
 
                webutils.flash(_('An error occurred during pull from remote location'),
 
                        category='error')
 
            raise HTTPFound(location=url('edit_repo_remote', repo_name=c.repo_name))
 
        return base.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)
 
                meta.Session().commit()
 
            except Exception as e:
 
                log.error(traceback.format_exc())
 
                webutils.flash(_('An error occurred during deletion of repository stats'),
 
                        category='error')
 
            raise HTTPFound(location=url('edit_repo_statistics', repo_name=c.repo_name))
 

	
 
        return base.render('admin/repos/repo_edit.html')
kallithea/controllers/api/api.py
Show inline comments
 
@@ -501,1718 +501,1714 @@ class ApiController(JSONRPCController):
 
                extern_name=extern_name
 
            )
 
            meta.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=None,
 
                    email=None, password=None,
 
                    firstname=None, lastname=None,
 
                    active=None, admin=None,
 
                    extern_type=None, extern_name=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)
 
            meta.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)
 
            meta.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
 
        read access to user group.
 

	
 
        :param usergroupid: id of user_group to edit
 
        :type usergroupid: str or int
 

	
 
        OUTPUT::
 

	
 
            id : <id_given_in_input>
 
            result : None if group not exist
 
                     {
 
                       "users_group_id" : "<id>",
 
                       "group_name" :     "<groupname>",
 
                       "active":          "<bool>",
 
                       "members" :  [<user_obj>,...]
 
                     }
 
            error : null
 

	
 
        """
 
        user_group = get_user_group_or_error(usergroupid)
 
        if not HasPermissionAny('hg.admin')():
 
            if not HasUserGroupPermissionLevel('read')(user_group.users_group_name):
 
                raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
 

	
 
        data = user_group.get_api_data()
 
        return data
 

	
 
    # permission check inside
 
    def get_user_groups(self):
 
        """
 
        Lists all existing user groups. This command can be executed only using
 
        api_key belonging to user with admin rights or user who has at least
 
        read access to user group.
 

	
 

	
 
        OUTPUT::
 

	
 
            id : <id_given_in_input>
 
            result : [<user_group_obj>,...]
 
            error : null
 
        """
 

	
 
        return [
 
            user_group.get_api_data()
 
            for user_group in UserGroupList(db.UserGroup.query().all(), perm_level='read')
 
        ]
 

	
 
    @HasPermissionAnyDecorator('hg.admin', 'hg.usergroup.create.true')
 
    def create_user_group(self, group_name, description='',
 
                          owner=None, active=True):
 
        """
 
        Creates new user group. This command can be executed only using api_key
 
        belonging to user with admin rights or an user who has create user group
 
        permission
 

	
 
        :param group_name: name of new user group
 
        :type group_name: str
 
        :param description: group description
 
        :type description: str
 
        :param owner: owner of group. If not passed apiuser is the owner
 
        :type owner: Optional(str or int)
 
        :param active: group is active
 
        :type active: Optional(bool)
 

	
 
        OUTPUT::
 

	
 
            id : <id_given_in_input>
 
            result: {
 
                      "msg": "created new user group `<groupname>`",
 
                      "user_group": <user_group_object>
 
                    }
 
            error:  null
 

	
 
        ERROR OUTPUT::
 

	
 
          id : <id_given_in_input>
 
          result : null
 
          error :  {
 
            "user group `<group name>` already exist"
 
            or
 
            "failed to create group `<group name>`"
 
          }
 

	
 
        """
 

	
 
        if UserGroupModel().get_by_name(group_name):
 
            raise JSONRPCError("user group `%s` already exist" % (group_name,))
 

	
 
        try:
 
            if owner is None:
 
                owner = request.authuser.user_id
 

	
 
            owner = get_user_or_error(owner)
 
            ug = UserGroupModel().create(name=group_name, description=description,
 
                                         owner=owner, active=active)
 
            meta.Session().commit()
 
            return dict(
 
                msg='created new user group `%s`' % group_name,
 
                user_group=ug.get_api_data()
 
            )
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            raise JSONRPCError('failed to create group `%s`' % (group_name,))
 

	
 
    # permission check inside
 
    def update_user_group(self, usergroupid, group_name=None,
 
                          description=None, owner=None,
 
                          active=None):
 
        """
 
        Updates given usergroup.  This command can be executed only using api_key
 
        belonging to user with admin rights or an admin of given user group
 

	
 
        :param usergroupid: id of user group to update
 
        :type usergroupid: str or int
 
        :param group_name: name of new user group
 
        :type group_name: str
 
        :param description: group description
 
        :type description: str
 
        :param owner: owner of group.
 
        :type owner: Optional(str or int)
 
        :param active: group is active
 
        :type active: Optional(bool)
 

	
 
        OUTPUT::
 

	
 
          id : <id_given_in_input>
 
          result : {
 
            "msg": 'updated user group ID:<user group id> <user group name>',
 
            "user_group": <user_group_object>
 
          }
 
          error :  null
 

	
 
        ERROR OUTPUT::
 

	
 
          id : <id_given_in_input>
 
          result : null
 
          error :  {
 
            "failed to update user group `<user group name>`"
 
          }
 

	
 
        """
 
        user_group = get_user_group_or_error(usergroupid)
 
        if not HasPermissionAny('hg.admin')():
 
            if not HasUserGroupPermissionLevel('admin')(user_group.users_group_name):
 
                raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
 

	
 
        if owner is not None:
 
            owner = get_user_or_error(owner)
 

	
 
        updates = {}
 
        store_update(updates, group_name, 'users_group_name')
 
        store_update(updates, description, 'user_group_description')
 
        store_update(updates, owner, 'owner')
 
        store_update(updates, active, 'users_group_active')
 
        try:
 
            UserGroupModel().update(user_group, updates)
 
            meta.Session().commit()
 
            return dict(
 
                msg='updated user group ID:%s %s' % (user_group.users_group_id,
 
                                                     user_group.users_group_name),
 
                user_group=user_group.get_api_data()
 
            )
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            raise JSONRPCError('failed to update user group `%s`' % (usergroupid,))
 

	
 
    # permission check inside
 
    def delete_user_group(self, usergroupid):
 
        """
 
        Delete given user group by user group id or name.
 
        This command can be executed only using api_key
 
        belonging to user with admin rights or an admin of given user group
 

	
 
        :param usergroupid:
 
        :type usergroupid: int
 

	
 
        OUTPUT::
 

	
 
          id : <id_given_in_input>
 
          result : {
 
            "msg": "deleted user group ID:<user_group_id> <user_group_name>"
 
          }
 
          error :  null
 

	
 
        ERROR OUTPUT::
 

	
 
          id : <id_given_in_input>
 
          result : null
 
          error :  {
 
            "failed to delete user group ID:<user_group_id> <user_group_name>"
 
            or
 
            "RepoGroup assigned to <repo_groups_list>"
 
          }
 

	
 
        """
 
        user_group = get_user_group_or_error(usergroupid)
 
        if not HasPermissionAny('hg.admin')():
 
            if not HasUserGroupPermissionLevel('admin')(user_group.users_group_name):
 
                raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
 

	
 
        try:
 
            UserGroupModel().delete(user_group)
 
            meta.Session().commit()
 
            return dict(
 
                msg='deleted user group ID:%s %s' %
 
                    (user_group.users_group_id, user_group.users_group_name),
 
                user_group=None
 
            )
 
        except UserGroupsAssignedException as e:
 
            log.error(traceback.format_exc())
 
            raise JSONRPCError(str(e))
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            raise JSONRPCError('failed to delete user group ID:%s %s' %
 
                               (user_group.users_group_id,
 
                                user_group.users_group_name)
 
                               )
 

	
 
    # permission check inside
 
    def add_user_to_user_group(self, usergroupid, userid):
 
        """
 
        Adds a user to a user group. If user exists in that group success will be
 
        `false`. This command can be executed only using api_key
 
        belonging to user with admin rights  or an admin of given user group
 

	
 
        :param usergroupid:
 
        :type usergroupid: int
 
        :param userid:
 
        :type userid: int
 

	
 
        OUTPUT::
 

	
 
          id : <id_given_in_input>
 
          result : {
 
              "success": True|False # depends on if member is in group
 
              "msg": "added member `<username>` to user group `<groupname>` |
 
                      User is already in that group"
 

	
 
          }
 
          error :  null
 

	
 
        ERROR OUTPUT::
 

	
 
          id : <id_given_in_input>
 
          result : null
 
          error :  {
 
            "failed to add member to user group `<user_group_name>`"
 
          }
 

	
 
        """
 
        user = get_user_or_error(userid)
 
        user_group = get_user_group_or_error(usergroupid)
 
        if not HasPermissionAny('hg.admin')():
 
            if not HasUserGroupPermissionLevel('admin')(user_group.users_group_name):
 
                raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
 

	
 
        try:
 
            ugm = UserGroupModel().add_user_to_group(user_group, user)
 
            success = True if ugm is not True else False
 
            msg = 'added member `%s` to user group `%s`' % (
 
                user.username, user_group.users_group_name
 
            )
 
            msg = msg if success else 'User is already in that group'
 
            meta.Session().commit()
 

	
 
            return dict(
 
                success=success,
 
                msg=msg
 
            )
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            raise JSONRPCError(
 
                'failed to add member to user group `%s`' % (
 
                    user_group.users_group_name,
 
                )
 
            )
 

	
 
    # permission check inside
 
    def remove_user_from_user_group(self, usergroupid, userid):
 
        """
 
        Removes a user from a user group. If user is not in given group success will
 
        be `false`. This command can be executed only
 
        using api_key belonging to user with admin rights or an admin of given user group
 

	
 
        :param usergroupid:
 
        :param userid:
 

	
 

	
 
        OUTPUT::
 

	
 
            id : <id_given_in_input>
 
            result: {
 
                      "success":  True|False,  # depends on if member is in group
 
                      "msg": "removed member <username> from user group <groupname> |
 
                              User wasn't in group"
 
                    }
 
            error:  null
 

	
 
        """
 
        user = get_user_or_error(userid)
 
        user_group = get_user_group_or_error(usergroupid)
 
        if not HasPermissionAny('hg.admin')():
 
            if not HasUserGroupPermissionLevel('admin')(user_group.users_group_name):
 
                raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
 

	
 
        try:
 
            success = UserGroupModel().remove_user_from_group(user_group, user)
 
            msg = 'removed member `%s` from user group `%s`' % (
 
                user.username, user_group.users_group_name
 
            )
 
            msg = msg if success else "User wasn't in group"
 
            meta.Session().commit()
 
            return dict(success=success, msg=msg)
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            raise JSONRPCError(
 
                'failed to remove member from user group `%s`' % (
 
                    user_group.users_group_name,
 
                )
 
            )
 

	
 
    # permission check inside
 
    def get_repo(self, repoid,
 
                 with_revision_names=False,
 
                 with_pullrequests=False):
 
        """
 
        Gets an existing repository by it's name or repository_id. Members will return
 
        either users_group or user associated to that repository. This command can be
 
        executed only using api_key belonging to user with admin
 
        rights or regular user that have at least read access to repository.
 

	
 
        :param repoid: repository name or repository id
 
        :type repoid: str or int
 

	
 
        OUTPUT::
 

	
 
          id : <id_given_in_input>
 
          result : {
 
            {
 
                "repo_id" :          "<repo_id>",
 
                "repo_name" :        "<reponame>"
 
                "repo_type" :        "<repo_type>",
 
                "clone_uri" :        "<clone_uri>",
 
                "enable_downloads":  "<bool>",
 
                "enable_statistics": "<bool>",
 
                "private":           "<bool>",
 
                "created_on" :       "<date_time_created>",
 
                "description" :      "<description>",
 
                "landing_rev":       "<landing_rev>",
 
                "last_changeset":    {
 
                                       "author":   "<full_author>",
 
                                       "date":     "<date_time_of_commit>",
 
                                       "message":  "<commit_message>",
 
                                       "raw_id":   "<raw_id>",
 
                                       "revision": "<numeric_revision>",
 
                                       "short_id": "<short_id>"
 
                                     }
 
                "owner":             "<repo_owner>",
 
                "fork_of":           "<name_of_fork_parent>",
 
                "members" :     [
 
                                  {
 
                                    "name":     "<username>",
 
                                    "type" :    "user",
 
                                    "permission" : "repository.(read|write|admin)"
 
                                  },
 
                                  …
 
                                  {
 
                                    "name":     "<usergroup name>",
 
                                    "type" :    "user_group",
 
                                    "permission" : "usergroup.(read|write|admin)"
 
                                  },
 
                                  …
 
                                ]
 
                 "followers":   [<user_obj>, ...],
 
                 <if with_revision_names == True>
 
                 "tags": {
 
                            "<tagname>": "<raw_id>",
 
                            ...
 
                         },
 
                 "branches": {
 
                            "<branchname>": "<raw_id>",
 
                            ...
 
                         },
 
                 "bookmarks": {
 
                            "<bookmarkname>": "<raw_id>",
 
                            ...
 
                         },
 
            }
 
          }
 
          error :  null
 

	
 
        """
 
        repo = get_repo_or_error(repoid)
 

	
 
        if not HasPermissionAny('hg.admin')():
 
            if not HasRepoPermissionLevel('read')(repo.repo_name):
 
                raise JSONRPCError('repository `%s` does not exist' % (repoid,))
 

	
 
        members = []
 
        for user in repo.repo_to_perm:
 
            perm = user.permission.permission_name
 
            user = user.user
 
            user_data = {
 
                'name': user.username,
 
                'type': "user",
 
                'permission': perm
 
            }
 
            members.append(user_data)
 

	
 
        for user_group in repo.users_group_to_perm:
 
            perm = user_group.permission.permission_name
 
            user_group = user_group.users_group
 
            user_group_data = {
 
                'name': user_group.users_group_name,
 
                'type': "user_group",
 
                'permission': perm
 
            }
 
            members.append(user_group_data)
 

	
 
        followers = [
 
            uf.user.get_api_data()
 
            for uf in repo.followers
 
        ]
 

	
 
        data = repo.get_api_data(with_revision_names=with_revision_names,
 
                                 with_pullrequests=with_pullrequests)
 
        data['members'] = members
 
        data['followers'] = followers
 
        return data
 

	
 
    # permission check inside
 
    def get_repos(self):
 
        """
 
        Lists all existing repositories. This command can be executed only using
 
        api_key belonging to user with admin rights or regular user that have
 
        admin, write or read access to repository.
 

	
 

	
 
        OUTPUT::
 

	
 
            id : <id_given_in_input>
 
            result: [
 
                      {
 
                        "repo_id" :          "<repo_id>",
 
                        "repo_name" :        "<reponame>"
 
                        "repo_type" :        "<repo_type>",
 
                        "clone_uri" :        "<clone_uri>",
 
                        "private": :         "<bool>",
 
                        "created_on" :       "<datetimecreated>",
 
                        "description" :      "<description>",
 
                        "landing_rev":       "<landing_rev>",
 
                        "owner":             "<repo_owner>",
 
                        "fork_of":           "<name_of_fork_parent>",
 
                        "enable_downloads":  "<bool>",
 
                        "enable_statistics": "<bool>",
 
                      },
 
                      …
 
                    ]
 
            error:  null
 
        """
 
        if not HasPermissionAny('hg.admin')():
 
            repos = request.authuser.get_all_user_repos()
 
        else:
 
            repos = db.Repository.query()
 

	
 
        return [
 
            repo.get_api_data()
 
            for repo in repos
 
        ]
 

	
 
    # permission check inside
 
    def get_repo_nodes(self, repoid, revision, root_path,
 
                       ret_type='all'):
 
        """
 
        returns a list of nodes and it's children in a flat list for a given path
 
        at given revision. It's possible to specify ret_type to show only `files` or
 
        `dirs`.  This command can be executed only using api_key belonging to
 
        user with admin rights or regular user that have at least read access to repository.
 

	
 
        :param repoid: repository name or repository id
 
        :type repoid: str or int
 
        :param revision: revision for which listing should be done
 
        :type revision: str
 
        :param root_path: path from which start displaying
 
        :type root_path: str
 
        :param ret_type: return type 'all|files|dirs' nodes
 
        :type ret_type: Optional(str)
 

	
 

	
 
        OUTPUT::
 

	
 
            id : <id_given_in_input>
 
            result: [
 
                      {
 
                        "name" :        "<name>"
 
                        "type" :        "<type>",
 
                      },
 
                      …
 
                    ]
 
            error:  null
 
        """
 
        repo = get_repo_or_error(repoid)
 

	
 
        if not HasPermissionAny('hg.admin')():
 
            if not HasRepoPermissionLevel('read')(repo.repo_name):
 
                raise JSONRPCError('repository `%s` does not exist' % (repoid,))
 

	
 
        _map = {}
 
        try:
 
            _d, _f = ScmModel().get_nodes(repo, revision, root_path,
 
                                          flat=False)
 
            _map = {
 
                'all': _d + _f,
 
                'files': _f,
 
                'dirs': _d,
 
            }
 
            return _map[ret_type]
 
        except KeyError:
 
            raise JSONRPCError('ret_type must be one of %s'
 
                               % (','.join(sorted(_map))))
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            raise JSONRPCError(
 
                'failed to get repo: `%s` nodes' % repo.repo_name
 
            )
 

	
 
    @HasPermissionAnyDecorator('hg.admin', 'hg.create.repository')
 
    def create_repo(self, repo_name, owner=None,
 
                    repo_type=None, description='',
 
                    private=False, clone_uri=None,
 
                    landing_rev='rev:tip',
 
                    enable_statistics=None,
 
                    enable_downloads=None,
 
                    copy_permissions=False):
 
        """
 
        Creates a repository. The repository name contains the full path, but the
 
        parent repository group must exist. For example "foo/bar/baz" require the groups
 
        "foo" and "bar" (with "foo" as parent), and create "baz" repository with
 
        "bar" as group. This command can be executed only using api_key
 
        belonging to user with admin rights or regular user that have create
 
        repository permission. Regular users cannot specify owner parameter
 

	
 
        :param repo_name: repository name
 
        :type repo_name: str
 
        :param owner: user_id or username
 
        :type owner: Optional(str)
 
        :param repo_type: 'hg' or 'git'
 
        :type repo_type: Optional(str)
 
        :param description: repository description
 
        :type description: Optional(str)
 
        :param private:
 
        :type private: bool
 
        :param clone_uri:
 
        :type clone_uri: str
 
        :param landing_rev: <rev_type>:<rev>
 
        :type landing_rev: str
 
        :param enable_downloads:
 
        :type enable_downloads: bool
 
        :param enable_statistics:
 
        :type enable_statistics: bool
 
        :param copy_permissions: Copy permission from group that repository is
 
            being created.
 
        :type copy_permissions: bool
 

	
 
        OUTPUT::
 

	
 
            id : <id_given_in_input>
 
            result: {
 
                      "msg": "Created new repository `<reponame>`",
 
                      "success": true,
 
                      "task": "<celery task id or None if done sync>"
 
                    }
 
            error:  null
 

	
 
        ERROR OUTPUT::
 

	
 
          id : <id_given_in_input>
 
          result : null
 
          error :  {
 
             'failed to create repository `<repo_name>`
 
          }
 

	
 
        """
 
        if not HasPermissionAny('hg.admin')():
 
            if owner is not None:
 
                # forbid setting owner for non-admins
 
                raise JSONRPCError(
 
                    'Only Kallithea admin can specify `owner` param'
 
                )
 
        if owner is None:
 
            owner = request.authuser.user_id
 

	
 
        owner = get_user_or_error(owner)
 

	
 
        if RepoModel().get_by_repo_name(repo_name):
 
            raise JSONRPCError("repo `%s` already exist" % repo_name)
 

	
 
        defs = db.Setting.get_default_repo_settings(strip_prefix=True)
 
        if private is None:
 
            private = defs.get('repo_private') or False
 
        if repo_type is None:
 
            repo_type = defs.get('repo_type')
 
        if enable_statistics is None:
 
            enable_statistics = defs.get('repo_enable_statistics')
 
        if enable_downloads is None:
 
            enable_downloads = defs.get('repo_enable_downloads')
 

	
 
        try:
 
            repo_name_parts = repo_name.split('/')
 
            repo_group = None
 
            if len(repo_name_parts) > 1:
 
                group_name = '/'.join(repo_name_parts[:-1])
 
                repo_group = db.RepoGroup.get_by_group_name(group_name)
 
                if repo_group is None:
 
                    raise JSONRPCError("repo group `%s` not found" % group_name)
 
            data = dict(
 
                repo_name=repo_name_parts[-1],
 
                repo_name_full=repo_name,
 
                repo_type=repo_type,
 
                repo_description=description,
 
                repo_private=private,
 
                clone_uri=clone_uri,
 
                repo_group=repo_group,
 
                repo_landing_rev=landing_rev,
 
                enable_statistics=enable_statistics,
 
                enable_downloads=enable_downloads,
 
                repo_copy_permissions=copy_permissions,
 
            )
 

	
 
            task = RepoModel().create(form_data=data, cur_user=owner.username)
 
            task_id = task.task_id
 
            # no commit, it's done in RepoModel, or async via celery
 
            return dict(
 
                msg="Created new repository `%s`" % (repo_name,),
 
                success=True,  # cannot return the repo data here since fork
 
                               # can be done async
 
                task=task_id
 
            )
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            raise JSONRPCError(
 
                'failed to create repository `%s`' % (repo_name,))
 

	
 
    # permission check inside
 
    def update_repo(self, repoid, name=None,
 
                    owner=None,
 
                    group=None,
 
                    description=None, private=None,
 
                    clone_uri=None, landing_rev=None,
 
                    enable_statistics=None,
 
                    enable_downloads=None):
 

	
 
        """
 
        Updates repo
 

	
 
        :param repoid: repository name or repository id
 
        :type repoid: str or int
 
        :param name:
 
        :param owner:
 
        :param group:
 
        :param description:
 
        :param private:
 
        :param clone_uri:
 
        :param landing_rev:
 
        :param enable_statistics:
 
        :param enable_downloads:
 
        """
 
        repo = get_repo_or_error(repoid)
 
        if not HasPermissionAny('hg.admin')():
 
            if not HasRepoPermissionLevel('admin')(repo.repo_name):
 
                raise JSONRPCError('repository `%s` does not exist' % (repoid,))
 

	
 
            if (name != repo.repo_name and
 
                not HasPermissionAny('hg.create.repository')()
 
            ):
 
                raise JSONRPCError('no permission to create (or move) repositories')
 

	
 
            if owner is not None:
 
                # forbid setting owner for non-admins
 
                raise JSONRPCError(
 
                    'Only Kallithea admin can specify `owner` param'
 
                )
 

	
 
        updates = {}
 
        repo_group = group
 
        if repo_group is not None:
 
            repo_group = get_repo_group_or_error(repo_group)
 
            repo_group = repo_group.group_id
 
        try:
 
            store_update(updates, name, 'repo_name')
 
            store_update(updates, repo_group, 'repo_group')
 
            store_update(updates, owner, 'owner')
 
            store_update(updates, description, 'repo_description')
 
            store_update(updates, private, 'repo_private')
 
            store_update(updates, clone_uri, 'clone_uri')
 
            store_update(updates, landing_rev, 'repo_landing_rev')
 
            store_update(updates, enable_statistics, 'repo_enable_statistics')
 
            store_update(updates, enable_downloads, 'repo_enable_downloads')
 

	
 
            RepoModel().update(repo, **updates)
 
            meta.Session().commit()
 
            return dict(
 
                msg='updated repo ID:%s %s' % (repo.repo_id, repo.repo_name),
 
                repository=repo.get_api_data()
 
            )
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            raise JSONRPCError('failed to update repo `%s`' % repoid)
 

	
 
    @HasPermissionAnyDecorator('hg.admin', 'hg.fork.repository')
 
    def fork_repo(self, repoid, fork_name,
 
                  owner=None,
 
                  description='', copy_permissions=False,
 
                  private=False, landing_rev='rev:tip'):
 
        """
 
        Creates a fork of given repo. In case of using celery this will
 
        immediately return success message, while fork is going to be created
 
        asynchronous. This command can be executed only using api_key belonging to
 
        user with admin rights or regular user that have fork permission, and at least
 
        read access to forking repository. Regular users cannot specify owner parameter.
 

	
 
        :param repoid: repository name or repository id
 
        :type repoid: str or int
 
        :param fork_name:
 
        :param owner:
 
        :param description:
 
        :param copy_permissions:
 
        :param private:
 
        :param landing_rev:
 

	
 
        INPUT::
 

	
 
            id : <id_for_response>
 
            api_key : "<api_key>"
 
            args:     {
 
                        "repoid" :          "<reponame or repo_id>",
 
                        "fork_name":        "<forkname>",
 
                        "owner":            "<username or user_id = Optional(=apiuser)>",
 
                        "description":      "<description>",
 
                        "copy_permissions": "<bool>",
 
                        "private":          "<bool>",
 
                        "landing_rev":      "<landing_rev>"
 
                      }
 

	
 
        OUTPUT::
 

	
 
            id : <id_given_in_input>
 
            result: {
 
                      "msg": "Created fork of `<reponame>` as `<forkname>`",
 
                      "success": true,
 
                      "task": "<celery task id or None if done sync>"
 
                    }
 
            error:  null
 

	
 
        """
 
        repo = get_repo_or_error(repoid)
 
        repo_name = repo.repo_name
 

	
 
        _repo = RepoModel().get_by_repo_name(fork_name)
 
        if _repo:
 
            type_ = 'fork' if _repo.fork else 'repo'
 
            raise JSONRPCError("%s `%s` already exist" % (type_, fork_name))
 

	
 
        if HasPermissionAny('hg.admin')():
 
            pass
 
        elif HasRepoPermissionLevel('read')(repo.repo_name):
 
            if owner is not None:
 
                # forbid setting owner for non-admins
 
                raise JSONRPCError(
 
                    'Only Kallithea admin can specify `owner` param'
 
                )
 

	
 
            if not HasPermissionAny('hg.create.repository')():
 
                raise JSONRPCError('no permission to create repositories')
 
        else:
 
            raise JSONRPCError('repository `%s` does not exist' % (repoid,))
 

	
 
        if owner is None:
 
            owner = request.authuser.user_id
 

	
 
        owner = get_user_or_error(owner)
 

	
 
        try:
 
            fork_name_parts = fork_name.split('/')
 
            repo_group = None
 
            if len(fork_name_parts) > 1:
 
                group_name = '/'.join(fork_name_parts[:-1])
 
                repo_group = db.RepoGroup.get_by_group_name(group_name)
 
                if repo_group is None:
 
                    raise JSONRPCError("repo group `%s` not found" % group_name)
 

	
 
            form_data = dict(
 
                repo_name=fork_name_parts[-1],
 
                repo_name_full=fork_name,
 
                repo_group=repo_group,
 
                repo_type=repo.repo_type,
 
                description=description,
 
                private=private,
 
                copy_permissions=copy_permissions,
 
                landing_rev=landing_rev,
 
                update_after_clone=False,
 
                fork_parent_id=repo.repo_id,
 
            )
 
            task = RepoModel().create_fork(form_data, cur_user=owner.username)
 
            # no commit, it's done in RepoModel, or async via celery
 
            task_id = task.task_id
 
            return dict(
 
                msg='Created fork of `%s` as `%s`' % (repo.repo_name,
 
                                                      fork_name),
 
                success=True,  # cannot return the repo data here since fork
 
                               # can be done async
 
                task=task_id
 
            )
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            raise JSONRPCError(
 
                'failed to fork repository `%s` as `%s`' % (repo_name,
 
                                                            fork_name)
 
            )
 

	
 
    # permission check inside
 
    def delete_repo(self, repoid, forks=''):
 
        """
 
        Deletes a repository. This command can be executed only using api_key belonging
 
        to user with admin rights or regular user that have admin access to repository.
 
        When `forks` param is set it's possible to detach or delete forks of deleting
 
        repository
 

	
 
        :param repoid: repository name or repository id
 
        :type repoid: str or int
 
        :param forks: `detach` or `delete`, what do do with attached forks for repo
 
        :type forks: Optional(str)
 

	
 
        OUTPUT::
 

	
 
            id : <id_given_in_input>
 
            result: {
 
                      "msg": "Deleted repository `<reponame>`",
 
                      "success": true
 
                    }
 
            error:  null
 

	
 
        """
 
        repo = get_repo_or_error(repoid)
 

	
 
        if not HasPermissionAny('hg.admin')():
 
            if not HasRepoPermissionLevel('admin')(repo.repo_name):
 
                raise JSONRPCError('repository `%s` does not exist' % (repoid,))
 

	
 
        try:
 
            handle_forks = forks
 
            _forks_msg = ''
 
            _forks = [f for f in repo.forks]
 
            if handle_forks == 'detach':
 
                _forks_msg = ' ' + 'Detached %s forks' % len(_forks)
 
            elif handle_forks == 'delete':
 
                _forks_msg = ' ' + 'Deleted %s forks' % len(_forks)
 
            elif _forks:
 
                raise JSONRPCError(
 
                    'Cannot delete `%s` it still contains attached forks' %
 
                    (repo.repo_name,)
 
                )
 

	
 
            RepoModel().delete(repo, forks=forks)
 
            meta.Session().commit()
 
            return dict(
 
                msg='Deleted repository `%s`%s' % (repo.repo_name, _forks_msg),
 
                success=True
 
            )
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            raise JSONRPCError(
 
                'failed to delete repository `%s`' % (repo.repo_name,)
 
            )
 

	
 
    @HasPermissionAnyDecorator('hg.admin')
 
    def grant_user_permission(self, repoid, userid, perm):
 
        """
 
        Grant permission for user on given repository, or update existing one
 
        if found. 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
 
        :param userid:
 
        :param perm: (repository.(none|read|write|admin))
 
        :type perm: str
 

	
 
        OUTPUT::
 

	
 
            id : <id_given_in_input>
 
            result: {
 
                      "msg" : "Granted perm: `<perm>` for user: `<username>` in repo: `<reponame>`",
 
                      "success": true
 
                    }
 
            error:  null
 
        """
 
        repo = get_repo_or_error(repoid)
 
        user = get_user_or_error(userid)
 
        perm = get_perm_or_error(perm)
 

	
 
        try:
 

	
 
            RepoModel().grant_user_permission(repo=repo, user=user, perm=perm)
 

	
 
            meta.Session().commit()
 
            return dict(
 
                msg='Granted perm: `%s` for user: `%s` in repo: `%s`' % (
 
                    perm.permission_name, user.username, repo.repo_name
 
                ),
 
                success=True
 
            )
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            raise JSONRPCError(
 
                'failed to edit permission for user: `%s` in repo: `%s`' % (
 
                    userid, repoid
 
                )
 
            )
 

	
 
    @HasPermissionAnyDecorator('hg.admin')
 
    def revoke_user_permission(self, repoid, userid):
 
        """
 
        Revoke permission for user on given repository. 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
 
        :param userid:
 

	
 
        OUTPUT::
 

	
 
            id : <id_given_in_input>
 
            result: {
 
                      "msg" : "Revoked perm for user: `<username>` in repo: `<reponame>`",
 
                      "success": true
 
                    }
 
            error:  null
 

	
 
        """
 

	
 
        repo = get_repo_or_error(repoid)
 
        user = get_user_or_error(userid)
 
        try:
 
            RepoModel().revoke_user_permission(repo=repo, user=user)
 
            meta.Session().commit()
 
            return dict(
 
                msg='Revoked perm for user: `%s` in repo: `%s`' % (
 
                    user.username, repo.repo_name
 
                ),
 
                success=True
 
            )
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            raise JSONRPCError(
 
                'failed to edit permission for user: `%s` in repo: `%s`' % (
 
                    userid, repoid
 
                )
 
            )
 

	
 
    # permission check inside
 
    def grant_user_group_permission(self, repoid, usergroupid, perm):
 
        """
 
        Grant permission for user group on given repository, or update
 
        existing one if found. 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
 
        :param usergroupid: id of usergroup
 
        :type usergroupid: str or int
 
        :param perm: (repository.(none|read|write|admin))
 
        :type perm: str
 

	
 
        OUTPUT::
 

	
 
          id : <id_given_in_input>
 
          result : {
 
            "msg" : "Granted perm: `<perm>` for group: `<usersgroupname>` in repo: `<reponame>`",
 
            "success": true
 

	
 
          }
 
          error :  null
 

	
 
        ERROR OUTPUT::
 

	
 
          id : <id_given_in_input>
 
          result : null
 
          error :  {
 
            "failed to edit permission for user group: `<usergroup>` in repo `<repo>`'
 
          }
 

	
 
        """
 
        repo = get_repo_or_error(repoid)
 
        perm = get_perm_or_error(perm)
 
        user_group = get_user_group_or_error(usergroupid)
 
        if not HasPermissionAny('hg.admin')():
 
            if not HasRepoPermissionLevel('admin')(repo.repo_name):
 
                raise JSONRPCError('repository `%s` does not exist' % (repoid,))
 

	
 
            if not HasUserGroupPermissionLevel('read')(user_group.users_group_name):
 
                raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
 

	
 
        try:
 
            RepoModel().grant_user_group_permission(
 
                repo=repo, group_name=user_group, perm=perm)
 

	
 
            meta.Session().commit()
 
            return dict(
 
                msg='Granted perm: `%s` for user group: `%s` in '
 
                    'repo: `%s`' % (
 
                        perm.permission_name, user_group.users_group_name,
 
                        repo.repo_name
 
                    ),
 
                success=True
 
            )
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            raise JSONRPCError(
 
                'failed to edit permission for user group: `%s` in '
 
                'repo: `%s`' % (
 
                    usergroupid, repo.repo_name
 
                )
 
            )
 

	
 
    # permission check inside
 
    def revoke_user_group_permission(self, repoid, usergroupid):
 
        """
 
        Revoke permission for user group on given repository. 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
 
        :param usergroupid:
 

	
 
        OUTPUT::
 

	
 
            id : <id_given_in_input>
 
            result: {
 
                      "msg" : "Revoked perm for group: `<usersgroupname>` in repo: `<reponame>`",
 
                      "success": true
 
                    }
 
            error:  null
 
        """
 
        repo = get_repo_or_error(repoid)
 
        user_group = get_user_group_or_error(usergroupid)
 
        if not HasPermissionAny('hg.admin')():
 
            if not HasRepoPermissionLevel('admin')(repo.repo_name):
 
                raise JSONRPCError('repository `%s` does not exist' % (repoid,))
 

	
 
            if not HasUserGroupPermissionLevel('read')(user_group.users_group_name):
 
                raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
 

	
 
        try:
 
            RepoModel().revoke_user_group_permission(
 
                repo=repo, group_name=user_group)
 

	
 
            meta.Session().commit()
 
            return dict(
 
                msg='Revoked perm for user group: `%s` in repo: `%s`' % (
 
                    user_group.users_group_name, repo.repo_name
 
                ),
 
                success=True
 
            )
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            raise JSONRPCError(
 
                'failed to edit permission for user group: `%s` in '
 
                'repo: `%s`' % (
 
                    user_group.users_group_name, repo.repo_name
 
                )
 
            )
 

	
 
    @HasPermissionAnyDecorator('hg.admin')
 
    def get_repo_group(self, repogroupid):
 
        """
 
        Returns given repo group together with permissions, and repositories
 
        inside the group
 

	
 
        :param repogroupid: id/name of repository group
 
        :type repogroupid: str or int
 
        """
 
        repo_group = get_repo_group_or_error(repogroupid)
 

	
 
        members = []
 
        for user in repo_group.repo_group_to_perm:
 
            perm = user.permission.permission_name
 
            user = user.user
 
            user_data = {
 
                'name': user.username,
 
                'type': "user",
 
                'permission': perm
 
            }
 
            members.append(user_data)
 

	
 
        for user_group in repo_group.users_group_to_perm:
 
            perm = user_group.permission.permission_name
 
            user_group = user_group.users_group
 
            user_group_data = {
 
                'name': user_group.users_group_name,
 
                'type': "user_group",
 
                'permission': perm
 
            }
 
            members.append(user_group_data)
 

	
 
        data = repo_group.get_api_data()
 
        data["members"] = members
 
        return data
 

	
 
    @HasPermissionAnyDecorator('hg.admin')
 
    def get_repo_groups(self):
 
        """
 
        Returns all repository groups
 

	
 
        """
 
        return [
 
            repo_group.get_api_data()
 
            for repo_group in db.RepoGroup.query()
 
        ]
 

	
 
    @HasPermissionAnyDecorator('hg.admin')
 
    def create_repo_group(self, group_name, description='',
 
                          owner=None,
 
                          parent=None,
 
                          copy_permissions=False):
 
        """
 
        Creates a repository group. This command can be executed only using
 
        api_key belonging to user with admin rights.
 

	
 
        :param group_name:
 
        :type group_name:
 
        :param description:
 
        :type description:
 
        :param owner:
 
        :type owner:
 
        :param parent:
 
        :type parent:
 
        :param copy_permissions:
 
        :type copy_permissions:
 

	
 
        OUTPUT::
 

	
 
          id : <id_given_in_input>
 
          result : {
 
              "msg": "created new repo group `<repo_group_name>`"
 
              "repo_group": <repogroup_object>
 
          }
 
          error :  null
 

	
 
        ERROR OUTPUT::
 

	
 
          id : <id_given_in_input>
 
          result : null
 
          error :  {
 
            failed to create repo group `<repogroupid>`
 
          }
 

	
 
        """
 
        if db.RepoGroup.get_by_group_name(group_name):
 
            raise JSONRPCError("repo group `%s` already exist" % (group_name,))
 

	
 
        if owner is None:
 
            owner = request.authuser.user_id
 
        group_description = description
 
        parent_group = None
 
        if parent is not None:
 
            parent_group = get_repo_group_or_error(parent)
 

	
 
        try:
 
            repo_group = RepoGroupModel().create(
 
                group_name=group_name,
 
                group_description=group_description,
 
                owner=owner,
 
                parent=parent_group,
 
                copy_permissions=copy_permissions
 
            )
 
            meta.Session().commit()
 
            return dict(
 
                msg='created new repo group `%s`' % group_name,
 
                repo_group=repo_group.get_api_data()
 
            )
 
        except Exception:
 

	
 
            log.error(traceback.format_exc())
 
            raise JSONRPCError('failed to create repo group `%s`' % (group_name,))
 

	
 
    @HasPermissionAnyDecorator('hg.admin')
 
    def update_repo_group(self, repogroupid, group_name=None,
 
                          description=None,
 
                          owner=None,
 
                          parent=None):
 
        repo_group = get_repo_group_or_error(repogroupid)
 

	
 
        updates = {}
 
        try:
 
            store_update(updates, group_name, 'group_name')
 
            store_update(updates, description, 'group_description')
 
            store_update(updates, owner, 'owner')
 
            store_update(updates, parent, 'parent_group')
 
            repo_group = RepoGroupModel().update(repo_group, updates)
 
            meta.Session().commit()
 
            return dict(
 
                msg='updated repository group ID:%s %s' % (repo_group.group_id,
 
                                                           repo_group.group_name),
 
                repo_group=repo_group.get_api_data()
 
            )
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            raise JSONRPCError('failed to update repository group `%s`'
 
                               % (repogroupid,))
 

	
 
    @HasPermissionAnyDecorator('hg.admin')
 
    def delete_repo_group(self, repogroupid):
 
        """
 

	
 
        :param repogroupid: name or id of repository group
 
        :type repogroupid: str or int
 

	
 
        OUTPUT::
 

	
 
          id : <id_given_in_input>
 
          result : {
 
            'msg': 'deleted repo group ID:<repogroupid> <repogroupname>
 
            'repo_group': null
 
          }
 
          error :  null
 

	
 
        ERROR OUTPUT::
 

	
 
          id : <id_given_in_input>
 
          result : null
 
          error :  {
 
            "failed to delete repo group ID:<repogroupid> <repogroupname>"
 
          }
 

	
 
        """
 
        repo_group = get_repo_group_or_error(repogroupid)
 

	
 
        try:
 
            RepoGroupModel().delete(repo_group)
 
            meta.Session().commit()
 
            return dict(
 
                msg='deleted repo group ID:%s %s' %
 
                    (repo_group.group_id, repo_group.group_name),
 
                repo_group=None
 
            )
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            raise JSONRPCError('failed to delete repo group ID:%s %s' %
 
                               (repo_group.group_id, repo_group.group_name)
 
                               )
 

	
 
    # permission check inside
 
    def grant_user_permission_to_repo_group(self, repogroupid, userid,
 
                                            perm, apply_to_children='none'):
 
        """
 
        Grant permission for user on given repository group, or update existing
 
        one if found. This command can be executed only using api_key belonging
 
        to user with admin rights, or user who has admin right to given repository
 
        group.
 

	
 
        :param repogroupid: name or id of repository group
 
        :type repogroupid: str or int
 
        :param userid:
 
        :param perm: (group.(none|read|write|admin))
 
        :type perm: str
 
        :param apply_to_children: 'none', 'repos', 'groups', 'all'
 
        :type apply_to_children: str
 

	
 
        OUTPUT::
 

	
 
            id : <id_given_in_input>
 
            result: {
 
                      "msg" : "Granted perm: `<perm>` (recursive:<apply_to_children>) for user: `<username>` in repo group: `<repo_group_name>`",
 
                      "success": true
 
                    }
 
            error:  null
 

	
 
        ERROR OUTPUT::
 

	
 
          id : <id_given_in_input>
 
          result : null
 
          error :  {
 
            "failed to edit permission for user: `<userid>` in repo group: `<repo_group_name>`"
 
          }
 

	
 
        """
 

	
 
        repo_group = get_repo_group_or_error(repogroupid)
 

	
 
        if not HasPermissionAny('hg.admin')():
 
            if not HasRepoGroupPermissionLevel('admin')(repo_group.group_name):
 
                raise JSONRPCError('repository group `%s` does not exist' % (repogroupid,))
 

	
 
        user = get_user_or_error(userid)
 
        perm = get_perm_or_error(perm, prefix='group.')
 

	
 
        try:
 
            RepoGroupModel().add_permission(repo_group=repo_group,
 
                                            obj=user,
 
                                            obj_type="user",
 
                                            perm=perm,
 
                                            recursive=apply_to_children)
 
            meta.Session().commit()
 
            return dict(
 
                msg='Granted perm: `%s` (recursive:%s) for user: `%s` in repo group: `%s`' % (
 
                    perm.permission_name, apply_to_children, user.username, repo_group.name
 
                ),
 
                success=True
 
            )
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            raise JSONRPCError(
 
                'failed to edit permission for user: `%s` in repo group: `%s`' % (
 
                    userid, repo_group.name))
 

	
 
    # permission check inside
 
    def revoke_user_permission_from_repo_group(self, repogroupid, userid,
 
                                               apply_to_children='none'):
 
        """
 
        Revoke permission for user on given repository group. This command can
 
        be executed only using api_key belonging to user with admin rights, or
 
        user who has admin right to given repository group.
 

	
 
        :param repogroupid: name or id of repository group
 
        :type repogroupid: str or int
 
        :param userid:
 
        :type userid:
 
        :param apply_to_children: 'none', 'repos', 'groups', 'all'
 
        :type apply_to_children: str
 

	
 
        OUTPUT::
 

	
 
            id : <id_given_in_input>
 
            result: {
 
                      "msg" : "Revoked perm (recursive:<apply_to_children>) for user: `<username>` in repo group: `<repo_group_name>`",
 
                      "success": true
 
                    }
 
            error:  null
 

	
 
        ERROR OUTPUT::
 

	
 
          id : <id_given_in_input>
 
          result : null
 
          error :  {
 
            "failed to edit permission for user: `<userid>` in repo group: `<repo_group_name>`"
 
          }
 

	
 
        """
 

	
 
        repo_group = get_repo_group_or_error(repogroupid)
 

	
 
        if not HasPermissionAny('hg.admin')():
 
            if not HasRepoGroupPermissionLevel('admin')(repo_group.group_name):
 
                raise JSONRPCError('repository group `%s` does not exist' % (repogroupid,))
 

	
 
        user = get_user_or_error(userid)
 

	
 
        try:
 
            RepoGroupModel().delete_permission(repo_group=repo_group,
 
                                               obj=user,
 
                                               obj_type="user",
 
                                               recursive=apply_to_children)
 

	
 
            meta.Session().commit()
 
            return dict(
 
                msg='Revoked perm (recursive:%s) for user: `%s` in repo group: `%s`' % (
 
                    apply_to_children, user.username, repo_group.name
 
                ),
 
                success=True
 
            )
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            raise JSONRPCError(
 
                'failed to edit permission for user: `%s` in repo group: `%s`' % (
 
                    userid, repo_group.name))
 

	
 
    # permission check inside
 
    def grant_user_group_permission_to_repo_group(
 
            self, repogroupid, usergroupid, perm,
 
            apply_to_children='none'):
 
        """
 
        Grant permission for user group on given repository group, or update
 
        existing one if found. This command can be executed only using
 
        api_key belonging to user with admin rights, or user who has admin
 
        right to given repository group.
 

	
 
        :param repogroupid: name or id of repository group
 
        :type repogroupid: str or int
 
        :param usergroupid: id of usergroup
 
        :type usergroupid: str or int
 
        :param perm: (group.(none|read|write|admin))
 
        :type perm: str
 
        :param apply_to_children: 'none', 'repos', 'groups', 'all'
 
        :type apply_to_children: str
 

	
 
        OUTPUT::
 

	
 
          id : <id_given_in_input>
 
          result : {
 
            "msg" : "Granted perm: `<perm>` (recursive:<apply_to_children>) for user group: `<usersgroupname>` in repo group: `<repo_group_name>`",
 
            "success": true
 

	
 
          }
 
          error :  null
 

	
 
        ERROR OUTPUT::
 

	
 
          id : <id_given_in_input>
 
          result : null
 
          error :  {
 
            "failed to edit permission for user group: `<usergroup>` in repo group: `<repo_group_name>`"
 
          }
 

	
 
        """
 
        repo_group = get_repo_group_or_error(repogroupid)
 
        perm = get_perm_or_error(perm, prefix='group.')
 
        user_group = get_user_group_or_error(usergroupid)
 
        if not HasPermissionAny('hg.admin')():
 
            if not HasRepoGroupPermissionLevel('admin')(repo_group.group_name):
 
                raise JSONRPCError(
 
                    'repository group `%s` does not exist' % (repogroupid,))
 

	
 
            if not HasUserGroupPermissionLevel('read')(user_group.users_group_name):
 
                raise JSONRPCError(
 
                    'user group `%s` does not exist' % (usergroupid,))
 

	
 
        try:
 
            RepoGroupModel().add_permission(repo_group=repo_group,
 
                                            obj=user_group,
 
                                            obj_type="user_group",
 
                                            perm=perm,
 
                                            recursive=apply_to_children)
 
            meta.Session().commit()
 
            return dict(
 
                msg='Granted perm: `%s` (recursive:%s) for user group: `%s` in repo group: `%s`' % (
 
                    perm.permission_name, apply_to_children,
 
                    user_group.users_group_name, repo_group.name
 
                ),
 
                success=True
 
            )
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            raise JSONRPCError(
 
                'failed to edit permission for user group: `%s` in '
 
                'repo group: `%s`' % (
 
                    usergroupid, repo_group.name
 
                )
 
            )
 

	
 
    # permission check inside
 
    def revoke_user_group_permission_from_repo_group(
 
            self, repogroupid, usergroupid,
 
            apply_to_children='none'):
 
        """
 
        Revoke permission for user group on given repository. This command can be
 
        executed only using api_key belonging to user with admin rights, or
 
        user who has admin right to given repository group.
 

	
 
        :param repogroupid: name or id of repository group
 
        :type repogroupid: str or int
 
        :param usergroupid:
 
        :param apply_to_children: 'none', 'repos', 'groups', 'all'
 
        :type apply_to_children: str
 

	
 
        OUTPUT::
 

	
 
            id : <id_given_in_input>
 
            result: {
 
                      "msg" : "Revoked perm (recursive:<apply_to_children>) for user group: `<usersgroupname>` in repo group: `<repo_group_name>`",
 
                      "success": true
 
                    }
 
            error:  null
 

	
 
        ERROR OUTPUT::
 

	
 
          id : <id_given_in_input>
 
          result : null
 
          error :  {
 
            "failed to edit permission for user group: `<usergroup>` in repo group: `<repo_group_name>`"
 
          }
 

	
 

	
 
        """
 
        repo_group = get_repo_group_or_error(repogroupid)
 
        user_group = get_user_group_or_error(usergroupid)
 
        if not HasPermissionAny('hg.admin')():
 
            if not HasRepoGroupPermissionLevel('admin')(repo_group.group_name):
 
                raise JSONRPCError(
 
                    'repository group `%s` does not exist' % (repogroupid,))
 

	
 
            if not HasUserGroupPermissionLevel('read')(user_group.users_group_name):
 
                raise JSONRPCError(
 
                    'user group `%s` does not exist' % (usergroupid,))
 

	
 
        try:
 
            RepoGroupModel().delete_permission(repo_group=repo_group,
 
                                               obj=user_group,
 
                                               obj_type="user_group",
 
                                               recursive=apply_to_children)
 
            meta.Session().commit()
 
            return dict(
 
                msg='Revoked perm (recursive:%s) for user group: `%s` in repo group: `%s`' % (
 
                    apply_to_children, user_group.users_group_name, repo_group.name
 
                ),
 
                success=True
 
            )
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            raise JSONRPCError(
 
                'failed to edit permission for user group: `%s` in repo group: `%s`' % (
 
                    user_group.users_group_name, repo_group.name
 
                )
 
            )
 

	
 
    def get_gist(self, gistid):
 
        """
 
        Get given gist by id
 

	
 
        :param gistid: id of private or public gist
 
        :type gistid: str
 
        """
 
        gist = get_gist_or_error(gistid)
 
        if not HasPermissionAny('hg.admin')():
 
            if gist.owner_id != request.authuser.user_id:
 
                raise JSONRPCError('gist `%s` does not exist' % (gistid,))
 
        return gist.get_api_data()
 

	
 
    def get_gists(self, userid=None):
 
        """
 
        Get all gists for given user. If userid is empty returned gists
 
        are for user who called the api
 

	
 
        :param userid: user to get gists for
 
        :type userid: Optional(str or int)
 
        """
 
        if not HasPermissionAny('hg.admin')():
 
            # make sure normal user does not pass someone else userid,
 
            # he is not allowed to do that
 
            if userid is not None and userid != request.authuser.user_id:
 
                raise JSONRPCError(
 
                    'userid is not the same as your user'
 
                )
 

	
 
        if userid is None:
 
            user_id = request.authuser.user_id
 
        else:
 
            user_id = get_user_or_error(userid).user_id
 

	
 
        return [
 
            gist.get_api_data()
 
            for gist in db.Gist().query()
 
                .filter_by(is_expired=False)
 
                .filter(db.Gist.owner_id == user_id)
 
                .order_by(db.Gist.created_on.desc())
 
        ]
 

	
 
    def create_gist(self, files, owner=None,
 
                    gist_type=db.Gist.GIST_PUBLIC, lifetime=-1,
 
                    description=''):
 

	
 
        """
 
        Creates new Gist
 

	
 
        :param files: files to be added to gist
 
            {'filename': {'content':'...', 'lexer': null},
 
             'filename2': {'content':'...', 'lexer': null}}
 
        :type files: dict
 
        :param owner: gist owner, defaults to api method caller
 
        :type owner: Optional(str or int)
 
        :param gist_type: type of gist 'public' or 'private'
 
        :type gist_type: Optional(str)
 
        :param lifetime: time in minutes of gist lifetime
 
        :type lifetime: Optional(int)
 
        :param description: gist description
 
        :type description: Optional(str)
 

	
 
        OUTPUT::
 

	
 
          id : <id_given_in_input>
kallithea/controllers/forks.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.forks
 
~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

	
 
forks 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 23, 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 tg import request
 
from tg import tmpl_context as c
 
from tg.i18n import ugettext as _
 
from webob.exc import HTTPFound, HTTPNotFound
 

	
 
import kallithea
 
from kallithea.controllers import base
 
from kallithea.lib import webutils
 
from kallithea.lib.auth import HasPermissionAnyDecorator, HasRepoPermissionLevel, HasRepoPermissionLevelDecorator, LoginRequired
 
from kallithea.lib.page import Page
 
from kallithea.lib.utils2 import safe_int
 
from kallithea.model import db
 
from kallithea.model.forms import RepoForkForm
 
from kallithea.model.repo import RepoModel
 
from kallithea.model.scm import AvailableRepoGroupChoices, ScmModel
 

	
 

	
 
log = logging.getLogger(__name__)
 

	
 

	
 
class ForksController(base.BaseRepoController):
 

	
 
    def __load_defaults(self):
 
        c.repo_groups = AvailableRepoGroupChoices('write')
 

	
 
        c.landing_revs_choices, c.landing_revs = ScmModel().get_repo_landing_revs()
 

	
 
        c.can_update = db.Ui.get_by_key('hooks', db.Ui.HOOK_UPDATE).ui_active
 

	
 
    def __load_data(self):
 
        """
 
        Load defaults settings for edit, and update
 
        """
 
        self.__load_defaults()
 

	
 
        c.repo_info = c.db_repo
 
        repo = c.db_repo.scm_instance
 

	
 
        if c.repo_info is None:
 
            raise HTTPNotFound()
 

	
 
        c.default_user_id = kallithea.DEFAULT_USER_ID
 
        c.in_public_journal = db.UserFollowing.query() \
 
            .filter(db.UserFollowing.user_id == c.default_user_id) \
 
            .filter(db.UserFollowing.follows_repository == c.repo_info).scalar()
 

	
 
        if c.repo_info.stats:
 
            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)
 

	
 
        defaults = RepoModel()._get_defaults(c.repo_name)
 
        # alter the description to indicate a fork
 
        defaults['description'] = ('fork of repository: %s \n%s'
 
                                   % (defaults['repo_name'],
 
                                      defaults['description']))
 
        # add suffix to fork
 
        defaults['repo_name'] = '%s-fork' % defaults['repo_name']
 

	
 
        return defaults
 

	
 
    @LoginRequired(allow_default_user=True)
 
    @HasRepoPermissionLevelDecorator('read')
 
    def forks(self, repo_name):
 
        p = safe_int(request.GET.get('page'), 1)
 
        repo_id = c.db_repo.repo_id
 
        d = []
 
        for r in db.Repository.get_repo_forks(repo_id):
 
            if not HasRepoPermissionLevel('read')(r.repo_name, 'get forks check'):
 
                continue
 
            d.append(r)
 
        c.forks_pager = Page(d, page=p, items_per_page=20)
 

	
 
        if request.environ.get('HTTP_X_PARTIAL_XHR'):
 
            return base.render('/forks/forks_data.html')
 

	
 
        return base.render('/forks/forks.html')
 

	
 
    @LoginRequired()
 
    @HasPermissionAnyDecorator('hg.admin', 'hg.fork.repository')
 
    @HasRepoPermissionLevelDecorator('read')
 
    def fork(self, repo_name):
 
        c.repo_info = db.Repository.get_by_repo_name(repo_name)
 
        if not c.repo_info:
 
            raise HTTPNotFound()
 

	
 
        defaults = self.__load_data()
 

	
 
        return htmlfill.render(
 
            base.render('forks/fork.html'),
 
            defaults=defaults,
 
            encoding="UTF-8",
 
            force_defaults=False)
 

	
 
    @LoginRequired()
 
    @HasPermissionAnyDecorator('hg.admin', 'hg.fork.repository')
 
    @HasRepoPermissionLevelDecorator('read')
 
    def fork_create(self, repo_name):
 
        self.__load_defaults()
 
        c.repo_info = db.Repository.get_by_repo_name(repo_name)
 
        _form = RepoForkForm(old_data={'repo_type': c.repo_info.repo_type},
 
                             repo_groups=c.repo_groups,
 
                             landing_revs=c.landing_revs_choices)()
 
        form_result = {}
 
        task_id = None
 
        try:
 
            form_result = _form.to_python(dict(request.POST))
 

	
 
            # an approximation that is better than nothing
 
            if not db.Ui.get_by_key('hooks', db.Ui.HOOK_UPDATE).ui_active:
 
                form_result['update_after_clone'] = False
 

	
 
            # create fork is done sometimes async on celery, db transaction
 
            # management is handled there.
 
            task = RepoModel().create_fork(form_result, request.authuser.user_id)
 
            task_id = task.task_id
 
        except formencode.Invalid as errors:
 
            return htmlfill.render(
 
                base.render('forks/fork.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())
 
            webutils.flash(_('An error occurred during repository forking %s') %
 
                    repo_name, category='error')
 

	
 
        raise HTTPFound(location=webutils.url('repo_creating_home',
 
                              repo_name=form_result['repo_name_full'],
 
                              task_id=task_id))
 
                              ))
kallithea/lib/celerylib/__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.lib.celerylib
 
~~~~~~~~~~~~~~~~~~~~~~~
 

	
 
celery libs 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 27, 2010
 
:author: marcink
 
:copyright: (c) 2013 RhodeCode GmbH, and others.
 
:license: GPLv3, see LICENSE.md for more details.
 
"""
 

	
 

	
 
import logging
 
import os
 
from hashlib import sha1
 

	
 
from decorator import decorator
 
from tg import config
 

	
 
import kallithea
 
from kallithea.lib.pidlock import DaemonLock, LockHeld
 
from kallithea.lib.utils2 import safe_bytes
 
from kallithea.model import meta
 

	
 

	
 
log = logging.getLogger(__name__)
 

	
 

	
 
class FakeTask(object):
 
    """Fake a sync result to make it look like a finished task"""
 

	
 
    def __init__(self, result):
 
        self.result = result
 

	
 
    def failed(self):
 
        return False
 

	
 
    traceback = None # if failed
 

	
 
    task_id = None
 

	
 

	
 
def task(f_org):
 
    """Wrapper of celery.task.task, running async if CELERY_APP
 
    """
 

	
 
    if kallithea.CELERY_APP:
 
        def f_async(*args, **kwargs):
 
            log.info('executing %s task', f_org.__name__)
 
            try:
 
                f_org(*args, **kwargs)
 
            finally:
 
                meta.Session.remove()  # prevent reuse of auto created db sessions
 
                log.info('executed %s task', f_org.__name__)
 
        runner = kallithea.CELERY_APP.task(name=f_org.__name__, ignore_result=True)(f_async)
 

	
 
        def f_wrapped(*args, **kwargs):
 
            t = runner.apply_async(args=args, kwargs=kwargs)
 
            log.info('executing task %s in async mode - id %s', f_org, t.task_id)
 
            return t
 
    else:
 
        def f_wrapped(*args, **kwargs):
 
            log.info('executing task %s in sync', f_org.__name__)
 
            try:
 
                f_org(*args, **kwargs)
 
            except Exception as e:
 
                log.error('exception executing sync task %s in sync: %r', f_org.__name__, e)
 
                raise # TODO: return this in FakeTask as with async tasks?
 
            return FakeTask(None)
 

	
 
    return f_wrapped
 

	
 

	
 
def __get_lockkey(func, *fargs, **fkwargs):
 
    params = list(fargs)
 
    params.extend(['%s-%s' % ar for ar in fkwargs.items()])
 

	
 
    func_name = str(func.__name__) if hasattr(func, '__name__') else str(func)
 

	
 
    lockkey = 'task_%s.lock' % \
 
        sha1(safe_bytes(func_name + '-' + '-'.join(str(x) for x in params))).hexdigest()
 
    return lockkey
 

	
 

	
 
def locked_task(func):
 
    def __wrapper(func, *fargs, **fkwargs):
 
        lockkey = __get_lockkey(func, *fargs, **fkwargs)
 
        log.info('running task with lockkey %s', lockkey)
 
        try:
 
            l = DaemonLock(os.path.join(config['cache_dir'], lockkey))
 
            ret = func(*fargs, **fkwargs)
 
            l.release()
 
            return ret
 
        except LockHeld:
 
            log.info('LockHeld')
 
            return 'Task with key %s already running' % lockkey
 

	
 
    return decorator(__wrapper, func)
kallithea/templates/admin/repos/repo_creating.html
Show inline comments
 
## -*- coding: utf-8 -*-
 
<%inherit file="/base/base.html"/>
 

	
 
## don't trigger flash messages on this page
 
<%def name="flash_msg()">
 
</%def>
 

	
 
<%block name="title">
 
    ${_('%s Creating Repository') % c.repo_name}
 
</%block>
 

	
 
<%def name="breadcrumbs_links()">
 
    ${_('Creating repository')} ${c.repo}
 
</%def>
 

	
 
<%block name="header_menu">
 
    ${self.menu('repositories')}
 
</%block>
 
<%def name="main()">
 
<div class="panel panel-primary">
 
    <div class="panel-heading clearfix">
 
        ${self.breadcrumbs()}
 
    </div>
 

	
 
    <div class="panel-body">
 
            <h4 class="text-center">
 
                ${_('Repository "%(repo_name)s" is being created, you will be redirected when this process is finished.' % {'repo_name':c.repo_name})}
 
            </h4>
 

	
 
        <div id="progress">
 
            <div class="progress progress-striped active">
 
                <div class="progress-bar" role="progressbar"
 
                    aria-valuenow="100" aria-valuemin="0" aria-valuemax="100">
 
                </div>
 
            </div>
 
        </div>
 
        <div id="progress_error" style="display: none;">
 
            <div class="alert alert-danger">
 
                ${_("We're sorry but error occurred during this operation. Please check your Kallithea server logs, or contact administrator.")}
 
            </div>
 
        </div>
 
    </div>
 
</div>
 

	
 
<script>
 
'use strict';
 
(function worker() {
 
  $.ajax({
 
    url: ${h.js(h.url('repo_check_home', repo_name=c.repo_name, repo=c.repo, task_id=c.task_id))},
 
    url: ${h.js(h.url('repo_check_home', repo_name=c.repo_name, repo=c.repo))},
 
    success: function(data) {
 
      if(data.result === true){
 
          //redirect to created fork if our ajax loop tells us to do so.
 
          window.location = ${h.js(h.url('summary_home', repo_name = c.repo))};
 
      }
 
    },
 
    complete: function(resp) {
 
      if (resp.status == 200){
 
          // Schedule the next request when the current one's complete
 
          setTimeout(worker, 1000);
 
      }
 
      else{
 
          $("#progress").html($('#progress_error').html());
 
      }
 
    }
 
  });
 
})();
 
</script>
 
</%def>
kallithea/templates/ini/template.ini.mako
Show inline comments
 
## -*- coding: utf-8 -*-
 
<%text>##</%text>#################################################################################
 
<%text>##</%text>#################################################################################
 
<%text>##</%text> Kallithea config file generated with kallithea-cli ${'%-27s' % version       }##
 
<%text>##</%text>                                                                               ##
 
<%text>##</%text> The %(here)s variable will generally be replaced with the parent directory of ##
 
<%text>##</%text> this file. Other use of % must be escaped as %% .                             ##
 
<%text>##</%text>#################################################################################
 
<%text>##</%text>#################################################################################
 

	
 
[DEFAULT]
 

	
 
<%text>##</%text>##############################################################################
 
<%text>##</%text> Email settings                                                             ##
 
<%text>##</%text>                                                                            ##
 
<%text>##</%text> Refer to the documentation ("Email settings") for more details.            ##
 
<%text>##</%text>                                                                            ##
 
<%text>##</%text> It is recommended to use a valid sender address that passes access         ##
 
<%text>##</%text> validation and spam filtering in mail servers.                             ##
 
<%text>##</%text>##############################################################################
 

	
 
<%text>##</%text> 'From' header for application emails. You can optionally add a name.
 
<%text>##</%text> Default:
 
#app_email_from = Kallithea
 
<%text>##</%text> Examples:
 
#app_email_from = Kallithea <kallithea-noreply@example.com>
 
#app_email_from = kallithea-noreply@example.com
 

	
 
<%text>##</%text> Subject prefix for application emails.
 
<%text>##</%text> A space between this prefix and the real subject is automatically added.
 
<%text>##</%text> Default:
 
#email_prefix =
 
<%text>##</%text> Example:
 
#email_prefix = [Kallithea]
 

	
 
<%text>##</%text> Recipients for error emails and fallback recipients of application mails.
 
<%text>##</%text> Multiple addresses can be specified, comma-separated.
 
<%text>##</%text> Only addresses are allowed, do not add any name part.
 
<%text>##</%text> Default:
 
#email_to =
 
<%text>##</%text> Examples:
 
#email_to = admin@example.com
 
#email_to = admin@example.com,another_admin@example.com
 
email_to =
 

	
 
<%text>##</%text> 'From' header for error emails. You can optionally add a name.
 
<%text>##</%text> Default: (none)
 
<%text>##</%text> Examples:
 
#error_email_from = Kallithea Errors <kallithea-noreply@example.com>
 
#error_email_from = kallithea_errors@example.com
 
error_email_from =
 

	
 
<%text>##</%text> SMTP server settings
 
<%text>##</%text> If specifying credentials, make sure to use secure connections.
 
<%text>##</%text> Default: Send unencrypted unauthenticated mails to the specified smtp_server.
 
<%text>##</%text> For "SSL", use smtp_use_ssl = true and smtp_port = 465.
 
<%text>##</%text> For "STARTTLS", use smtp_use_tls = true and smtp_port = 587.
 
smtp_server =
 
smtp_username =
 
smtp_password =
 
smtp_port =
 
smtp_use_ssl = false
 
smtp_use_tls = false
 

	
 
%if http_server != 'uwsgi':
 
<%text>##</%text> Entry point for 'gearbox serve'
 
[server:main]
 
host = ${host}
 
port = ${port}
 

	
 
%if http_server == 'gearbox':
 
<%text>##</%text> Gearbox serve uses the built-in development web server ##
 
use = egg:gearbox#wsgiref
 
<%text>##</%text> nr of worker threads to spawn
 
threadpool_workers = 1
 
<%text>##</%text> max request before thread respawn
 
threadpool_max_requests = 100
 
<%text>##</%text> option to use threads of process
 
use_threadpool = true
 

	
 
%elif http_server == 'gevent':
 
<%text>##</%text> Gearbox serve uses the gevent web server ##
 
use = egg:gearbox#gevent
 

	
 
%elif http_server == 'waitress':
 
<%text>##</%text> Gearbox serve uses the Waitress web server ##
 
use = egg:waitress#main
 
<%text>##</%text> avoid multi threading
 
threads = 1
 
<%text>##</%text> allow push of repos bigger than the default of 1 GB
 
max_request_body_size = 107374182400
 
<%text>##</%text> use poll instead of select, fixes fd limits, may not work on old
 
<%text>##</%text> windows systems.
 
#asyncore_use_poll = True
 

	
 
%elif http_server == 'gunicorn':
 
<%text>##</%text> Gearbox serve uses the Gunicorn web server ##
 
use = egg:gunicorn#main
 
<%text>##</%text> number of process workers. You must set `instance_id = *` when this option
 
<%text>##</%text> is set to more than one worker
 
workers = 4
 
<%text>##</%text> process name
 
proc_name = kallithea
 
<%text>##</%text> type of worker class, one of sync, eventlet, gevent, tornado
 
<%text>##</%text> recommended for bigger setup is using of of other than sync one
 
worker_class = sync
 
max_requests = 1000
 
<%text>##</%text> amount of time a worker can handle request before it gets killed and
 
<%text>##</%text> restarted
 
timeout = 3600
 

	
 
%endif
 
%else:
 
<%text>##</%text> UWSGI ##
 
[uwsgi]
 
<%text>##</%text> Note: this section is parsed by the uWSGI .ini parser when run as:
 
<%text>##</%text> uwsgi --venv /srv/kallithea/venv --ini-paste-logged my.ini
 
<%text>##</%text> Note: in uWSGI 2.0.18 or older, pastescript needs to be installed to
 
<%text>##</%text> get correct application logging. In later versions this is not necessary.
 
<%text>##</%text> pip install pastescript
 

	
 
<%text>##</%text> HTTP Basics:
 
http-socket = ${host}:${port}
 
buffer-size = 65535                    ; Mercurial will use huge GET headers for discovery
 

	
 
<%text>##</%text> Scaling:
 
master = true                          ; Use separate master and worker processes
 
auto-procname = true                   ; Name worker processes accordingly
 
lazy = true                            ; App *must* be loaded in workers - db connections can't be shared
 
workers = 4                            ; On demand scaling up to this many worker processes
 
cheaper = 1                            ; Initial and on demand scaling down to this many worker processes
 
max-requests = 1000                    ; Graceful reload of worker processes to avoid leaks
 

	
 
<%text>##</%text> Tweak defaults:
 
strict = true                          ; Fail on unknown config directives
 
enable-threads = true                  ; Enable Python threads (not threaded workers)
 
vacuum = true                          ; Delete sockets during shutdown
 
single-interpreter = true
 
die-on-term = true                     ; Shutdown when receiving SIGTERM (default is respawn)
 
need-app = true                        ; Exit early if no app can be loaded.
 
reload-on-exception = true             ; Don't assume that the application worker can process more requests after a severe error
 

	
 
%endif
 
<%text>##</%text> middleware for hosting the WSGI application under a URL prefix
 
#[filter:proxy-prefix]
 
#use = egg:PasteDeploy#prefix
 
#prefix = /<your-prefix>
 

	
 
[app:main]
 
use = egg:kallithea
 
<%text>##</%text> enable proxy prefix middleware
 
#filter-with = proxy-prefix
 

	
 
full_stack = true
 
static_files = true
 

	
 
<%text>##</%text> Internationalization (see setup documentation for details)
 
<%text>##</%text> By default, the languages requested by the browser are used if available, with English as default.
 
<%text>##</%text> Set i18n.enabled=false to disable automatic language choice.
 
#i18n.enabled = true
 
<%text>##</%text> To Force a language, set i18n.enabled=false and specify the language in i18n.lang.
 
<%text>##</%text> Valid values are the names of subdirectories in kallithea/i18n with a LC_MESSAGES/kallithea.mo
 
#i18n.lang = en
 

	
 
cache_dir = %(here)s/data
 
index_dir = %(here)s/data/index
 

	
 
<%text>##</%text> uncomment and set this path to use archive download cache
 
archive_cache_dir = %(here)s/data/tarballcache
 

	
 
<%text>##</%text> change this to unique ID for security
 
app_instance_uuid = ${uuid()}
 

	
 
<%text>##</%text> cut off limit for large diffs (size in bytes)
 
cut_off_limit = 256000
 

	
 
<%text>##</%text> force https in Kallithea, fixes https redirects, assumes it's always https
 
force_https = false
 

	
 
<%text>##</%text> use Strict-Transport-Security headers
 
use_htsts = false
 

	
 
<%text>##</%text> number of commits stats will parse on each iteration
 
commit_parse_limit = 25
 

	
 
<%text>##</%text> Path to Python executable to be used for git hooks.
 
<%text>##</%text> This value will be written inside the git hook scripts as the text
 
<%text>##</%text> after '#!' (shebang). When empty or not defined, the value of
 
<%text>##</%text> 'sys.executable' at the time of installation of the git hooks is
 
<%text>##</%text> used, which is correct in many cases but for example not when using uwsgi.
 
<%text>##</%text> If you change this setting, you should reinstall the Git hooks via
 
<%text>##</%text> Admin > Settings > Remap and Rescan.
 
#git_hook_interpreter = /srv/kallithea/venv/bin/python3
 
%if git_hook_interpreter:
 
git_hook_interpreter = ${git_hook_interpreter}
 
%endif
 

	
 
<%text>##</%text> path to git executable
 
git_path = git
 

	
 
<%text>##</%text> git rev filter option, --all is the default filter, if you need to
 
<%text>##</%text> hide all refs in changelog switch this to --branches --tags
 
#git_rev_filter = --branches --tags
 

	
 
<%text>##</%text> RSS feed options
 
rss_cut_off_limit = 256000
 
rss_items_per_page = 10
 
rss_include_diff = false
 

	
 
<%text>##</%text> options for showing and identifying changesets
 
show_sha_length = 12
 
show_revision_number = false
 

	
 
<%text>##</%text> Canonical URL to use when creating full URLs in UI and texts.
 
<%text>##</%text> Useful when the site is available under different names or protocols.
 
<%text>##</%text> Defaults to what is provided in the WSGI environment.
 
#canonical_url = https://kallithea.example.com/repos
 

	
 
<%text>##</%text> gist URL alias, used to create nicer urls for gist. This should be an
 
<%text>##</%text> url that does rewrites to _admin/gists/<gistid>.
 
<%text>##</%text> example: http://gist.example.com/{gistid}. Empty means use the internal
 
<%text>##</%text> Kallithea url, ie. http[s]://kallithea.example.com/_admin/gists/<gistid>
 
gist_alias_url =
 

	
 
<%text>##</%text> default encoding used to convert from and to unicode
 
<%text>##</%text> can be also a comma separated list of encoding in case of mixed encodings
 
default_encoding = utf-8
 

	
 
<%text>##</%text> Set Mercurial encoding, similar to setting HGENCODING before launching Kallithea
 
hgencoding = utf-8
 

	
 
<%text>##</%text> issue tracker for Kallithea (leave blank to disable, absent for default)
 
#bugtracker = https://bitbucket.org/conservancy/kallithea/issues
 

	
 
<%text>##</%text> issue tracking mapping for commit messages, comments, PR descriptions, ...
 
<%text>##</%text> Refer to the documentation ("Integration with issue trackers") for more details.
 

	
 
<%text>##</%text> regular expression to match issue references
 
<%text>##</%text> This pattern may/should contain parenthesized groups, that can
 
<%text>##</%text> be referred to in issue_server_link or issue_sub using Python backreferences
 
<%text>##</%text> (e.g. \1, \2, ...). You can also create named groups with '(?P<groupname>)'.
 
<%text>##</%text> To require mandatory whitespace before the issue pattern, use:
 
<%text>##</%text> (?:^|(?<=\s)) before the actual pattern, and for mandatory whitespace
 
<%text>##</%text> behind the issue pattern, use (?:$|(?=\s)) after the actual pattern.
 

	
 
issue_pat = #(\d+)
 

	
 
<%text>##</%text> server url to the issue
 
<%text>##</%text> This pattern may/should contain backreferences to parenthesized groups in issue_pat.
 
<%text>##</%text> A backreference can be \1, \2, ... or \g<groupname> if you specified a named group
 
<%text>##</%text> called 'groupname' in issue_pat.
 
<%text>##</%text> The special token {repo} is replaced with the full repository name
 
<%text>##</%text> including repository groups, while {repo_name} is replaced with just
 
<%text>##</%text> the name of the repository.
 

	
 
issue_server_link = https://issues.example.com/{repo}/issue/\1
 

	
 
<%text>##</%text> substitution pattern to use as the link text
 
<%text>##</%text> If issue_sub is empty, the text matched by issue_pat is retained verbatim
 
<%text>##</%text> for the link text. Otherwise, the link text is that of issue_sub, with any
 
<%text>##</%text> backreferences to groups in issue_pat replaced.
 

	
 
issue_sub =
 

	
 
<%text>##</%text> issue_pat, issue_server_link and issue_sub can have suffixes to specify
 
<%text>##</%text> multiple patterns, to other issues server, wiki or others
 
<%text>##</%text> below an example how to create a wiki pattern
 
<%text>##</%text> wiki-some-id -> https://wiki.example.com/some-id
 

	
 
#issue_pat_wiki = wiki-(\S+)
 
#issue_server_link_wiki = https://wiki.example.com/\1
 
#issue_sub_wiki = WIKI-\1
 

	
 
<%text>##</%text> alternative return HTTP header for failed authentication. Default HTTP
 
<%text>##</%text> response is 401 HTTPUnauthorized. Currently Mercurial clients have trouble with
 
<%text>##</%text> handling that. Set this variable to 403 to return HTTPForbidden
 
auth_ret_code =
 

	
 
<%text>##</%text> allows to change the repository location in settings page
 
allow_repo_location_change = True
 

	
 
<%text>##</%text> allows to setup custom hooks in settings page
 
allow_custom_hooks_settings = True
 

	
 
<%text>##</%text> extra extensions for indexing, space separated and without the leading '.'.
 
#index.extensions =
 
#    gemfile
 
#    lock
 

	
 
<%text>##</%text> extra filenames for indexing, space separated
 
#index.filenames =
 
#    .dockerignore
 
#    .editorconfig
 
#    INSTALL
 
#    CHANGELOG
 

	
 
<%text>##</%text>##################################
 
<%text>##</%text>            SSH CONFIG          ##
 
<%text>##</%text>##################################
 

	
 
<%text>##</%text> SSH is disabled by default, until an Administrator decides to enable it.
 
ssh_enabled = false
 

	
 
<%text>##</%text> File where users' SSH keys will be stored *if* ssh_enabled is true.
 
#ssh_authorized_keys = /home/kallithea/.ssh/authorized_keys
 
%if user_home_path:
 
ssh_authorized_keys = ${user_home_path}/.ssh/authorized_keys
 
%endif
 

	
 
<%text>##</%text> Path to be used in ssh_authorized_keys file to invoke kallithea-cli with ssh-serve.
 
#kallithea_cli_path = /srv/kallithea/venv/bin/kallithea-cli
 
%if kallithea_cli_path:
 
kallithea_cli_path = ${kallithea_cli_path}
 
%endif
 

	
 
<%text>##</%text> Locale to be used in the ssh-serve command.
 
<%text>##</%text> This is needed because an SSH client may try to use its own locale
 
<%text>##</%text> settings, which may not be available on the server.
 
<%text>##</%text> See `locale -a` for valid values on this system.
 
#ssh_locale = C.UTF-8
 
%if ssh_locale:
 
ssh_locale = ${ssh_locale}
 
%endif
 

	
 
<%text>##</%text>##################################
 
<%text>##</%text>         CELERY CONFIG          ##
 
<%text>##</%text>##################################
 

	
 
<%text>##</%text> Note: Celery doesn't support Windows.
 
use_celery = false
 

	
 
<%text>##</%text> Celery config settings from https://docs.celeryproject.org/en/4.4.0/userguide/configuration.html prefixed with 'celery.'.
 

	
 
<%text>##</%text> Example: use the message queue on the local virtual host 'kallitheavhost' as the RabbitMQ user 'kallithea':
 
celery.broker_url = amqp://kallithea:thepassword@localhost:5672/kallitheavhost
 

	
 
celery.result_backend = db+sqlite:///celery-results.db
 

	
 
#celery.amqp.task.result.expires = 18000
 

	
 
celery.worker_concurrency = 2
 
celery.worker_max_tasks_per_child = 100
 

	
 
<%text>##</%text>##################################
 
<%text>##</%text>          BEAKER CACHE          ##
 
<%text>##</%text>##################################
 

	
 
beaker.cache.data_dir = %(here)s/data/cache/data
 
beaker.cache.lock_dir = %(here)s/data/cache/lock
 

	
 
beaker.cache.regions = long_term,long_term_file
 

	
 
beaker.cache.long_term.type = memory
 
beaker.cache.long_term.expire = 36000
 
beaker.cache.long_term.key_length = 256
 

	
 
beaker.cache.long_term_file.type = file
 
beaker.cache.long_term_file.expire = 604800
 
beaker.cache.long_term_file.key_length = 256
 

	
 
<%text>##</%text>##################################
 
<%text>##</%text>        BEAKER SESSION          ##
 
<%text>##</%text>##################################
 

	
 
<%text>##</%text> Name of session cookie. Should be unique for a given host and path, even when running
 
<%text>##</%text> on different ports. Otherwise, cookie sessions will be shared and messed up.
 
session.key = kallithea
 
<%text>##</%text> Sessions should always only be accessible by the browser, not directly by JavaScript.
 
session.httponly = true
 
<%text>##</%text> Session lifetime. 2592000 seconds is 30 days.
 
session.timeout = 2592000
 

	
 
<%text>##</%text> Server secret used with HMAC to ensure integrity of cookies.
 
session.secret = ${uuid()}
 
<%text>##</%text> Further, encrypt the data with AES.
 
#session.encrypt_key = <key_for_encryption>
 
#session.validate_key = <validation_key>
 

	
 
<%text>##</%text> Type of storage used for the session, current types are
 
<%text>##</%text> dbm, file, memcached, database, and memory.
 

	
 
<%text>##</%text> File system storage of session data. (default)
 
#session.type = file
 

	
 
<%text>##</%text> Cookie only, store all session data inside the cookie. Requires secure secrets.
 
#session.type = cookie
 

	
 
<%text>##</%text> Database storage of session data.
 
#session.type = ext:database
 
#session.sa.url = postgresql://postgres:qwe@localhost/kallithea
 
#session.table_name = db_session
 

	
 
<%text>##</%text>##################################
 
<%text>##</%text>        ERROR HANDLING          ##
 
<%text>##</%text>##################################
 

	
 
<%text>##</%text> Show a nice error page for application HTTP errors and exceptions (default true)
 
#errorpage.enabled = true
 

	
 
<%text>##</%text> Enable Backlash client-side interactive debugger (default false)
 
<%text>##</%text> WARNING: *THIS MUST BE false IN PRODUCTION ENVIRONMENTS!!!*
 
<%text>##</%text> This debug mode will allow all visitors to execute malicious code.
 
#debug = false
 

	
 
<%text>##</%text> Enable Backlash server-side error reporting (unless debug mode handles it client-side) (default true)
 
#trace_errors.enable = true
 
<%text>##</%text> Errors will be reported by mail if trace_errors.error_email is set.
 

	
 
<%text>##</%text> Propagate email settings to ErrorReporter of TurboGears2
 
<%text>##</%text> You do not normally need to change these lines
 
get trace_errors.smtp_server = smtp_server
 
get trace_errors.smtp_port = smtp_port
 
get trace_errors.from_address = error_email_from
 
get trace_errors.error_email = email_to
 
get trace_errors.smtp_username = smtp_username
 
get trace_errors.smtp_password = smtp_password
 
get trace_errors.smtp_use_tls = smtp_use_tls
 

	
 
<%text>##</%text>################################
 
<%text>##</%text>        LOGVIEW CONFIG        ##
 
<%text>##</%text>################################
 

	
 
logview.sqlalchemy = #faa
 
logview.pylons.templating = #bfb
 
logview.pylons.util = #eee
 

	
 
<%text>##</%text>#######################
 
<%text>##</%text>      DB CONFIG      ##
 
<%text>##</%text>#######################
 

	
 
%if database_engine == 'sqlite':
 
sqlalchemy.url = sqlite:///%(here)s/kallithea.db?timeout=60
 
%else:
 
#sqlalchemy.url = sqlite:///%(here)s/kallithea.db?timeout=60
 
%endif
 
%if database_engine == 'postgres':
 
sqlalchemy.url = postgresql://kallithea:password@localhost/kallithea
 
%else:
 
#sqlalchemy.url = postgresql://kallithea:password@localhost/kallithea
 
%endif
 
%if database_engine == 'mysql':
 
sqlalchemy.url = mysql://kallithea:password@localhost/kallithea?charset=utf8mb4
 
%else:
 
#sqlalchemy.url = mysql://kallithea:password@localhost/kallithea?charset=utf8mb4
 
%endif
 
<%text>##</%text> Note: the mysql:// prefix should also be used for MariaDB
 

	
 
sqlalchemy.pool_recycle = 3600
 

	
 
<%text>##</%text>##############################
 
<%text>##</%text>   ALEMBIC CONFIGURATION    ##
 
<%text>##</%text>##############################
 

	
 
[alembic]
 
script_location = kallithea:alembic
 

	
 
<%text>##</%text>##############################
 
<%text>##</%text>   LOGGING CONFIGURATION    ##
 
<%text>##</%text>##############################
 

	
 
[loggers]
 
keys = root, routes, kallithea, sqlalchemy, tg, gearbox, beaker, templates, whoosh_indexer, werkzeug, backlash
 

	
 
[handlers]
 
keys = console, console_color, console_color_sql, null
 

	
 
[formatters]
 
keys = generic, color_formatter, color_formatter_sql
 

	
 
<%text>##</%text>###########
 
<%text>##</%text> LOGGERS ##
 
<%text>##</%text>###########
 

	
 
[logger_root]
 
level = NOTSET
 
handlers = console
 
<%text>##</%text> For coloring based on log level:
 
#handlers = console_color
 

	
 
[logger_routes]
 
level = WARN
 
handlers =
 
qualname = routes.middleware
 
<%text>##</%text> "level = DEBUG" logs the route matched and routing variables.
 

	
 
[logger_beaker]
 
level = WARN
 
handlers =
 
qualname = beaker.container
 

	
 
[logger_templates]
 
level = WARN
 
handlers =
 
qualname = pylons.templating
 

	
 
[logger_kallithea]
 
level = WARN
 
handlers =
 
qualname = kallithea
 

	
 
[logger_tg]
 
level = WARN
 
handlers =
 
qualname = tg
 

	
 
[logger_gearbox]
 
level = WARN
 
handlers =
 
qualname = gearbox
 

	
 
[logger_sqlalchemy]
 
level = WARN
 
handlers =
 
qualname = sqlalchemy.engine
 
<%text>##</%text> For coloring based on log level and pretty printing of SQL:
 
#level = INFO
 
#handlers = console_color_sql
 
#propagate = 0
 

	
 
[logger_whoosh_indexer]
 
level = WARN
 
handlers =
 
qualname = whoosh_indexer
 

	
 
[logger_werkzeug]
 
level = WARN
 
handlers =
 
qualname = werkzeug
 

	
 
[logger_backlash]
 
level = WARN
 
handlers =
 
qualname = backlash
 

	
 
<%text>##</%text>############
 
<%text>##</%text> HANDLERS ##
 
<%text>##</%text>############
 

	
 
[handler_console]
 
class = StreamHandler
 
args = (sys.stderr,)
 
formatter = generic
 

	
 
[handler_console_color]
 
<%text>##</%text> ANSI color coding based on log level
 
class = StreamHandler
 
args = (sys.stderr,)
 
formatter = color_formatter
 

	
 
[handler_console_color_sql]
 
<%text>##</%text> ANSI color coding and pretty printing of SQL statements
 
class = StreamHandler
 
args = (sys.stderr,)
 
formatter = color_formatter_sql
 

	
 
[handler_null]
 
class = NullHandler
 
args = ()
 

	
 
<%text>##</%text>##############
 
<%text>##</%text> FORMATTERS ##
 
<%text>##</%text>##############
 

	
 
[formatter_generic]
 
format = %(asctime)s.%(msecs)03d %(levelname)-5.5s [%(name)s] %(message)s
 
datefmt = %Y-%m-%d %H:%M:%S
 

	
 
[formatter_color_formatter]
 
class = kallithea.lib.colored_formatter.ColorFormatter
 
format = %(asctime)s.%(msecs)03d %(levelname)-5.5s [%(name)s] %(message)s
 
datefmt = %Y-%m-%d %H:%M:%S
 

	
 
[formatter_color_formatter_sql]
 
class = kallithea.lib.colored_formatter.ColorFormatterSql
 
format = %(asctime)s.%(msecs)03d %(levelname)-5.5s [%(name)s] %(message)s
 
datefmt = %Y-%m-%d %H:%M:%S
 

	
 
<%text>##</%text>###############
 
<%text>##</%text> SSH LOGGING ##
 
<%text>##</%text>###############
 

	
 
<%text>##</%text> The default loggers use 'handler_console' that uses StreamHandler with
 
<%text>##</%text> destination 'sys.stderr'. In the context of the SSH server process, these log
 
<%text>##</%text> messages would be sent to the client, which is normally not what you want.
 
<%text>##</%text> By default, when running ssh-serve, just use NullHandler and disable logging
 
<%text>##</%text> completely. For other logging options, see:
 
<%text>##</%text> https://docs.python.org/2/library/logging.handlers.html
 

	
 
[ssh_serve:logger_root]
 
level = CRITICAL
 
handlers = null
 

	
 
<%text>##</%text> Note: If logging is configured with other handlers, they might need similar
 
<%text>##</%text> muting for ssh-serve too.
kallithea/tests/api/api_base.py
Show inline comments
 
@@ -24,2029 +24,2022 @@ import string
 
import mock
 
import pytest
 

	
 
from kallithea.lib import ext_json
 
from kallithea.lib.auth import AuthUser
 
from kallithea.lib.utils2 import ascii_bytes
 
from kallithea.model import db, meta
 
from kallithea.model.changeset_status import ChangesetStatusModel
 
from kallithea.model.gist import GistModel
 
from kallithea.model.pull_request import PullRequestModel
 
from kallithea.model.repo import RepoModel
 
from kallithea.model.repo_group import RepoGroupModel
 
from kallithea.model.scm import ScmModel
 
from kallithea.model.user import UserModel
 
from kallithea.model.user_group import UserGroupModel
 
from kallithea.tests import base
 
from kallithea.tests.fixture import Fixture, raise_exception
 

	
 

	
 
API_URL = '/_admin/api'
 
TEST_USER_GROUP = 'test_user_group'
 
TEST_REPO_GROUP = 'test_repo_group'
 

	
 
fixture = Fixture()
 

	
 

	
 
def _build_data(apikey, method, **kw):
 
    """
 
    Builds API data with given random ID
 
    For convenience, the json is returned as str
 
    """
 
    random_id = random.randrange(1, 9999)
 
    return random_id, ext_json.dumps({
 
        "id": random_id,
 
        "api_key": apikey,
 
        "method": method,
 
        "args": kw
 
    })
 

	
 

	
 
jsonify = lambda obj: ext_json.loads(ext_json.dumps(obj))
 

	
 

	
 
def api_call(test_obj, params):
 
    response = test_obj.app.post(API_URL, content_type='application/json',
 
                                 params=params)
 
    return response
 

	
 

	
 
## helpers
 
def make_user_group(name=TEST_USER_GROUP):
 
    gr = fixture.create_user_group(name, cur_user=base.TEST_USER_ADMIN_LOGIN)
 
    UserGroupModel().add_user_to_group(user_group=gr,
 
                                       user=base.TEST_USER_ADMIN_LOGIN)
 
    meta.Session().commit()
 
    return gr
 

	
 

	
 
def make_repo_group(name=TEST_REPO_GROUP):
 
    gr = fixture.create_repo_group(name, cur_user=base.TEST_USER_ADMIN_LOGIN)
 
    meta.Session().commit()
 
    return gr
 

	
 

	
 
class _BaseTestApi(object):
 
    REPO = None
 
    REPO_TYPE = None
 

	
 
    @classmethod
 
    def setup_class(cls):
 
        cls.usr = db.User.get_by_username(base.TEST_USER_ADMIN_LOGIN)
 
        cls.apikey = cls.usr.api_key
 
        cls.test_user = UserModel().create_or_update(
 
            username='test-api',
 
            password='test',
 
            email='test@example.com',
 
            firstname='first',
 
            lastname='last'
 
        )
 
        meta.Session().commit()
 
        cls.TEST_USER_LOGIN = cls.test_user.username
 
        cls.apikey_regular = cls.test_user.api_key
 

	
 
    @classmethod
 
    def teardown_class(cls):
 
        pass
 

	
 
    def setup_method(self, method):
 
        make_user_group()
 
        make_repo_group()
 

	
 
    def teardown_method(self, method):
 
        fixture.destroy_user_group(TEST_USER_GROUP)
 
        fixture.destroy_gists()
 
        fixture.destroy_repo_group(TEST_REPO_GROUP)
 

	
 
    def _compare_ok(self, id_, expected, given):
 
        expected = jsonify({
 
            'id': id_,
 
            'error': None,
 
            'result': expected
 
        })
 
        given = ext_json.loads(given)
 
        assert expected == given, (expected, given)
 

	
 
    def _compare_error(self, id_, expected, given):
 
        expected = jsonify({
 
            'id': id_,
 
            'error': expected,
 
            'result': None
 
        })
 
        given = ext_json.loads(given)
 
        assert expected == given, (expected, given)
 

	
 
    def test_api_wrong_key(self):
 
        id_, params = _build_data('trololo', 'get_user')
 
        response = api_call(self, params)
 

	
 
        expected = 'Invalid API key'
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_missing_non_optional_param(self):
 
        id_, params = _build_data(self.apikey, 'get_repo')
 
        response = api_call(self, params)
 

	
 
        expected = 'Missing non optional `repoid` arg in JSON DATA'
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_missing_non_optional_param_args_null(self):
 
        id_, params = _build_data(self.apikey, 'get_repo')
 
        params = params.replace('"args": {}', '"args": null')
 
        response = api_call(self, params)
 

	
 
        expected = 'Missing non optional `repoid` arg in JSON DATA'
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_missing_non_optional_param_args_bad(self):
 
        id_, params = _build_data(self.apikey, 'get_repo')
 
        params = params.replace('"args": {}', '"args": 1')
 
        response = api_call(self, params)
 

	
 
        expected = 'Missing non optional `repoid` arg in JSON DATA'
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_args_is_null(self):
 
        id_, params = _build_data(self.apikey, 'get_users', )
 
        params = params.replace('"args": {}', '"args": null')
 
        response = api_call(self, params)
 
        assert response.status == '200 OK'
 

	
 
    def test_api_args_is_bad(self):
 
        id_, params = _build_data(self.apikey, 'get_users', )
 
        params = params.replace('"args": {}', '"args": 1')
 
        response = api_call(self, params)
 
        assert response.status == '200 OK'
 

	
 
    def test_api_args_different_args(self):
 
        expected = {
 
            'ascii_letters': string.ascii_letters,
 
            'ws': string.whitespace,
 
            'printables': string.printable
 
        }
 
        id_, params = _build_data(self.apikey, 'test', args=expected)
 
        response = api_call(self, params)
 
        assert response.status == '200 OK'
 
        self._compare_ok(id_, expected, response.body)
 

	
 
    def test_api_get_users(self):
 
        id_, params = _build_data(self.apikey, 'get_users', )
 
        response = api_call(self, params)
 
        ret_all = []
 
        _users = db.User.query().filter_by(is_default_user=False) \
 
            .order_by(db.User.username).all()
 
        for usr in _users:
 
            ret = usr.get_api_data()
 
            ret_all.append(jsonify(ret))
 
        expected = ret_all
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    def test_api_get_user(self):
 
        id_, params = _build_data(self.apikey, 'get_user',
 
                                  userid=base.TEST_USER_ADMIN_LOGIN)
 
        response = api_call(self, params)
 

	
 
        usr = db.User.get_by_username(base.TEST_USER_ADMIN_LOGIN)
 
        ret = usr.get_api_data()
 
        ret['permissions'] = AuthUser(dbuser=usr).permissions
 

	
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    def test_api_get_user_that_does_not_exist(self):
 
        id_, params = _build_data(self.apikey, 'get_user',
 
                                  userid='trololo')
 
        response = api_call(self, params)
 

	
 
        expected = "user `%s` does not exist" % 'trololo'
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_get_user_without_giving_userid(self):
 
        id_, params = _build_data(self.apikey, 'get_user')
 
        response = api_call(self, params)
 

	
 
        usr = db.User.get_by_username(base.TEST_USER_ADMIN_LOGIN)
 
        ret = usr.get_api_data()
 
        ret['permissions'] = AuthUser(dbuser=usr).permissions
 

	
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    def test_api_get_user_without_giving_userid_non_admin(self):
 
        id_, params = _build_data(self.apikey_regular, 'get_user')
 
        response = api_call(self, params)
 

	
 
        usr = db.User.get_by_username(self.TEST_USER_LOGIN)
 
        ret = usr.get_api_data()
 
        ret['permissions'] = AuthUser(dbuser=usr).permissions
 

	
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    def test_api_get_user_with_giving_userid_non_admin(self):
 
        id_, params = _build_data(self.apikey_regular, 'get_user',
 
                                  userid=self.TEST_USER_LOGIN)
 
        response = api_call(self, params)
 

	
 
        expected = 'userid is not the same as your user'
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_pull_remote(self):
 
        # Note: pulling from local repos is a mis-feature - it will bypass access control
 
        # ... but ok, if the path already has been set in the database
 
        repo_name = 'test_pull'
 
        r = fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
 
        # hack around that clone_uri can't be set to to a local path
 
        # (as shown by test_api_create_repo_clone_uri_local)
 
        r.clone_uri = os.path.join(db.Ui.get_by_key('paths', '/').ui_value, self.REPO)
 
        meta.Session().commit()
 

	
 
        pre_cached_tip = [repo.get_api_data()['last_changeset']['short_id'] for repo in db.Repository.query().filter(db.Repository.repo_name == repo_name)]
 

	
 
        id_, params = _build_data(self.apikey, 'pull',
 
                                  repoid=repo_name,)
 
        response = api_call(self, params)
 

	
 
        expected = {'msg': 'Pulled from `%s`' % repo_name,
 
                    'repository': repo_name}
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
        post_cached_tip = [repo.get_api_data()['last_changeset']['short_id'] for repo in db.Repository.query().filter(db.Repository.repo_name == repo_name)]
 

	
 
        fixture.destroy_repo(repo_name)
 

	
 
        assert pre_cached_tip != post_cached_tip
 

	
 
    def test_api_pull_fork(self):
 
        fork_name = 'fork'
 
        fixture.create_fork(self.REPO, fork_name)
 
        id_, params = _build_data(self.apikey, 'pull',
 
                                  repoid=fork_name,)
 
        response = api_call(self, params)
 

	
 
        expected = {'msg': 'Pulled from `%s`' % fork_name,
 
                    'repository': fork_name}
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
        fixture.destroy_repo(fork_name)
 

	
 
    def test_api_pull_error_no_remote_no_fork(self):
 
        # should fail because no clone_uri is set
 
        id_, params = _build_data(self.apikey, 'pull',
 
                                  repoid=self.REPO, )
 
        response = api_call(self, params)
 

	
 
        expected = 'Unable to pull changes from `%s`' % self.REPO
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_pull_custom_remote(self):
 
        repo_name = 'test_pull_custom_remote'
 
        fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
 

	
 
        custom_remote_path = os.path.join(db.Ui.get_by_key('paths', '/').ui_value, self.REPO)
 

	
 
        id_, params = _build_data(self.apikey, 'pull',
 
                                  repoid=repo_name,
 
                                  clone_uri=custom_remote_path)
 
        response = api_call(self, params)
 

	
 
        expected = {'msg': 'Pulled from `%s`' % repo_name,
 
                    'repository': repo_name}
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
        fixture.destroy_repo(repo_name)
 

	
 
    def test_api_rescan_repos(self):
 
        id_, params = _build_data(self.apikey, 'rescan_repos')
 
        response = api_call(self, params)
 

	
 
        expected = {'added': [], 'removed': []}
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(ScmModel, 'repo_scan', raise_exception)
 
    def test_api_rescann_error(self):
 
        id_, params = _build_data(self.apikey, 'rescan_repos', )
 
        response = api_call(self, params)
 

	
 
        expected = 'Error occurred during rescan repositories action'
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_create_existing_user(self):
 
        id_, params = _build_data(self.apikey, 'create_user',
 
                                  username=base.TEST_USER_ADMIN_LOGIN,
 
                                  email='test@example.com',
 
                                  password='trololo')
 
        response = api_call(self, params)
 

	
 
        expected = "user `%s` already exist" % base.TEST_USER_ADMIN_LOGIN
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_create_user_with_existing_email(self):
 
        id_, params = _build_data(self.apikey, 'create_user',
 
                                  username=base.TEST_USER_ADMIN_LOGIN + 'new',
 
                                  email=base.TEST_USER_REGULAR_EMAIL,
 
                                  password='trololo')
 
        response = api_call(self, params)
 

	
 
        expected = "email `%s` already exist" % base.TEST_USER_REGULAR_EMAIL
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_create_user(self):
 
        username = 'test_new_api_user'
 
        email = username + "@example.com"
 

	
 
        id_, params = _build_data(self.apikey, 'create_user',
 
                                  username=username,
 
                                  email=email,
 
                                  password='trololo')
 
        response = api_call(self, params)
 

	
 
        usr = db.User.get_by_username(username)
 
        ret = dict(
 
            msg='created new user `%s`' % username,
 
            user=jsonify(usr.get_api_data())
 
        )
 

	
 
        try:
 
            expected = ret
 
            self._compare_ok(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_user(usr.user_id)
 

	
 
    def test_api_create_user_without_password(self):
 
        username = 'test_new_api_user_passwordless'
 
        email = username + "@example.com"
 

	
 
        id_, params = _build_data(self.apikey, 'create_user',
 
                                  username=username,
 
                                  email=email)
 
        response = api_call(self, params)
 

	
 
        usr = db.User.get_by_username(username)
 
        ret = dict(
 
            msg='created new user `%s`' % username,
 
            user=jsonify(usr.get_api_data())
 
        )
 
        try:
 
            expected = ret
 
            self._compare_ok(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_user(usr.user_id)
 

	
 
    def test_api_create_user_with_extern_name(self):
 
        username = 'test_new_api_user_passwordless'
 
        email = username + "@example.com"
 

	
 
        id_, params = _build_data(self.apikey, 'create_user',
 
                                  username=username,
 
                                  email=email, extern_name='internal')
 
        response = api_call(self, params)
 

	
 
        usr = db.User.get_by_username(username)
 
        ret = dict(
 
            msg='created new user `%s`' % username,
 
            user=jsonify(usr.get_api_data())
 
        )
 
        try:
 
            expected = ret
 
            self._compare_ok(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_user(usr.user_id)
 

	
 
    @mock.patch.object(UserModel, 'create_or_update', raise_exception)
 
    def test_api_create_user_when_exception_happened(self):
 

	
 
        username = 'test_new_api_user'
 
        email = username + "@example.com"
 

	
 
        id_, params = _build_data(self.apikey, 'create_user',
 
                                  username=username,
 
                                  email=email,
 
                                  password='trololo')
 
        response = api_call(self, params)
 
        expected = 'failed to create user `%s`' % username
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_delete_user(self):
 
        usr = UserModel().create_or_update(username='test_user',
 
                                           password='qweqwe',
 
                                           email='u232@example.com',
 
                                           firstname='u1', lastname='u1')
 
        meta.Session().commit()
 
        username = usr.username
 
        email = usr.email
 
        usr_id = usr.user_id
 
        ## DELETE THIS USER NOW
 

	
 
        id_, params = _build_data(self.apikey, 'delete_user',
 
                                  userid=username, )
 
        response = api_call(self, params)
 

	
 
        ret = {'msg': 'deleted user ID:%s %s' % (usr_id, username),
 
               'user': None}
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(UserModel, 'delete', raise_exception)
 
    def test_api_delete_user_when_exception_happened(self):
 
        usr = UserModel().create_or_update(username='test_user',
 
                                           password='qweqwe',
 
                                           email='u232@example.com',
 
                                           firstname='u1', lastname='u1')
 
        meta.Session().commit()
 
        username = usr.username
 

	
 
        id_, params = _build_data(self.apikey, 'delete_user',
 
                                  userid=username, )
 
        response = api_call(self, params)
 
        ret = 'failed to delete user ID:%s %s' % (usr.user_id,
 
                                                  usr.username)
 
        expected = ret
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @base.parametrize('name,expected', [
 
        ('firstname', 'new_username'),
 
        ('lastname', 'new_username'),
 
        ('email', 'new_username'),
 
        ('admin', True),
 
        ('admin', False),
 
        ('extern_type', 'ldap'),
 
        ('extern_type', None),
 
        ('extern_name', 'test'),
 
        ('extern_name', None),
 
        ('active', False),
 
        ('active', True),
 
        ('password', 'newpass'),
 
    ])
 
    def test_api_update_user(self, name, expected):
 
        usr = db.User.get_by_username(self.TEST_USER_LOGIN)
 
        kw = {name: expected,
 
              'userid': usr.user_id}
 
        id_, params = _build_data(self.apikey, 'update_user', **kw)
 
        response = api_call(self, params)
 

	
 
        ret = {
 
            'msg': 'updated user ID:%s %s' % (
 
                usr.user_id, self.TEST_USER_LOGIN),
 
            'user': jsonify(db.User \
 
                .get_by_username(self.TEST_USER_LOGIN) \
 
                .get_api_data())
 
        }
 

	
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    def test_api_update_user_no_changed_params(self):
 
        usr = db.User.get_by_username(base.TEST_USER_ADMIN_LOGIN)
 
        ret = jsonify(usr.get_api_data())
 
        id_, params = _build_data(self.apikey, 'update_user',
 
                                  userid=base.TEST_USER_ADMIN_LOGIN)
 

	
 
        response = api_call(self, params)
 
        ret = {
 
            'msg': 'updated user ID:%s %s' % (
 
                usr.user_id, base.TEST_USER_ADMIN_LOGIN),
 
            'user': ret
 
        }
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    def test_api_update_user_by_user_id(self):
 
        usr = db.User.get_by_username(base.TEST_USER_ADMIN_LOGIN)
 
        ret = jsonify(usr.get_api_data())
 
        id_, params = _build_data(self.apikey, 'update_user',
 
                                  userid=usr.user_id)
 

	
 
        response = api_call(self, params)
 
        ret = {
 
            'msg': 'updated user ID:%s %s' % (
 
                usr.user_id, base.TEST_USER_ADMIN_LOGIN),
 
            'user': ret
 
        }
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    def test_api_update_user_default_user(self):
 
        usr = db.User.get_default_user()
 
        id_, params = _build_data(self.apikey, 'update_user',
 
                                  userid=usr.user_id)
 

	
 
        response = api_call(self, params)
 
        expected = 'editing default user is forbidden'
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(UserModel, 'update_user', raise_exception)
 
    def test_api_update_user_when_exception_happens(self):
 
        usr = db.User.get_by_username(base.TEST_USER_ADMIN_LOGIN)
 
        ret = jsonify(usr.get_api_data())
 
        id_, params = _build_data(self.apikey, 'update_user',
 
                                  userid=usr.user_id)
 

	
 
        response = api_call(self, params)
 
        ret = 'failed to update user `%s`' % usr.user_id
 

	
 
        expected = ret
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_get_repo(self):
 
        new_group = 'some_new_group'
 
        make_user_group(new_group)
 
        RepoModel().grant_user_group_permission(repo=self.REPO,
 
                                                group_name=new_group,
 
                                                perm='repository.read')
 
        meta.Session().commit()
 
        id_, params = _build_data(self.apikey, 'get_repo',
 
                                  repoid=self.REPO)
 
        response = api_call(self, params)
 
        assert "tags" not in response.json['result']
 
        assert 'pull_requests' not in response.json['result']
 

	
 
        repo = RepoModel().get_by_repo_name(self.REPO)
 
        ret = repo.get_api_data()
 

	
 
        members = []
 
        followers = []
 
        for user in repo.repo_to_perm:
 
            perm = user.permission.permission_name
 
            user = user.user
 
            user_data = {'name': user.username, 'type': "user",
 
                         'permission': perm}
 
            members.append(user_data)
 

	
 
        for user_group in repo.users_group_to_perm:
 
            perm = user_group.permission.permission_name
 
            user_group = user_group.users_group
 
            user_group_data = {'name': user_group.users_group_name,
 
                               'type': "user_group", 'permission': perm}
 
            members.append(user_group_data)
 

	
 
        for user in repo.followers:
 
            followers.append(user.user.get_api_data())
 

	
 
        ret['members'] = members
 
        ret['followers'] = followers
 

	
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 
        fixture.destroy_user_group(new_group)
 

	
 
        id_, params = _build_data(self.apikey, 'get_repo', repoid=self.REPO,
 
                                  with_revision_names=True,
 
                                  with_pullrequests=True)
 
        response = api_call(self, params)
 
        assert "v0.2.0" in response.json['result']['tags']
 
        assert 'pull_requests' in response.json['result']
 

	
 
    @base.parametrize('grant_perm', [
 
        ('repository.admin'),
 
        ('repository.write'),
 
        ('repository.read'),
 
    ])
 
    def test_api_get_repo_by_non_admin(self, grant_perm):
 
        RepoModel().grant_user_permission(repo=self.REPO,
 
                                          user=self.TEST_USER_LOGIN,
 
                                          perm=grant_perm)
 
        meta.Session().commit()
 
        id_, params = _build_data(self.apikey_regular, 'get_repo',
 
                                  repoid=self.REPO)
 
        response = api_call(self, params)
 

	
 
        repo = RepoModel().get_by_repo_name(self.REPO)
 
        assert len(repo.repo_to_perm) >= 2  # make sure we actually are testing something - probably the default 2 permissions, possibly more
 

	
 
        expected = repo.get_api_data()
 

	
 
        members = []
 
        for user in repo.repo_to_perm:
 
            perm = user.permission.permission_name
 
            user_obj = user.user
 
            user_data = {'name': user_obj.username, 'type': "user",
 
                         'permission': perm}
 
            members.append(user_data)
 
        for user_group in repo.users_group_to_perm:
 
            perm = user_group.permission.permission_name
 
            user_group_obj = user_group.users_group
 
            user_group_data = {'name': user_group_obj.users_group_name,
 
                               'type': "user_group", 'permission': perm}
 
            members.append(user_group_data)
 
        expected['members'] = members
 

	
 
        followers = []
 

	
 
        for user in repo.followers:
 
            followers.append(user.user.get_api_data())
 

	
 
        expected['followers'] = followers
 

	
 
        try:
 
            self._compare_ok(id_, expected, given=response.body)
 
        finally:
 
            RepoModel().revoke_user_permission(self.REPO, self.TEST_USER_LOGIN)
 

	
 
    def test_api_get_repo_by_non_admin_no_permission_to_repo(self):
 
        RepoModel().grant_user_permission(repo=self.REPO,
 
                                          user=db.User.DEFAULT_USER_NAME,
 
                                          perm='repository.none')
 
        try:
 
            RepoModel().grant_user_permission(repo=self.REPO,
 
                                              user=self.TEST_USER_LOGIN,
 
                                              perm='repository.none')
 

	
 
            id_, params = _build_data(self.apikey_regular, 'get_repo',
 
                                      repoid=self.REPO)
 
            response = api_call(self, params)
 

	
 
            expected = 'repository `%s` does not exist' % (self.REPO)
 
            self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            RepoModel().grant_user_permission(repo=self.REPO,
 
                                              user=db.User.DEFAULT_USER_NAME,
 
                                              perm='repository.read')
 

	
 
    def test_api_get_repo_that_doesn_not_exist(self):
 
        id_, params = _build_data(self.apikey, 'get_repo',
 
                                  repoid='no-such-repo')
 
        response = api_call(self, params)
 

	
 
        ret = 'repository `%s` does not exist' % 'no-such-repo'
 
        expected = ret
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_get_repos(self):
 
        id_, params = _build_data(self.apikey, 'get_repos')
 
        response = api_call(self, params)
 

	
 
        expected = jsonify([
 
            repo.get_api_data()
 
            for repo in db.Repository.query()
 
        ])
 

	
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    def test_api_get_repos_non_admin(self):
 
        id_, params = _build_data(self.apikey_regular, 'get_repos')
 
        response = api_call(self, params)
 

	
 
        expected = jsonify([
 
            repo.get_api_data()
 
            for repo in AuthUser(dbuser=db.User.get_by_username(self.TEST_USER_LOGIN)).get_all_user_repos()
 
        ])
 

	
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    @base.parametrize('name,ret_type', [
 
        ('all', 'all'),
 
        ('dirs', 'dirs'),
 
        ('files', 'files'),
 
    ])
 
    def test_api_get_repo_nodes(self, name, ret_type):
 
        rev = 'tip'
 
        path = '/'
 
        id_, params = _build_data(self.apikey, 'get_repo_nodes',
 
                                  repoid=self.REPO, revision=rev,
 
                                  root_path=path,
 
                                  ret_type=ret_type)
 
        response = api_call(self, params)
 

	
 
        # we don't the actual return types here since it's tested somewhere
 
        # else
 
        expected = response.json['result']
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    def test_api_get_repo_nodes_bad_revisions(self):
 
        rev = 'i-dont-exist'
 
        path = '/'
 
        id_, params = _build_data(self.apikey, 'get_repo_nodes',
 
                                  repoid=self.REPO, revision=rev,
 
                                  root_path=path, )
 
        response = api_call(self, params)
 

	
 
        expected = 'failed to get repo: `%s` nodes' % self.REPO
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_get_repo_nodes_bad_path(self):
 
        rev = 'tip'
 
        path = '/idontexits'
 
        id_, params = _build_data(self.apikey, 'get_repo_nodes',
 
                                  repoid=self.REPO, revision=rev,
 
                                  root_path=path, )
 
        response = api_call(self, params)
 

	
 
        expected = 'failed to get repo: `%s` nodes' % self.REPO
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_get_repo_nodes_bad_ret_type(self):
 
        rev = 'tip'
 
        path = '/'
 
        ret_type = 'error'
 
        id_, params = _build_data(self.apikey, 'get_repo_nodes',
 
                                  repoid=self.REPO, revision=rev,
 
                                  root_path=path,
 
                                  ret_type=ret_type)
 
        response = api_call(self, params)
 

	
 
        expected = ('ret_type must be one of %s'
 
                    % (','.join(sorted(['files', 'dirs', 'all']))))
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @base.parametrize('name,ret_type,grant_perm', [
 
        ('all', 'all', 'repository.write'),
 
        ('dirs', 'dirs', 'repository.admin'),
 
        ('files', 'files', 'repository.read'),
 
    ])
 
    def test_api_get_repo_nodes_by_regular_user(self, name, ret_type, grant_perm):
 
        RepoModel().grant_user_permission(repo=self.REPO,
 
                                          user=self.TEST_USER_LOGIN,
 
                                          perm=grant_perm)
 
        meta.Session().commit()
 

	
 
        rev = 'tip'
 
        path = '/'
 
        id_, params = _build_data(self.apikey_regular, 'get_repo_nodes',
 
                                  repoid=self.REPO, revision=rev,
 
                                  root_path=path,
 
                                  ret_type=ret_type)
 
        response = api_call(self, params)
 

	
 
        # we don't the actual return types here since it's tested somewhere
 
        # else
 
        expected = response.json['result']
 
        try:
 
            self._compare_ok(id_, expected, given=response.body)
 
        finally:
 
            RepoModel().revoke_user_permission(self.REPO, self.TEST_USER_LOGIN)
 

	
 
    def test_api_create_repo(self):
 
        repo_name = 'api-repo'
 
        id_, params = _build_data(self.apikey, 'create_repo',
 
                                  repo_name=repo_name,
 
                                  owner=base.TEST_USER_ADMIN_LOGIN,
 
                                  repo_type=self.REPO_TYPE,
 
        )
 
        response = api_call(self, params)
 

	
 
        repo = RepoModel().get_by_repo_name(repo_name)
 
        assert repo is not None
 
        ret = {
 
            'msg': 'Created new repository `%s`' % repo_name,
 
            'success': True,
 
            'task': None,
 
        }
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 
        fixture.destroy_repo(repo_name)
 

	
 
    @base.parametrize('repo_name', [
 
        '',
 
        '.',
 
        '..',
 
        ':',
 
        '/',
 
        '<test>',
 
    ])
 
    def test_api_create_repo_bad_names(self, repo_name):
 
        id_, params = _build_data(self.apikey, 'create_repo',
 
                                  repo_name=repo_name,
 
                                  owner=base.TEST_USER_ADMIN_LOGIN,
 
                                  repo_type=self.REPO_TYPE,
 
        )
 
        response = api_call(self, params)
 
        if repo_name == '/':
 
            expected = "repo group `` not found"
 
            self._compare_error(id_, expected, given=response.body)
 
        else:
 
            expected = "failed to create repository `%s`" % repo_name
 
            self._compare_error(id_, expected, given=response.body)
 
        fixture.destroy_repo(repo_name)
 

	
 
    def test_api_create_repo_clone_uri_local(self):
 
        # cloning from local repos was a mis-feature - it would bypass access control
 
        # TODO: introduce other test coverage of actual remote cloning
 
        clone_uri = os.path.join(base.TESTS_TMP_PATH, self.REPO)
 
        repo_name = 'api-repo'
 
        id_, params = _build_data(self.apikey, 'create_repo',
 
                                  repo_name=repo_name,
 
                                  owner=base.TEST_USER_ADMIN_LOGIN,
 
                                  repo_type=self.REPO_TYPE,
 
                                  clone_uri=clone_uri,
 
        )
 
        response = api_call(self, params)
 
        expected = "failed to create repository `%s`" % repo_name
 
        self._compare_error(id_, expected, given=response.body)
 
        fixture.destroy_repo(repo_name)
 

	
 
    def test_api_create_repo_and_repo_group(self):
 
        repo_group_name = 'my_gr'
 
        repo_name = '%s/api-repo' % repo_group_name
 

	
 
        # repo creation can no longer also create repo group
 
        id_, params = _build_data(self.apikey, 'create_repo',
 
                                  repo_name=repo_name,
 
                                  owner=base.TEST_USER_ADMIN_LOGIN,
 
                                  repo_type=self.REPO_TYPE,)
 
        response = api_call(self, params)
 
        expected = 'repo group `%s` not found' % repo_group_name
 
        self._compare_error(id_, expected, given=response.body)
 
        assert RepoModel().get_by_repo_name(repo_name) is None
 

	
 
        # create group before creating repo
 
        rg = fixture.create_repo_group(repo_group_name)
 
        meta.Session().commit()
 

	
 
        id_, params = _build_data(self.apikey, 'create_repo',
 
                                  repo_name=repo_name,
 
                                  owner=base.TEST_USER_ADMIN_LOGIN,
 
                                  repo_type=self.REPO_TYPE,)
 
        response = api_call(self, params)
 
        expected = {
 
            'msg': 'Created new repository `%s`' % repo_name,
 
            'success': True,
 
            'task': None,
 
        }
 
        self._compare_ok(id_, expected, given=response.body)
 
        repo = RepoModel().get_by_repo_name(repo_name)
 
        assert repo is not None
 

	
 
        fixture.destroy_repo(repo_name)
 
        fixture.destroy_repo_group(repo_group_name)
 

	
 
    def test_api_create_repo_in_repo_group_without_permission(self):
 
        repo_group_basename = 'api-repo-repo'
 
        repo_group_name = '%s/%s' % (TEST_REPO_GROUP, repo_group_basename)
 
        repo_name = '%s/api-repo' % repo_group_name
 

	
 
        top_group = db.RepoGroup.get_by_group_name(TEST_REPO_GROUP)
 
        assert top_group
 
        rg = fixture.create_repo_group(repo_group_basename, parent_group_id=top_group)
 
        meta.Session().commit()
 
        RepoGroupModel().grant_user_permission(repo_group_name,
 
                                               self.TEST_USER_LOGIN,
 
                                               'group.none')
 
        meta.Session().commit()
 

	
 
        id_, params = _build_data(self.apikey_regular, 'create_repo',
 
                                  repo_name=repo_name,
 
                                  repo_type=self.REPO_TYPE,
 
        )
 
        response = api_call(self, params)
 

	
 
        # Current result when API access control is different from Web:
 
        ret = {
 
            'msg': 'Created new repository `%s`' % repo_name,
 
            'success': True,
 
            'task': None,
 
        }
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 
        fixture.destroy_repo(repo_name)
 

	
 
        # Expected and arguably more correct result:
 
        #expected = 'failed to create repository `%s`' % repo_name
 
        #self._compare_error(id_, expected, given=response.body)
 

	
 
        fixture.destroy_repo_group(repo_group_name)
 

	
 
    def test_api_create_repo_unknown_owner(self):
 
        repo_name = 'api-repo'
 
        owner = 'i-dont-exist'
 
        id_, params = _build_data(self.apikey, 'create_repo',
 
                                  repo_name=repo_name,
 
                                  owner=owner,
 
                                  repo_type=self.REPO_TYPE,
 
        )
 
        response = api_call(self, params)
 
        expected = 'user `%s` does not exist' % owner
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_create_repo_dont_specify_owner(self):
 
        repo_name = 'api-repo'
 
        owner = 'i-dont-exist'
 
        id_, params = _build_data(self.apikey, 'create_repo',
 
                                  repo_name=repo_name,
 
                                  repo_type=self.REPO_TYPE,
 
        )
 
        response = api_call(self, params)
 

	
 
        repo = RepoModel().get_by_repo_name(repo_name)
 
        assert repo is not None
 
        ret = {
 
            'msg': 'Created new repository `%s`' % repo_name,
 
            'success': True,
 
            'task': None,
 
        }
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 
        fixture.destroy_repo(repo_name)
 

	
 
    def test_api_create_repo_by_non_admin(self):
 
        repo_name = 'api-repo'
 
        owner = 'i-dont-exist'
 
        id_, params = _build_data(self.apikey_regular, 'create_repo',
 
                                  repo_name=repo_name,
 
                                  repo_type=self.REPO_TYPE,
 
        )
 
        response = api_call(self, params)
 

	
 
        repo = RepoModel().get_by_repo_name(repo_name)
 
        assert repo is not None
 
        ret = {
 
            'msg': 'Created new repository `%s`' % repo_name,
 
            'success': True,
 
            'task': None,
 
        }
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 
        fixture.destroy_repo(repo_name)
 

	
 
    def test_api_create_repo_by_non_admin_specify_owner(self):
 
        repo_name = 'api-repo'
 
        owner = 'i-dont-exist'
 
        id_, params = _build_data(self.apikey_regular, 'create_repo',
 
                                  repo_name=repo_name,
 
                                  repo_type=self.REPO_TYPE,
 
                                  owner=owner)
 
        response = api_call(self, params)
 

	
 
        expected = 'Only Kallithea admin can specify `owner` param'
 
        self._compare_error(id_, expected, given=response.body)
 
        fixture.destroy_repo(repo_name)
 

	
 
    def test_api_create_repo_exists(self):
 
        repo_name = self.REPO
 
        id_, params = _build_data(self.apikey, 'create_repo',
 
                                  repo_name=repo_name,
 
                                  owner=base.TEST_USER_ADMIN_LOGIN,
 
                                  repo_type=self.REPO_TYPE,)
 
        response = api_call(self, params)
 
        expected = "repo `%s` already exist" % repo_name
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_create_repo_dot_dot(self):
 
        # it is only possible to create repositories in existing repo groups - and '..' can't be used
 
        group_name = '%s/..' % TEST_REPO_GROUP
 
        repo_name = '%s/%s' % (group_name, 'could-be-outside')
 
        id_, params = _build_data(self.apikey, 'create_repo',
 
                                  repo_name=repo_name,
 
                                  owner=base.TEST_USER_ADMIN_LOGIN,
 
                                  repo_type=self.REPO_TYPE,)
 
        response = api_call(self, params)
 
        expected = 'repo group `%s` not found' % group_name
 
        self._compare_error(id_, expected, given=response.body)
 
        fixture.destroy_repo(repo_name)
 

	
 
    @mock.patch.object(RepoModel, 'create', raise_exception)
 
    def test_api_create_repo_exception_occurred(self):
 
        repo_name = 'api-repo'
 
        id_, params = _build_data(self.apikey, 'create_repo',
 
                                  repo_name=repo_name,
 
                                  owner=base.TEST_USER_ADMIN_LOGIN,
 
                                  repo_type=self.REPO_TYPE,)
 
        response = api_call(self, params)
 
        expected = 'failed to create repository `%s`' % repo_name
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @base.parametrize('changing_attr,updates', [
 
        ('owner', {'owner': base.TEST_USER_REGULAR_LOGIN}),
 
        ('description', {'description': 'new description'}),
 
        ('clone_uri', {'clone_uri': 'http://example.com/repo'}), # will fail - pulling from non-existing repo should fail
 
        ('clone_uri', {'clone_uri': '/repo'}), # will fail - pulling from local repo was a mis-feature - it would bypass access control
 
        ('clone_uri', {'clone_uri': None}),
 
        ('landing_rev', {'landing_rev': 'branch:master'}),
 
        ('enable_statistics', {'enable_statistics': True}),
 
        ('enable_downloads', {'enable_downloads': True}),
 
        ('name', {'name': 'new_repo_name'}),
 
        ('repo_group', {'group': 'test_group_for_update'}),
 
    ])
 
    def test_api_update_repo(self, changing_attr, updates):
 
        repo_name = 'api_update_me'
 
        repo = fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
 
        if changing_attr == 'repo_group':
 
            fixture.create_repo_group(updates['group'])
 

	
 
        id_, params = _build_data(self.apikey, 'update_repo',
 
                                  repoid=repo_name, **updates)
 
        response = api_call(self, params)
 
        if changing_attr == 'name':
 
            repo_name = updates['name']
 
        if changing_attr == 'repo_group':
 
            repo_name = '/'.join([updates['group'], repo_name])
 
        try:
 
            if changing_attr == 'clone_uri' and updates['clone_uri']:
 
                expected = 'failed to update repo `%s`' % repo_name
 
                self._compare_error(id_, expected, given=response.body)
 
            else:
 
                expected = {
 
                    'msg': 'updated repo ID:%s %s' % (repo.repo_id, repo_name),
 
                    'repository': repo.get_api_data()
 
                }
 
                self._compare_ok(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_repo(repo_name)
 
            if changing_attr == 'repo_group':
 
                fixture.destroy_repo_group(updates['group'])
 

	
 
    @base.parametrize('changing_attr,updates', [
 
        ('owner', {'owner': base.TEST_USER_REGULAR_LOGIN}),
 
        ('description', {'description': 'new description'}),
 
        ('clone_uri', {'clone_uri': 'http://example.com/repo'}), # will fail - pulling from non-existing repo should fail
 
        ('clone_uri', {'clone_uri': '/repo'}), # will fail - pulling from local repo was a mis-feature - it would bypass access control
 
        ('clone_uri', {'clone_uri': None}),
 
        ('landing_rev', {'landing_rev': 'branch:master'}),
 
        ('enable_statistics', {'enable_statistics': True}),
 
        ('enable_downloads', {'enable_downloads': True}),
 
        ('name', {'name': 'new_repo_name'}),
 
        ('repo_group', {'group': 'test_group_for_update'}),
 
    ])
 
    def test_api_update_group_repo(self, changing_attr, updates):
 
        group_name = 'lololo'
 
        fixture.create_repo_group(group_name)
 
        repo_name = '%s/api_update_me' % group_name
 
        repo = fixture.create_repo(repo_name, repo_group=group_name, repo_type=self.REPO_TYPE)
 
        if changing_attr == 'repo_group':
 
            fixture.create_repo_group(updates['group'])
 

	
 
        id_, params = _build_data(self.apikey, 'update_repo',
 
                                  repoid=repo_name, **updates)
 
        response = api_call(self, params)
 
        if changing_attr == 'name':
 
            repo_name = '%s/%s' % (group_name, updates['name'])
 
        if changing_attr == 'repo_group':
 
            repo_name = '/'.join([updates['group'], repo_name.rsplit('/', 1)[-1]])
 
        try:
 
            if changing_attr == 'clone_uri' and updates['clone_uri']:
 
                expected = 'failed to update repo `%s`' % repo_name
 
                self._compare_error(id_, expected, given=response.body)
 
            else:
 
                expected = {
 
                    'msg': 'updated repo ID:%s %s' % (repo.repo_id, repo_name),
 
                    'repository': repo.get_api_data()
 
                }
 
                self._compare_ok(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_repo(repo_name)
 
            if changing_attr == 'repo_group':
 
                fixture.destroy_repo_group(updates['group'])
 
        fixture.destroy_repo_group(group_name)
 

	
 
    def test_api_update_repo_repo_group_does_not_exist(self):
 
        repo_name = 'admin_owned'
 
        fixture.create_repo(repo_name)
 
        updates = {'group': 'test_group_for_update'}
 
        id_, params = _build_data(self.apikey, 'update_repo',
 
                                  repoid=repo_name, **updates)
 
        response = api_call(self, params)
 
        try:
 
            expected = 'repository group `%s` does not exist' % updates['group']
 
            self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_repo(repo_name)
 

	
 
    def test_api_update_repo_regular_user_not_allowed(self):
 
        repo_name = 'admin_owned'
 
        fixture.create_repo(repo_name)
 
        updates = {'description': 'something else'}
 
        id_, params = _build_data(self.apikey_regular, 'update_repo',
 
                                  repoid=repo_name, **updates)
 
        response = api_call(self, params)
 
        try:
 
            expected = 'repository `%s` does not exist' % repo_name
 
            self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_repo(repo_name)
 

	
 
    @mock.patch.object(RepoModel, 'update', raise_exception)
 
    def test_api_update_repo_exception_occurred(self):
 
        repo_name = 'api_update_me'
 
        fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
 
        id_, params = _build_data(self.apikey, 'update_repo',
 
                                  repoid=repo_name, owner=base.TEST_USER_ADMIN_LOGIN,)
 
        response = api_call(self, params)
 
        try:
 
            expected = 'failed to update repo `%s`' % repo_name
 
            self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_repo(repo_name)
 

	
 
    def test_api_update_repo_regular_user_change_repo_name(self):
 
        repo_name = 'admin_owned'
 
        new_repo_name = 'new_repo_name'
 
        fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
 
        RepoModel().grant_user_permission(repo=repo_name,
 
                                          user=self.TEST_USER_LOGIN,
 
                                          perm='repository.admin')
 
        UserModel().revoke_perm('default', 'hg.create.repository')
 
        UserModel().grant_perm('default', 'hg.create.none')
 
        updates = {'name': new_repo_name}
 
        id_, params = _build_data(self.apikey_regular, 'update_repo',
 
                                  repoid=repo_name, **updates)
 
        response = api_call(self, params)
 
        try:
 
            expected = 'no permission to create (or move) repositories'
 
            self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_repo(repo_name)
 
            fixture.destroy_repo(new_repo_name)
 

	
 
    def test_api_update_repo_regular_user_change_repo_name_allowed(self):
 
        repo_name = 'admin_owned'
 
        new_repo_name = 'new_repo_name'
 
        repo = fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
 
        RepoModel().grant_user_permission(repo=repo_name,
 
                                          user=self.TEST_USER_LOGIN,
 
                                          perm='repository.admin')
 
        UserModel().revoke_perm('default', 'hg.create.none')
 
        UserModel().grant_perm('default', 'hg.create.repository')
 
        updates = {'name': new_repo_name}
 
        id_, params = _build_data(self.apikey_regular, 'update_repo',
 
                                  repoid=repo_name, **updates)
 
        response = api_call(self, params)
 
        try:
 
            expected = {
 
                'msg': 'updated repo ID:%s %s' % (repo.repo_id, new_repo_name),
 
                'repository': repo.get_api_data()
 
            }
 
            self._compare_ok(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_repo(repo_name)
 
            fixture.destroy_repo(new_repo_name)
 

	
 
    def test_api_update_repo_regular_user_change_owner(self):
 
        repo_name = 'admin_owned'
 
        fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
 
        RepoModel().grant_user_permission(repo=repo_name,
 
                                          user=self.TEST_USER_LOGIN,
 
                                          perm='repository.admin')
 
        updates = {'owner': base.TEST_USER_ADMIN_LOGIN}
 
        id_, params = _build_data(self.apikey_regular, 'update_repo',
 
                                  repoid=repo_name, **updates)
 
        response = api_call(self, params)
 
        try:
 
            expected = 'Only Kallithea admin can specify `owner` param'
 
            self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_repo(repo_name)
 

	
 
    def test_api_delete_repo(self):
 
        repo_name = 'api_delete_me'
 
        fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
 

	
 
        id_, params = _build_data(self.apikey, 'delete_repo',
 
                                  repoid=repo_name, )
 
        response = api_call(self, params)
 

	
 
        ret = {
 
            'msg': 'Deleted repository `%s`' % repo_name,
 
            'success': True
 
        }
 
        try:
 
            expected = ret
 
            self._compare_ok(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_repo(repo_name)
 

	
 
    def test_api_delete_repo_by_non_admin(self):
 
        repo_name = 'api_delete_me'
 
        fixture.create_repo(repo_name, repo_type=self.REPO_TYPE,
 
                            cur_user=self.TEST_USER_LOGIN)
 
        id_, params = _build_data(self.apikey_regular, 'delete_repo',
 
                                  repoid=repo_name, )
 
        response = api_call(self, params)
 

	
 
        ret = {
 
            'msg': 'Deleted repository `%s`' % repo_name,
 
            'success': True
 
        }
 
        try:
 
            expected = ret
 
            self._compare_ok(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_repo(repo_name)
 

	
 
    def test_api_delete_repo_by_non_admin_no_permission(self):
 
        repo_name = 'api_delete_me'
 
        fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
 
        try:
 
            id_, params = _build_data(self.apikey_regular, 'delete_repo',
 
                                      repoid=repo_name, )
 
            response = api_call(self, params)
 
            expected = 'repository `%s` does not exist' % (repo_name)
 
            self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_repo(repo_name)
 

	
 
    def test_api_delete_repo_exception_occurred(self):
 
        repo_name = 'api_delete_me'
 
        fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
 
        try:
 
            with mock.patch.object(RepoModel, 'delete', raise_exception):
 
                id_, params = _build_data(self.apikey, 'delete_repo',
 
                                          repoid=repo_name, )
 
                response = api_call(self, params)
 

	
 
                expected = 'failed to delete repository `%s`' % repo_name
 
                self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_repo(repo_name)
 

	
 
    def test_api_fork_repo(self):
 
        fork_name = 'api-repo-fork'
 
        id_, params = _build_data(self.apikey, 'fork_repo',
 
                                  repoid=self.REPO,
 
                                  fork_name=fork_name,
 
                                  owner=base.TEST_USER_ADMIN_LOGIN,
 
        )
 
        response = api_call(self, params)
 

	
 
        ret = {
 
            'msg': 'Created fork of `%s` as `%s`' % (self.REPO,
 
                                                     fork_name),
 
            'success': True,
 
            'task': None,
 
        }
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 
        fixture.destroy_repo(fork_name)
 

	
 
    @base.parametrize('fork_name', [
 
        'api-repo-fork',
 
        '%s/api-repo-fork' % TEST_REPO_GROUP,
 
    ])
 
    def test_api_fork_repo_non_admin(self, fork_name):
 
        id_, params = _build_data(self.apikey_regular, 'fork_repo',
 
                                  repoid=self.REPO,
 
                                  fork_name=fork_name,
 
        )
 
        response = api_call(self, params)
 

	
 
        ret = {
 
            'msg': 'Created fork of `%s` as `%s`' % (self.REPO,
 
                                                     fork_name),
 
            'success': True,
 
            'task': None,
 
        }
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 
        fixture.destroy_repo(fork_name)
 

	
 
    def test_api_fork_repo_non_admin_specify_owner(self):
 
        fork_name = 'api-repo-fork'
 
        id_, params = _build_data(self.apikey_regular, 'fork_repo',
 
                                  repoid=self.REPO,
 
                                  fork_name=fork_name,
 
                                  owner=base.TEST_USER_ADMIN_LOGIN,
 
        )
 
        response = api_call(self, params)
 
        expected = 'Only Kallithea admin can specify `owner` param'
 
        self._compare_error(id_, expected, given=response.body)
 
        fixture.destroy_repo(fork_name)
 

	
 
    def test_api_fork_repo_non_admin_no_permission_to_fork(self):
 
        RepoModel().grant_user_permission(repo=self.REPO,
 
                                          user=db.User.DEFAULT_USER_NAME,
 
                                          perm='repository.none')
 
        try:
 
            fork_name = 'api-repo-fork'
 
            id_, params = _build_data(self.apikey_regular, 'fork_repo',
 
                                      repoid=self.REPO,
 
                                      fork_name=fork_name,
 
            )
 
            response = api_call(self, params)
 
            expected = 'repository `%s` does not exist' % (self.REPO)
 
            self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            RepoModel().grant_user_permission(repo=self.REPO,
 
                                              user=db.User.DEFAULT_USER_NAME,
 
                                              perm='repository.read')
 
            fixture.destroy_repo(fork_name)
 

	
 
    @base.parametrize('name,perm', [
 
        ('read', 'repository.read'),
 
        ('write', 'repository.write'),
 
        ('admin', 'repository.admin'),
 
    ])
 
    def test_api_fork_repo_non_admin_no_create_repo_permission(self, name, perm):
 
        fork_name = 'api-repo-fork'
 
        # regardless of base repository permission, forking is disallowed
 
        # when repository creation is disabled
 
        RepoModel().grant_user_permission(repo=self.REPO,
 
                                          user=self.TEST_USER_LOGIN,
 
                                          perm=perm)
 
        UserModel().revoke_perm('default', 'hg.create.repository')
 
        UserModel().grant_perm('default', 'hg.create.none')
 
        id_, params = _build_data(self.apikey_regular, 'fork_repo',
 
                                  repoid=self.REPO,
 
                                  fork_name=fork_name,
 
        )
 
        response = api_call(self, params)
 
        expected = 'no permission to create repositories'
 
        self._compare_error(id_, expected, given=response.body)
 
        fixture.destroy_repo(fork_name)
 

	
 
    def test_api_fork_repo_unknown_owner(self):
 
        fork_name = 'api-repo-fork'
 
        owner = 'i-dont-exist'
 
        id_, params = _build_data(self.apikey, 'fork_repo',
 
                                  repoid=self.REPO,
 
                                  fork_name=fork_name,
 
                                  owner=owner,
 
        )
 
        response = api_call(self, params)
 
        expected = 'user `%s` does not exist' % owner
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_fork_repo_fork_exists(self):
 
        fork_name = 'api-repo-fork'
 
        fixture.create_fork(self.REPO, fork_name)
 

	
 
        try:
 
            fork_name = 'api-repo-fork'
 

	
 
            id_, params = _build_data(self.apikey, 'fork_repo',
 
                                      repoid=self.REPO,
 
                                      fork_name=fork_name,
 
                                      owner=base.TEST_USER_ADMIN_LOGIN,
 
            )
 
            response = api_call(self, params)
 

	
 
            expected = "fork `%s` already exist" % fork_name
 
            self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_repo(fork_name)
 

	
 
    def test_api_fork_repo_repo_exists(self):
 
        fork_name = self.REPO
 

	
 
        id_, params = _build_data(self.apikey, 'fork_repo',
 
                                  repoid=self.REPO,
 
                                  fork_name=fork_name,
 
                                  owner=base.TEST_USER_ADMIN_LOGIN,
 
        )
 
        response = api_call(self, params)
 

	
 
        expected = "repo `%s` already exist" % fork_name
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(RepoModel, 'create_fork', raise_exception)
 
    def test_api_fork_repo_exception_occurred(self):
 
        fork_name = 'api-repo-fork'
 
        id_, params = _build_data(self.apikey, 'fork_repo',
 
                                  repoid=self.REPO,
 
                                  fork_name=fork_name,
 
                                  owner=base.TEST_USER_ADMIN_LOGIN,
 
        )
 
        response = api_call(self, params)
 

	
 
        expected = 'failed to fork repository `%s` as `%s`' % (self.REPO,
 
                                                               fork_name)
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_get_user_group(self):
 
        id_, params = _build_data(self.apikey, 'get_user_group',
 
                                  usergroupid=TEST_USER_GROUP)
 
        response = api_call(self, params)
 

	
 
        user_group = UserGroupModel().get_group(TEST_USER_GROUP)
 
        members = []
 
        for user in user_group.members:
 
            user = user.user
 
            members.append(user.get_api_data())
 

	
 
        ret = user_group.get_api_data()
 
        ret['members'] = members
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    def test_api_get_user_groups(self):
 
        gr_name = 'test_user_group2'
 
        make_user_group(gr_name)
 

	
 
        try:
 
            id_, params = _build_data(self.apikey, 'get_user_groups', )
 
            response = api_call(self, params)
 

	
 
            expected = []
 
            for gr_name in [TEST_USER_GROUP, 'test_user_group2']:
 
                user_group = UserGroupModel().get_group(gr_name)
 
                ret = user_group.get_api_data()
 
                expected.append(ret)
 
            self._compare_ok(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_user_group(gr_name)
 

	
 
    def test_api_create_user_group(self):
 
        group_name = 'some_new_group'
 
        id_, params = _build_data(self.apikey, 'create_user_group',
 
                                  group_name=group_name)
 
        response = api_call(self, params)
 

	
 
        ret = {
 
            'msg': 'created new user group `%s`' % group_name,
 
            'user_group': jsonify(UserGroupModel() \
 
                .get_by_name(group_name) \
 
                .get_api_data())
 
        }
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
        fixture.destroy_user_group(group_name)
 

	
 
    def test_api_get_user_group_that_exist(self):
 
        id_, params = _build_data(self.apikey, 'create_user_group',
 
                                  group_name=TEST_USER_GROUP)
 
        response = api_call(self, params)
 

	
 
        expected = "user group `%s` already exist" % TEST_USER_GROUP
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(UserGroupModel, 'create', raise_exception)
 
    def test_api_get_user_group_exception_occurred(self):
 
        group_name = 'exception_happens'
 
        id_, params = _build_data(self.apikey, 'create_user_group',
 
                                  group_name=group_name)
 
        response = api_call(self, params)
 

	
 
        expected = 'failed to create group `%s`' % group_name
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @base.parametrize('changing_attr,updates', [
 
        ('group_name', {'group_name': 'new_group_name'}),
 
        ('group_name', {'group_name': 'test_group_for_update'}),
 
        ('owner', {'owner': base.TEST_USER_REGULAR_LOGIN}),
 
        ('active', {'active': False}),
 
        ('active', {'active': True}),
 
    ])
 
    def test_api_update_user_group(self, changing_attr, updates):
 
        gr_name = 'test_group_for_update'
 
        user_group = fixture.create_user_group(gr_name)
 
        try:
 
            id_, params = _build_data(self.apikey, 'update_user_group',
 
                                      usergroupid=gr_name, **updates)
 
            response = api_call(self, params)
 
            expected = {
 
               'msg': 'updated user group ID:%s %s' % (user_group.users_group_id,
 
                                                     user_group.users_group_name),
 
               'user_group': user_group.get_api_data()
 
            }
 
            self._compare_ok(id_, expected, given=response.body)
 
        finally:
 
            if changing_attr == 'group_name':
 
                # switch to updated name for proper cleanup
 
                gr_name = updates['group_name']
 
            fixture.destroy_user_group(gr_name)
 

	
 
    @mock.patch.object(UserGroupModel, 'update', raise_exception)
 
    def test_api_update_user_group_exception_occurred(self):
 
        gr_name = 'test_group'
 
        fixture.create_user_group(gr_name)
 
        try:
 
            id_, params = _build_data(self.apikey, 'update_user_group',
 
                                      usergroupid=gr_name)
 
            response = api_call(self, params)
 
            expected = 'failed to update user group `%s`' % gr_name
 
            self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_user_group(gr_name)
 

	
 
    def test_api_add_user_to_user_group(self):
 
        gr_name = 'test_group'
 
        fixture.create_user_group(gr_name)
 
        try:
 
            id_, params = _build_data(self.apikey, 'add_user_to_user_group',
 
                                      usergroupid=gr_name,
 
                                      userid=base.TEST_USER_ADMIN_LOGIN)
 
            response = api_call(self, params)
 
            expected = {
 
            'msg': 'added member `%s` to user group `%s`' % (
 
                    base.TEST_USER_ADMIN_LOGIN, gr_name),
 
            'success': True
 
            }
 
            self._compare_ok(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_user_group(gr_name)
 

	
 
    def test_api_add_user_to_user_group_that_doesnt_exist(self):
 
        id_, params = _build_data(self.apikey, 'add_user_to_user_group',
 
                                  usergroupid='false-group',
 
                                  userid=base.TEST_USER_ADMIN_LOGIN)
 
        response = api_call(self, params)
 

	
 
        expected = 'user group `%s` does not exist' % 'false-group'
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(UserGroupModel, 'add_user_to_group', raise_exception)
 
    def test_api_add_user_to_user_group_exception_occurred(self):
 
        gr_name = 'test_group'
 
        fixture.create_user_group(gr_name)
 
        try:
 
            id_, params = _build_data(self.apikey, 'add_user_to_user_group',
 
                                      usergroupid=gr_name,
 
                                      userid=base.TEST_USER_ADMIN_LOGIN)
 
            response = api_call(self, params)
 
            expected = 'failed to add member to user group `%s`' % gr_name
 
            self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_user_group(gr_name)
 

	
 
    def test_api_remove_user_from_user_group(self):
 
        gr_name = 'test_group_3'
 
        gr = fixture.create_user_group(gr_name)
 
        UserGroupModel().add_user_to_group(gr, user=base.TEST_USER_ADMIN_LOGIN)
 
        meta.Session().commit()
 
        try:
 
            id_, params = _build_data(self.apikey, 'remove_user_from_user_group',
 
                                      usergroupid=gr_name,
 
                                      userid=base.TEST_USER_ADMIN_LOGIN)
 
            response = api_call(self, params)
 
            expected = {
 
                'msg': 'removed member `%s` from user group `%s`' % (
 
                    base.TEST_USER_ADMIN_LOGIN, gr_name
 
                ),
 
                'success': True}
 
            self._compare_ok(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_user_group(gr_name)
 

	
 
    @mock.patch.object(UserGroupModel, 'remove_user_from_group', raise_exception)
 
    def test_api_remove_user_from_user_group_exception_occurred(self):
 
        gr_name = 'test_group_3'
 
        gr = fixture.create_user_group(gr_name)
 
        UserGroupModel().add_user_to_group(gr, user=base.TEST_USER_ADMIN_LOGIN)
 
        meta.Session().commit()
 
        try:
 
            id_, params = _build_data(self.apikey, 'remove_user_from_user_group',
 
                                      usergroupid=gr_name,
 
                                      userid=base.TEST_USER_ADMIN_LOGIN)
 
            response = api_call(self, params)
 
            expected = 'failed to remove member from user group `%s`' % gr_name
 
            self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_user_group(gr_name)
 

	
 
    def test_api_delete_user_group(self):
 
        gr_name = 'test_group'
 
        ugroup = fixture.create_user_group(gr_name)
 
        gr_id = ugroup.users_group_id
 
        try:
 
            id_, params = _build_data(self.apikey, 'delete_user_group',
 
                                      usergroupid=gr_name)
 
            response = api_call(self, params)
 
            expected = {
 
                'user_group': None,
 
                'msg': 'deleted user group ID:%s %s' % (gr_id, gr_name)
 
            }
 
            self._compare_ok(id_, expected, given=response.body)
 
        finally:
 
            if UserGroupModel().get_by_name(gr_name):
 
                fixture.destroy_user_group(gr_name)
 

	
 
    def test_api_delete_user_group_that_is_assigned(self):
 
        gr_name = 'test_group'
 
        ugroup = fixture.create_user_group(gr_name)
 
        gr_id = ugroup.users_group_id
 

	
 
        ugr_to_perm = RepoModel().grant_user_group_permission(self.REPO, gr_name, 'repository.write')
 
        msg = 'User Group assigned to %s' % ugr_to_perm.repository.repo_name
 

	
 
        try:
 
            id_, params = _build_data(self.apikey, 'delete_user_group',
 
                                      usergroupid=gr_name)
 
            response = api_call(self, params)
 
            expected = msg
 
            self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            if UserGroupModel().get_by_name(gr_name):
 
                fixture.destroy_user_group(gr_name)
 

	
 
    def test_api_delete_user_group_exception_occurred(self):
 
        gr_name = 'test_group'
 
        ugroup = fixture.create_user_group(gr_name)
 
        gr_id = ugroup.users_group_id
 
        id_, params = _build_data(self.apikey, 'delete_user_group',
 
                                  usergroupid=gr_name)
 

	
 
        try:
 
            with mock.patch.object(UserGroupModel, 'delete', raise_exception):
 
                response = api_call(self, params)
 
                expected = 'failed to delete user group ID:%s %s' % (gr_id, gr_name)
 
                self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_user_group(gr_name)
 

	
 
    @base.parametrize('name,perm', [
 
        ('none', 'repository.none'),
 
        ('read', 'repository.read'),
 
        ('write', 'repository.write'),
 
        ('admin', 'repository.admin'),
 
    ])
 
    def test_api_grant_user_permission(self, name, perm):
 
        id_, params = _build_data(self.apikey,
 
                                  'grant_user_permission',
 
                                  repoid=self.REPO,
 
                                  userid=base.TEST_USER_ADMIN_LOGIN,
 
                                  perm=perm)
 
        response = api_call(self, params)
 

	
 
        ret = {
 
            'msg': 'Granted perm: `%s` for user: `%s` in repo: `%s`' % (
 
                perm, base.TEST_USER_ADMIN_LOGIN, self.REPO
 
            ),
 
            'success': True
 
        }
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    def test_api_grant_user_permission_wrong_permission(self):
 
        perm = 'haha.no.permission'
 
        id_, params = _build_data(self.apikey,
 
                                  'grant_user_permission',
 
                                  repoid=self.REPO,
 
                                  userid=base.TEST_USER_ADMIN_LOGIN,
 
                                  perm=perm)
 
        response = api_call(self, params)
 

	
 
        expected = 'permission `%s` does not exist' % perm
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(RepoModel, 'grant_user_permission', raise_exception)
 
    def test_api_grant_user_permission_exception_when_adding(self):
 
        perm = 'repository.read'
 
        id_, params = _build_data(self.apikey,
 
                                  'grant_user_permission',
 
                                  repoid=self.REPO,
 
                                  userid=base.TEST_USER_ADMIN_LOGIN,
 
                                  perm=perm)
 
        response = api_call(self, params)
 

	
 
        expected = 'failed to edit permission for user: `%s` in repo: `%s`' % (
 
            base.TEST_USER_ADMIN_LOGIN, self.REPO
 
        )
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_revoke_user_permission(self):
 
        id_, params = _build_data(self.apikey,
 
                                  'revoke_user_permission',
 
                                  repoid=self.REPO,
 
                                  userid=base.TEST_USER_ADMIN_LOGIN, )
 
        response = api_call(self, params)
 

	
 
        expected = {
 
            'msg': 'Revoked perm for user: `%s` in repo: `%s`' % (
 
                base.TEST_USER_ADMIN_LOGIN, self.REPO
 
            ),
 
            'success': True
 
        }
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(RepoModel, 'revoke_user_permission', raise_exception)
 
    def test_api_revoke_user_permission_exception_when_adding(self):
 
        id_, params = _build_data(self.apikey,
 
                                  'revoke_user_permission',
 
                                  repoid=self.REPO,
 
                                  userid=base.TEST_USER_ADMIN_LOGIN, )
 
        response = api_call(self, params)
 

	
 
        expected = 'failed to edit permission for user: `%s` in repo: `%s`' % (
 
            base.TEST_USER_ADMIN_LOGIN, self.REPO
 
        )
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @base.parametrize('name,perm', [
 
        ('none', 'repository.none'),
 
        ('read', 'repository.read'),
 
        ('write', 'repository.write'),
 
        ('admin', 'repository.admin'),
 
    ])
 
    def test_api_grant_user_group_permission(self, name, perm):
 
        id_, params = _build_data(self.apikey,
 
                                  'grant_user_group_permission',
 
                                  repoid=self.REPO,
 
                                  usergroupid=TEST_USER_GROUP,
 
                                  perm=perm)
 
        response = api_call(self, params)
 

	
 
        ret = {
 
            'msg': 'Granted perm: `%s` for user group: `%s` in repo: `%s`' % (
 
                perm, TEST_USER_GROUP, self.REPO
 
            ),
 
            'success': True
 
        }
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    def test_api_grant_user_group_permission_wrong_permission(self):
 
        perm = 'haha.no.permission'
 
        id_, params = _build_data(self.apikey,
 
                                  'grant_user_group_permission',
 
                                  repoid=self.REPO,
 
                                  usergroupid=TEST_USER_GROUP,
 
                                  perm=perm)
 
        response = api_call(self, params)
 

	
 
        expected = 'permission `%s` does not exist' % perm
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(RepoModel, 'grant_user_group_permission', raise_exception)
 
    def test_api_grant_user_group_permission_exception_when_adding(self):
 
        perm = 'repository.read'
 
        id_, params = _build_data(self.apikey,
 
                                  'grant_user_group_permission',
 
                                  repoid=self.REPO,
 
                                  usergroupid=TEST_USER_GROUP,
 
                                  perm=perm)
 
        response = api_call(self, params)
 

	
 
        expected = 'failed to edit permission for user group: `%s` in repo: `%s`' % (
 
            TEST_USER_GROUP, self.REPO
 
        )
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_revoke_user_group_permission(self):
 
        RepoModel().grant_user_group_permission(repo=self.REPO,
 
                                                group_name=TEST_USER_GROUP,
 
                                                perm='repository.read')
 
        meta.Session().commit()
 
        id_, params = _build_data(self.apikey,
 
                                  'revoke_user_group_permission',
 
                                  repoid=self.REPO,
 
                                  usergroupid=TEST_USER_GROUP, )
 
        response = api_call(self, params)
 

	
 
        expected = {
 
            'msg': 'Revoked perm for user group: `%s` in repo: `%s`' % (
 
                TEST_USER_GROUP, self.REPO
 
            ),
 
            'success': True
 
        }
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(RepoModel, 'revoke_user_group_permission', raise_exception)
 
    def test_api_revoke_user_group_permission_exception_when_adding(self):
 
        id_, params = _build_data(self.apikey,
 
                                  'revoke_user_group_permission',
 
                                  repoid=self.REPO,
 
                                  usergroupid=TEST_USER_GROUP, )
 
        response = api_call(self, params)
 

	
 
        expected = 'failed to edit permission for user group: `%s` in repo: `%s`' % (
 
            TEST_USER_GROUP, self.REPO
 
        )
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @base.parametrize('name,perm,apply_to_children', [
 
        ('none', 'group.none', 'none'),
 
        ('read', 'group.read', 'none'),
 
        ('write', 'group.write', 'none'),
 
        ('admin', 'group.admin', 'none'),
 

	
 
        ('none', 'group.none', 'all'),
 
        ('read', 'group.read', 'all'),
 
        ('write', 'group.write', 'all'),
 
        ('admin', 'group.admin', 'all'),
 

	
 
        ('none', 'group.none', 'repos'),
 
        ('read', 'group.read', 'repos'),
 
        ('write', 'group.write', 'repos'),
 
        ('admin', 'group.admin', 'repos'),
 

	
 
        ('none', 'group.none', 'groups'),
 
        ('read', 'group.read', 'groups'),
 
        ('write', 'group.write', 'groups'),
 
        ('admin', 'group.admin', 'groups'),
 
    ])
 
    def test_api_grant_user_permission_to_repo_group(self, name, perm, apply_to_children):
 
        id_, params = _build_data(self.apikey,
 
                                  'grant_user_permission_to_repo_group',
 
                                  repogroupid=TEST_REPO_GROUP,
 
                                  userid=base.TEST_USER_ADMIN_LOGIN,
 
                                  perm=perm, apply_to_children=apply_to_children)
 
        response = api_call(self, params)
 

	
 
        ret = {
 
            'msg': 'Granted perm: `%s` (recursive:%s) for user: `%s` in repo group: `%s`' % (
 
                perm, apply_to_children, base.TEST_USER_ADMIN_LOGIN, TEST_REPO_GROUP
 
            ),
 
            'success': True
 
        }
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    @base.parametrize('name,perm,apply_to_children,grant_admin,access_ok', [
 
        ('none_fails', 'group.none', 'none', False, False),
 
        ('read_fails', 'group.read', 'none', False, False),
 
        ('write_fails', 'group.write', 'none', False, False),
 
        ('admin_fails', 'group.admin', 'none', False, False),
 

	
 
        # with granted perms
 
        ('none_ok', 'group.none', 'none', True, True),
 
        ('read_ok', 'group.read', 'none', True, True),
 
        ('write_ok', 'group.write', 'none', True, True),
 
        ('admin_ok', 'group.admin', 'none', True, True),
 
    ])
 
    def test_api_grant_user_permission_to_repo_group_by_regular_user(
 
            self, name, perm, apply_to_children, grant_admin, access_ok):
 
        if grant_admin:
 
            RepoGroupModel().grant_user_permission(TEST_REPO_GROUP,
 
                                                   self.TEST_USER_LOGIN,
 
                                                   'group.admin')
 
            meta.Session().commit()
 

	
 
        id_, params = _build_data(self.apikey_regular,
 
                                  'grant_user_permission_to_repo_group',
 
                                  repogroupid=TEST_REPO_GROUP,
 
                                  userid=base.TEST_USER_ADMIN_LOGIN,
 
                                  perm=perm, apply_to_children=apply_to_children)
 
        response = api_call(self, params)
 
        if access_ok:
 
            ret = {
 
                'msg': 'Granted perm: `%s` (recursive:%s) for user: `%s` in repo group: `%s`' % (
 
                    perm, apply_to_children, base.TEST_USER_ADMIN_LOGIN, TEST_REPO_GROUP
 
                ),
 
                'success': True
 
            }
 
            expected = ret
 
            self._compare_ok(id_, expected, given=response.body)
 
        else:
 
            expected = 'repository group `%s` does not exist' % TEST_REPO_GROUP
 
            self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_grant_user_permission_to_repo_group_wrong_permission(self):
 
        perm = 'haha.no.permission'
 
        id_, params = _build_data(self.apikey,
 
                                  'grant_user_permission_to_repo_group',
 
                                  repogroupid=TEST_REPO_GROUP,
 
                                  userid=base.TEST_USER_ADMIN_LOGIN,
 
                                  perm=perm)
 
        response = api_call(self, params)
 

	
 
        expected = 'permission `%s` does not exist' % perm
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(RepoGroupModel, 'grant_user_permission', raise_exception)
 
    def test_api_grant_user_permission_to_repo_group_exception_when_adding(self):
 
        perm = 'group.read'
 
        id_, params = _build_data(self.apikey,
 
                                  'grant_user_permission_to_repo_group',
 
                                  repogroupid=TEST_REPO_GROUP,
 
                                  userid=base.TEST_USER_ADMIN_LOGIN,
 
                                  perm=perm)
 
        response = api_call(self, params)
 

	
 
        expected = 'failed to edit permission for user: `%s` in repo group: `%s`' % (
 
            base.TEST_USER_ADMIN_LOGIN, TEST_REPO_GROUP
 
        )
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @base.parametrize('name,apply_to_children', [
 
        ('none', 'none'),
 
        ('all', 'all'),
 
        ('repos', 'repos'),
 
        ('groups', 'groups'),
 
    ])
 
    def test_api_revoke_user_permission_from_repo_group(self, name, apply_to_children):
 
        RepoGroupModel().grant_user_permission(repo_group=TEST_REPO_GROUP,
 
                                               user=base.TEST_USER_ADMIN_LOGIN,
 
                                               perm='group.read',)
 
        meta.Session().commit()
 

	
 
        id_, params = _build_data(self.apikey,
 
                                  'revoke_user_permission_from_repo_group',
 
                                  repogroupid=TEST_REPO_GROUP,
 
                                  userid=base.TEST_USER_ADMIN_LOGIN,
 
                                  apply_to_children=apply_to_children,)
 
        response = api_call(self, params)
 

	
 
        expected = {
 
            'msg': 'Revoked perm (recursive:%s) for user: `%s` in repo group: `%s`' % (
 
                apply_to_children, base.TEST_USER_ADMIN_LOGIN, TEST_REPO_GROUP
 
            ),
 
            'success': True
 
        }
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    @base.parametrize('name,apply_to_children,grant_admin,access_ok', [
 
        ('none', 'none', False, False),
 
        ('all', 'all', False, False),
 
        ('repos', 'repos', False, False),
 
        ('groups', 'groups', False, False),
 

	
 
        # after granting admin rights
 
        ('none', 'none', False, False),
 
        ('all', 'all', False, False),
 
        ('repos', 'repos', False, False),
 
        ('groups', 'groups', False, False),
 
    ])
 
    def test_api_revoke_user_permission_from_repo_group_by_regular_user(
 
            self, name, apply_to_children, grant_admin, access_ok):
 
        RepoGroupModel().grant_user_permission(repo_group=TEST_REPO_GROUP,
 
                                               user=base.TEST_USER_ADMIN_LOGIN,
 
                                               perm='group.read',)
 
        meta.Session().commit()
 

	
 
        if grant_admin:
 
            RepoGroupModel().grant_user_permission(TEST_REPO_GROUP,
 
                                                   self.TEST_USER_LOGIN,
 
                                                   'group.admin')
 
            meta.Session().commit()
 

	
 
        id_, params = _build_data(self.apikey_regular,
 
                                  'revoke_user_permission_from_repo_group',
 
                                  repogroupid=TEST_REPO_GROUP,
 
                                  userid=base.TEST_USER_ADMIN_LOGIN,
 
                                  apply_to_children=apply_to_children,)
 
        response = api_call(self, params)
 
        if access_ok:
 
            expected = {
 
                'msg': 'Revoked perm (recursive:%s) for user: `%s` in repo group: `%s`' % (
 
                    apply_to_children, base.TEST_USER_ADMIN_LOGIN, TEST_REPO_GROUP
 
                ),
 
                'success': True
 
            }
 
            self._compare_ok(id_, expected, given=response.body)
 
        else:
 
            expected = 'repository group `%s` does not exist' % TEST_REPO_GROUP
 
            self._compare_error(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(RepoGroupModel, 'revoke_user_permission', raise_exception)
 
    def test_api_revoke_user_permission_from_repo_group_exception_when_adding(self):
 
        id_, params = _build_data(self.apikey,
 
                                  'revoke_user_permission_from_repo_group',
 
                                  repogroupid=TEST_REPO_GROUP,
 
                                  userid=base.TEST_USER_ADMIN_LOGIN, )
 
        response = api_call(self, params)
 

	
 
        expected = 'failed to edit permission for user: `%s` in repo group: `%s`' % (
 
            base.TEST_USER_ADMIN_LOGIN, TEST_REPO_GROUP
 
        )
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @base.parametrize('name,perm,apply_to_children', [
 
        ('none', 'group.none', 'none'),
 
        ('read', 'group.read', 'none'),
 
        ('write', 'group.write', 'none'),
 
        ('admin', 'group.admin', 'none'),
 

	
 
        ('none', 'group.none', 'all'),
 
        ('read', 'group.read', 'all'),
 
        ('write', 'group.write', 'all'),
 
        ('admin', 'group.admin', 'all'),
 

	
 
        ('none', 'group.none', 'repos'),
 
        ('read', 'group.read', 'repos'),
 
        ('write', 'group.write', 'repos'),
 
        ('admin', 'group.admin', 'repos'),
 

	
 
        ('none', 'group.none', 'groups'),
 
        ('read', 'group.read', 'groups'),
 
        ('write', 'group.write', 'groups'),
 
        ('admin', 'group.admin', 'groups'),
 
    ])
 
    def test_api_grant_user_group_permission_to_repo_group(self, name, perm, apply_to_children):
 
        id_, params = _build_data(self.apikey,
 
                                  'grant_user_group_permission_to_repo_group',
 
                                  repogroupid=TEST_REPO_GROUP,
 
                                  usergroupid=TEST_USER_GROUP,
 
                                  perm=perm,
 
                                  apply_to_children=apply_to_children,)
 
        response = api_call(self, params)
 

	
 
        ret = {
 
            'msg': 'Granted perm: `%s` (recursive:%s) for user group: `%s` in repo group: `%s`' % (
 
                perm, apply_to_children, TEST_USER_GROUP, TEST_REPO_GROUP
 
            ),
 
            'success': True
 
        }
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    @base.parametrize('name,perm,apply_to_children,grant_admin,access_ok', [
 
        ('none_fails', 'group.none', 'none', False, False),
 
        ('read_fails', 'group.read', 'none', False, False),
 
        ('write_fails', 'group.write', 'none', False, False),
 
        ('admin_fails', 'group.admin', 'none', False, False),
 

	
 
        # with granted perms
 
        ('none_ok', 'group.none', 'none', True, True),
 
        ('read_ok', 'group.read', 'none', True, True),
 
        ('write_ok', 'group.write', 'none', True, True),
 
        ('admin_ok', 'group.admin', 'none', True, True),
 
    ])
 
    def test_api_grant_user_group_permission_to_repo_group_by_regular_user(
 
            self, name, perm, apply_to_children, grant_admin, access_ok):
 
        if grant_admin:
 
            RepoGroupModel().grant_user_permission(TEST_REPO_GROUP,
 
                                                   self.TEST_USER_LOGIN,
 
                                                   'group.admin')
 
            meta.Session().commit()
 

	
 
        id_, params = _build_data(self.apikey_regular,
 
                                  'grant_user_group_permission_to_repo_group',
 
                                  repogroupid=TEST_REPO_GROUP,
 
                                  usergroupid=TEST_USER_GROUP,
 
                                  perm=perm,
 
                                  apply_to_children=apply_to_children,)
 
        response = api_call(self, params)
 
        if access_ok:
 
            ret = {
 
                'msg': 'Granted perm: `%s` (recursive:%s) for user group: `%s` in repo group: `%s`' % (
 
                    perm, apply_to_children, TEST_USER_GROUP, TEST_REPO_GROUP
 
                ),
 
                'success': True
0 comments (0 inline, 0 general)