@@ -34,2599 +34,2596 @@ from sqlalchemy import or_
from kallithea import EXTERN_TYPE_INTERNAL
from kallithea.controllers.api import JSONRPCController, JSONRPCError
from kallithea.lib.auth import (
PasswordGenerator, AuthUser, HasPermissionAllDecorator,
HasPermissionAnyDecorator, HasPermissionAnyApi, HasRepoPermissionAnyApi,
HasRepoGroupPermissionAnyApi, HasUserGroupPermissionAny)
from kallithea.lib.utils import map_groups, repo2db_mapper
from kallithea.lib.utils2 import (
str2bool, time_to_datetime, safe_int, Optional, OAttr)
from kallithea.model.meta import Session
from kallithea.model.repo_group import RepoGroupModel
from kallithea.model.scm import ScmModel, UserGroupList
from kallithea.model.repo import RepoModel
from kallithea.model.user import UserModel
from kallithea.model.user_group import UserGroupModel
from kallithea.model.gist import GistModel
from kallithea.model.db import (
Repository, Setting, UserIpMap, Permission, User, Gist,
RepoGroup)
from kallithea.lib.compat import json
from kallithea.lib.exceptions import (
DefaultUserException, UserGroupsAssignedException)
log = logging.getLogger(__name__)
def store_update(updates, attr, name):
"""
Stores param in updates dict if it's not instance of Optional
allows easy updates of passed in params
if not isinstance(attr, Optional):
updates[name] = attr
def get_user_or_error(userid):
Get user by id or name or return JsonRPCError if not found
:param userid:
user = UserModel().get_user(userid)
if user is None:
raise JSONRPCError("user `%s` does not exist" % (userid,))
return user
def get_repo_or_error(repoid):
Get repo by id or name or return JsonRPCError if not found
:param repoid:
repo = RepoModel().get_repo(repoid)
if repo is None:
raise JSONRPCError('repository `%s` does not exist' % (repoid,))
return repo
def get_repo_group_or_error(repogroupid):
Get repo group by id or name or return JsonRPCError if not found
:param repogroupid:
repo_group = RepoGroupModel()._get_repo_group(repogroupid)
if repo_group is None:
raise JSONRPCError(
'repository group `%s` does not exist' % (repogroupid,))
return repo_group
def get_user_group_or_error(usergroupid):
Get user group by id or name or return JsonRPCError if not found
:param usergroupid:
user_group = UserGroupModel().get_group(usergroupid)
if user_group is None:
raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
return user_group
def get_perm_or_error(permid, prefix=None):
Get permission by id or name or return JsonRPCError if not found
:param permid:
perm = Permission.get_by_key(permid)
if perm is None:
raise JSONRPCError('permission `%s` does not exist' % (permid,))
if prefix:
if not perm.permission_name.startswith(prefix):
raise JSONRPCError('permission `%s` is invalid, '
'should start with %s' % (permid, prefix))
return perm
def get_gist_or_error(gistid):
Get gist by id or gist_access_id or return JsonRPCError if not found
:param gistid:
gist = GistModel().get_gist(gistid)
if gist is None:
raise JSONRPCError('gist `%s` does not exist' % (gistid,))
return gist
class ApiController(JSONRPCController):
API Controller
Each method takes USER as first argument. This is then, based on given
API_KEY propagated as instance of user object who's making the call.
example function::
def func(apiuser,arg1, arg2,...):
pass
Each function should also **raise** JSONRPCError for any
errors that happens.
@HasPermissionAllDecorator('hg.admin')
def test(self, apiuser, args):
return args
def pull(self, apiuser, repoid):
Triggers a pull from remote location on given repo. Can be used to
automatically keep remote repos up to date. This command can be executed
only using api_key belonging to user with admin rights
:param apiuser: filled automatically from apikey
:type apiuser: AuthUser
:param repoid: repository name or repository id
:type repoid: str or int
OUTPUT::
id : <id_given_in_input>
result : {
"msg": "Pulled from `<repository name>`"
"repository": "<repository name>"
}
error : null
ERROR OUTPUT::
result : null
error : {
"Unable to pull changes from `<reponame>`"
repo = get_repo_or_error(repoid)
try:
ScmModel().pull_changes(repo.repo_name,
self.authuser.username)
return dict(
msg='Pulled from `%s`' % repo.repo_name,
repository=repo.repo_name
)
except Exception:
log.error(traceback.format_exc())
'Unable to pull changes from `%s`' % repo.repo_name
def rescan_repos(self, apiuser, remove_obsolete=Optional(False)):
Triggers rescan repositories action. If remove_obsolete is set
than also delete repos that are in database but not in the filesystem.
aka "clean zombies". This command can be executed only using api_key
belonging to user with admin rights.
:param remove_obsolete: deletes repositories from
database that are not found on the filesystem
:type remove_obsolete: Optional(bool)
'added': [<added repository name>,...]
'removed': [<removed repository name>,...]
'Error occurred during rescan repositories action'
rm_obsolete = Optional.extract(remove_obsolete)
added, removed = repo2db_mapper(ScmModel().repo_scan(),
remove_obsolete=rm_obsolete)
return {'added': added, 'removed': removed}
def invalidate_cache(self, apiuser, repoid):
Invalidate cache for repository.
This command can be executed only using api_key belonging to user with admin
rights or regular user that have write or admin or write access to repository.
'msg': Cache for repository `<repository name>` was invalidated,
'repository': <repository name>
'Error occurred during cache invalidation action'
if not HasPermissionAnyApi('hg.admin')(user=apiuser):
# check if we have admin permission for this repo !
if not HasRepoPermissionAnyApi('repository.admin',
'repository.write')(
user=apiuser, repo_name=repo.repo_name):
ScmModel().mark_for_invalidation(repo.repo_name)
msg='Cache for repository `%s` was invalidated' % (repoid,),
# permission check inside
def lock(self, apiuser, repoid, locked=Optional(None),
userid=Optional(OAttr('apiuser'))):
Set locking state on given repository by given user. If userid param
is skipped, then it is set to id of user who is calling this method.
If locked param is skipped then function shows current lock state of
given repo. This command can be executed only using api_key belonging
to user with admin rights or regular user that have admin or write
access to repository.
:param locked: lock state to be set
:type locked: Optional(bool)
:param userid: set lock as user
:type userid: Optional(str or int)
'repo': '<reponame>',
'locked': <bool: lock state>,
'locked_since': <int: lock timestamp>,
'locked_by': <username of person who made the lock>,
'lock_state_changed': <bool: True if lock state has been changed in this request>,
'msg': 'Repo `<reponame>` locked by `<username>` on <timestamp>.'
or
'msg': 'Repo `<repository name>` not locked.'
'msg': 'User `<user name>` set lock state for repo `<repository name>` to `<new lock state>`'
'Error occurred locking repository `<reponame>`
if HasPermissionAnyApi('hg.admin')(user=apiuser):
elif HasRepoPermissionAnyApi('repository.admin',
'repository.write')(user=apiuser,
repo_name=repo.repo_name):
# make sure normal user does not pass someone else userid,
# he is not allowed to do that
if not isinstance(userid, Optional) and userid != apiuser.user_id:
'userid is not the same as your user'
else:
if isinstance(userid, Optional):
userid = apiuser.user_id
user = get_user_or_error(userid)
if isinstance(locked, Optional):
lockobj = Repository.getlock(repo)
if lockobj[0] is None:
_d = {
'repo': repo.repo_name,
'locked': False,
'locked_since': None,
'locked_by': None,
'lock_state_changed': False,
'msg': 'Repo `%s` not locked.' % repo.repo_name
return _d
userid, time_ = lockobj
lock_user = get_user_or_error(userid)
'locked': True,
'locked_since': time_,
'locked_by': lock_user.username,
'msg': ('Repo `%s` locked by `%s` on `%s`.'
% (repo.repo_name, lock_user.username,
json.dumps(time_to_datetime(time_))))
# force locked state through a flag
locked = str2bool(locked)
if locked:
lock_time = time.time()
Repository.lock(repo, user.user_id, lock_time)
lock_time = None
Repository.unlock(repo)
'locked': locked,
'locked_since': lock_time,
'locked_by': user.username,
'lock_state_changed': True,
'msg': ('User `%s` set lock state for repo `%s` to `%s`'
% (user.username, repo.repo_name, locked))
'Error occurred locking repository `%s`' % repo.repo_name
def get_locks(self, apiuser, userid=Optional(OAttr('apiuser'))):
Get all repositories with locks for given userid, if
this command is run by non-admin account userid is set to user
who is calling this method, thus returning locks for himself.
:param userid: User to get locks for
[repo_object, repo_object,...]
ret = []
user = None
# show all locks
for r in Repository.getAll():
userid, time_ = r.locked
if time_:
_api_data = r.get_api_data()
# if we use userfilter just show the locks for this user
if user is not None:
if safe_int(userid) == user.user_id:
ret.append(_api_data)
return ret
def get_ip(self, apiuser, userid=Optional(OAttr('apiuser'))):
Shows IP address as seen from Kallithea server, together with all
defined IP addresses for given user. If userid is not passed data is
returned for user who's calling this function.
This command can be executed only using api_key belonging to user with
admin rights.
:param userid: username to show ips for
"server_ip_addr": "<ip_from_clien>",
"user_ips": [
{
"ip_addr": "<ip_with_mask>",
"ip_range": ["<start_ip>", "<end_ip>"],
},
...
]
ips = UserIpMap.query().filter(UserIpMap.user == user).all()
server_ip_addr=self.ip_addr,
user_ips=ips
# alias for old
show_ip = get_ip
def get_server_info(self, apiuser):
return server info, including Kallithea version and installed packages
'modules': [<module name>,...]
'py_version': <python version>,
'platform': <platform type>,
'kallithea_version': <kallithea version>
return Setting.get_server_info()
def get_user(self, apiuser, userid=Optional(OAttr('apiuser'))):
Gets a user by username or user_id, Returns empty result if user is
not found. If userid param is skipped it is set to id of user who is
calling this method. This command can be executed only using api_key
belonging to user with admin rights, or regular users that cannot
specify different userid than theirs
:param userid: user to get data for
result: None if user does not exist or
"user_id" : "<user_id>",
"api_key" : "<api_key>",
"api_keys": "[<list of all API keys including additional ones>]"
"username" : "<username>",
"firstname": "<firstname>",
"lastname" : "<lastname>",
"email" : "<email>",
"emails": "[<list of all emails including additional ones>]",
"ip_addresses": "[<ip_address_for_user>,...]",
"active" : "<bool: user active>",
"admin" :Â "<bool: user is admin>",
"extern_name" : "<extern_name>",
"extern_type" : "<extern type>
"last_login": "<last_login>",
"permissions": {
"global": ["hg.create.repository",
"repository.read",
"hg.register.manual_activate"],
"repositories": {"repo1": "repository.none"},
"repositories_groups": {"Group1": "group.read"}
error: null
data = user.get_api_data()
data['permissions'] = AuthUser(user_id=user.user_id).permissions
return data
def get_users(self, apiuser):
Lists all existing users. This command can be executed only using api_key
result: [<user_object>, ...]
result = []
users_list = User.query().order_by(User.username) \
.filter(User.username != User.DEFAULT_USER) \
.all()
for user in users_list:
result.append(user.get_api_data())
return result
def create_user(self, apiuser, username, email, password=Optional(''),
firstname=Optional(''), lastname=Optional(''),
active=Optional(True), admin=Optional(False),
extern_name=Optional(EXTERN_TYPE_INTERNAL),
extern_type=Optional(EXTERN_TYPE_INTERNAL)):
Creates new user. Returns new user object. This command can
be executed only using api_key belonging to user with admin rights.
:param username: new username
:type username: str or int
:param email: email
:type email: str
:param password: password
:type password: Optional(str)
:param firstname: firstname
:type firstname: Optional(str)
:param lastname: lastname
:type lastname: Optional(str)
:param active: active
:type active: Optional(bool)
:param admin: admin
:type admin: Optional(bool)
:param extern_name: name of extern
:type extern_name: Optional(str)
:param extern_type: extern_type
:type extern_type: Optional(str)
result: {
"msg" : "created new user `<username>`",
"user": <user_obj>
"user `<username>` already exist"
"email `<email>` already exist"
"failed to create user `<username>`"
if User.get_by_username(username):
raise JSONRPCError("user `%s` already exist" % (username,))
if User.get_by_email(email, case_insensitive=True):
raise JSONRPCError("email `%s` already exist" % (email,))
if Optional.extract(extern_name):
# generate temporary password if user is external
password = PasswordGenerator().gen_password(length=8)
user = UserModel().create_or_update(
username=Optional.extract(username),
password=Optional.extract(password),
email=Optional.extract(email),
firstname=Optional.extract(firstname),
lastname=Optional.extract(lastname),
active=Optional.extract(active),
admin=Optional.extract(admin),
extern_type=Optional.extract(extern_type),
extern_name=Optional.extract(extern_name)
Session().commit()
msg='created new user `%s`' % username,
user=user.get_api_data()
raise JSONRPCError('failed to create user `%s`' % (username,))
def update_user(self, apiuser, userid, username=Optional(None),
email=Optional(None),password=Optional(None),
firstname=Optional(None), lastname=Optional(None),
active=Optional(None), admin=Optional(None),
extern_type=Optional(None), extern_name=Optional(None),):
updates given user if such user exists. This command can
:param userid: userid to update
:type userid: str or int
:param extern_name:
:param extern_type:
"msg" : "updated user ID:<userid> <username>",
"user": <user_object>,
"failed to update user `<username>`"
# only non optional arguments will be stored in updates
updates = {}
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)
msg='updated user ID:%s %s' % (user.user_id, user.username),
except DefaultUserException:
raise JSONRPCError('editing default user is forbidden')
raise JSONRPCError('failed to update user `%s`' % (userid,))
def delete_user(self, apiuser, userid):
deletes given user if such user exists. This command can
:param userid: user to delete
"msg" : "deleted user ID:<userid> <username>",
"user": null
"failed to delete user ID:<userid> <username>"
UserModel().delete(userid)
msg='deleted user ID:%s %s' % (user.user_id, user.username),
user=None
raise JSONRPCError('failed to delete user ID:%s %s'
% (user.user_id, user.username))
def get_user_group(self, apiuser, 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
result : None if group not exist
"users_group_id" : "<id>",
"group_name" : "<groupname>",
"active": "<bool>",
"members" : [<user_obj>,...]
user_group = get_user_group_or_error(usergroupid)
# check if we have at least read permission for this user group !
_perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)
if not HasUserGroupPermissionAny(*_perms)(
user=apiuser, user_group_name=user_group.users_group_name):
data = user_group.get_api_data()
def get_user_groups(self, apiuser):
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
result : [<user_group_obj>,...]
extras = {'user': apiuser}
for user_group in UserGroupList(UserGroupModel().get_all(),
perm_set=_perms, extra_kwargs=extras):
result.append(user_group.get_api_data())
@HasPermissionAnyDecorator('hg.admin', 'hg.usergroup.create.true')
def create_user_group(self, apiuser, group_name, description=Optional(''),
owner=Optional(OAttr('apiuser')), active=Optional(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
"msg": "created new user group `<groupname>`",
"user_group": <user_group_object>
"user group `<group name>` already exist"
"failed to create group `<group name>`"
if UserGroupModel().get_by_name(group_name):
raise JSONRPCError("user group `%s` already exist" % (group_name,))
if isinstance(owner, Optional):
owner = apiuser.user_id
owner = get_user_or_error(owner)
active = Optional.extract(active)
description = Optional.extract(description)
ug = UserGroupModel().create(name=group_name, description=description,
owner=owner, active=active)
msg='created new user group `%s`' % group_name,
user_group=ug.get_api_data()
raise JSONRPCError('failed to create group `%s`' % (group_name,))
def update_user_group(self, apiuser, usergroupid, group_name=Optional(''),
description=Optional(''), owner=Optional(None),
active=Optional(True)):
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
:param owner: owner of group.
"msg": 'updated user group ID:<user group id> <user group name>',
"failed to update user group `<user group name>`"
# check if we have admin permission for this user group !
_perms = ('usergroup.admin',)
if not isinstance(owner, Optional):
store_update(updates, group_name, 'users_group_name')
store_update(updates, description, 'user_group_description')
store_update(updates, owner, 'user')
store_update(updates, active, 'users_group_active')
UserGroupModel().update(user_group, updates)
msg='updated user group ID:%s %s' % (user_group.users_group_id,
user_group.users_group_name),
user_group=user_group.get_api_data()
raise JSONRPCError('failed to update user group `%s`' % (usergroupid,))
def delete_user_group(self, apiuser, usergroupid):
Delete given user group by user group id or name.
This command can be executed only using api_key
:type usergroupid: int
"msg": "deleted user group ID:<user_group_id> <user_group_name>"
"failed to delete user group ID:<user_group_id> <user_group_name>"
"RepoGroup assigned to <repo_groups_list>"
UserGroupModel().delete(user_group)
msg='deleted user group ID:%s %s' %
(user_group.users_group_id, user_group.users_group_name),
user_group=None
except UserGroupsAssignedException as e:
raise JSONRPCError(str(e))
raise JSONRPCError('failed to delete user group ID:%s %s' %
(user_group.users_group_id,
user_group.users_group_name)
def add_user_to_user_group(self, apiuser, 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
:type userid: int
"success": True|False # depends on if member is in group
"msg": "added member `<username>` to user group `<groupname>` |
User is already in that group"
"failed to add member to user group `<user_group_name>`"
ugm = UserGroupModel().add_user_to_group(user_group, user)
success = True if ugm != 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'
success=success,
msg=msg
'failed to add member to user group `%s`' % (
user_group.users_group_name,
def remove_user_from_user_group(self, apiuser, 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
"success": True|False, # depends on if member is in group
"msg": "removed member <username> from user group <groupname> |
User wasn't in group"
success = UserGroupModel().remove_user_from_group(user_group, user)
msg = 'removed member `%s` from user group `%s`' % (
msg = msg if success else "User wasn't in group"
return dict(success=success, msg=msg)
'failed to remove member from user group `%s`' % (
def get_repo(self, apiuser, repoid):
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.
"repo_id" : "<repo_id>",
"repo_name" : "<reponame>"
"repo_type" : "<repo_type>",
"clone_uri" : "<clone_uri>",
"enable_downloads": "<bool>",
"enable_locking": "<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>, ...]
perms = ('repository.admin', 'repository.write', 'repository.read')
if not HasRepoPermissionAnyApi(*perms)(user=apiuser, repo_name=repo.repo_name):
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",
members.append(user_group_data)
for user in repo.followers:
followers.append(user.user.get_api_data())
data = repo.get_api_data()
data['members'] = members
data['followers'] = followers
def get_repos(self, apiuser):
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.
result: [
"private": : "<bool>",
"created_on" : "<datetimecreated>",
repos = RepoModel().get_all_user_repos(user=apiuser)
repos = RepoModel().get_all()
for repo in repos:
result.append(repo.get_api_data())
def get_repo_nodes(self, apiuser, repoid, revision, root_path,
ret_type=Optional('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 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)
"name" : "<name>"
"type" : "<type>",
ret_type = Optional.extract(ret_type)
_map = {}
_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(_map.keys())))
'failed to get repo: `%s` nodes' % repo.repo_name
@HasPermissionAnyDecorator('hg.admin', 'hg.create.repository')
def create_repo(self, apiuser, repo_name, owner=Optional(OAttr('apiuser')),
repo_type=Optional('hg'), description=Optional(''),
private=Optional(False), clone_uri=Optional(None),
landing_rev=Optional('rev:tip'),
enable_statistics=Optional(False),
enable_locking=Optional(False),
enable_downloads=Optional(False),
copy_permissions=Optional(False)):
Creates a repository. If repository name contains "/", all needed repository
groups will be created. For example "foo/bar/baz" will create groups
"foo", "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_locking:
:type enable_locking: bool
: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
"msg": "Created new repository `<reponame>`",
"success": true,
"task": "<celery task id or None if done sync>"
'failed to create repository `<repo_name>`
#forbid setting owner for non-admins
'Only Kallithea admin can specify `owner` param'
if RepoModel().get_by_repo_name(repo_name):
raise JSONRPCError("repo `%s` already exist" % repo_name)
defs = Setting.get_default_repo_settings(strip_prefix=True)
if isinstance(private, Optional):
private = defs.get('repo_private') or Optional.extract(private)
if isinstance(repo_type, Optional):
repo_type = defs.get('repo_type')
if isinstance(enable_statistics, Optional):
enable_statistics = defs.get('repo_enable_statistics')
if isinstance(enable_locking, Optional):
enable_locking = defs.get('repo_enable_locking')
if isinstance(enable_downloads, Optional):
enable_downloads = defs.get('repo_enable_downloads')
clone_uri = Optional.extract(clone_uri)
landing_rev = Optional.extract(landing_rev)
copy_permissions = Optional.extract(copy_permissions)
repo_name_cleaned = repo_name.split('/')[-1]
# create structure of groups and return the last group
repo_group = map_groups(repo_name)
data = dict(
repo_name=repo_name_cleaned,
repo_name_full=repo_name,
repo_type=repo_type,
repo_description=description,
owner=owner,
repo_private=private,
clone_uri=clone_uri,
repo_group=repo_group,
repo_landing_rev=landing_rev,
enable_statistics=enable_statistics,
enable_locking=enable_locking,
enable_downloads=enable_downloads,
repo_copy_permissions=copy_permissions,
task = RepoModel().create(form_data=data, cur_user=owner)
from celery.result import BaseAsyncResult
task_id = None
if isinstance(task, BaseAsyncResult):
task_id = task.task_id
# no commit, it's done in RepoModel, or async via celery
msg="Created new repository `%s`" % (repo_name,),
success=True, # cannot return the repo data here since fork
# can be done async
task=task_id
'failed to create repository `%s`' % (repo_name,))
def update_repo(self, apiuser, repoid, name=Optional(None),
owner=Optional(OAttr('apiuser')),
group=Optional(None),
description=Optional(''), private=Optional(False),
clone_uri=Optional(None), landing_rev=Optional('rev:tip'),
enable_downloads=Optional(False)):
Updates repo
:param name:
:param owner:
:param group:
:param description:
:param landing_rev:
if not HasRepoPermissionAnyApi('repository.admin')(user=apiuser,
if (name != repo.repo_name and
not HasPermissionAnyApi('hg.create.repository')(user=apiuser)
):
raise JSONRPCError('no permission to create (or move) repositories')
updates = {
# update function requires this.
'repo_name': repo.repo_name
repo_group = group
if not isinstance(repo_group, Optional):
repo_group = get_repo_group_or_error(repo_group)
repo_group = repo_group.group_id
store_update(updates, name, 'repo_name')
store_update(updates, repo_group, 'repo_group')
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_locking, 'repo_enable_locking')
store_update(updates, enable_downloads, 'repo_enable_downloads')
RepoModel().update(repo, **updates)
msg='updated repo ID:%s %s' % (repo.repo_id, repo.repo_name),
repository=repo.get_api_data()
raise JSONRPCError('failed to update repo `%s`' % repoid)
@HasPermissionAnyDecorator('hg.admin', 'hg.fork.repository')
def fork_repo(self, apiuser, repoid, fork_name,
description=Optional(''), copy_permissions=Optional(False),
private=Optional(False), landing_rev=Optional('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 fork_name:
:param copy_permissions:
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>",
"landing_rev": "<landing_rev>"
"msg": "Created fork of `<reponame>` as `<forkname>`",
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))
'repository.write',
'repository.read')(user=apiuser,
if not HasPermissionAnyApi('hg.create.repository')(user=apiuser):
raise JSONRPCError('no permission to create repositories')
group = map_groups(fork_name)
form_data = dict(
repo_name=fork_name,
repo_name_full=fork_name,
repo_group=group,
repo_type=repo.repo_type,
description=Optional.extract(description),
private=Optional.extract(private),
copy_permissions=Optional.extract(copy_permissions),
landing_rev=Optional.extract(landing_rev),
update_after_clone=False,
fork_parent_id=repo.repo_id,
task = RepoModel().create_fork(form_data, cur_user=owner)
msg='Created fork of `%s` as `%s`' % (repo.repo_name,
fork_name),
'failed to fork repository `%s` as `%s`' % (repo_name,
fork_name)
def delete_repo(self, apiuser, repoid, forks=Optional('')):
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 forks: `detach` or `delete`, what do do with attached forks for repo
:type forks: Optional(str)
"msg": "Deleted repository `<reponame>`",
"success": true
handle_forks = Optional.extract(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:
'Cannot delete `%s` it still contains attached forks' %
(repo.repo_name,)
RepoModel().delete(repo, forks=forks)
msg='Deleted repository `%s`%s' % (repo.repo_name, _forks_msg),
success=True
'failed to delete repository `%s`' % (repo.repo_name,)
def grant_user_permission(self, apiuser, 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 perm: (repository.(none|read|write|admin))
:type perm: str
"msg" : "Granted perm: `<perm>` for user: `<username>` in repo: `<reponame>`",
perm = get_perm_or_error(perm)
RepoModel().grant_user_permission(repo=repo, user=user, perm=perm)
msg='Granted perm: `%s` for user: `%s` in repo: `%s`' % (
perm.permission_name, user.username, repo.repo_name
),
'failed to edit permission for user: `%s` in repo: `%s`' % (
userid, repoid
def revoke_user_permission(self, apiuser, repoid, userid):
Revoke permission for user on given repository. This command can be executed
only using api_key belonging to user with admin rights.
"msg" : "Revoked perm for user: `<username>` in repo: `<reponame>`",
RepoModel().revoke_user_permission(repo=repo, user=user)
msg='Revoked perm for user: `%s` in repo: `%s`' % (
user.username, repo.repo_name
def grant_user_group_permission(self, apiuser, 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 usergroupid: id of usergroup
"msg" : "Granted perm: `<perm>` for group: `<usersgroupname>` in repo: `<reponame>`",
"failed to edit permission for user group: `<usergroup>` in repo `<repo>`'
_perms = ('repository.admin',)
if not HasRepoPermissionAnyApi(*_perms)(
RepoModel().grant_user_group_permission(
repo=repo, group_name=user_group, perm=perm)
msg='Granted perm: `%s` for user group: `%s` in '
'repo: `%s`' % (
perm.permission_name, user_group.users_group_name,
repo.repo_name
'failed to edit permission for user group: `%s` in '
usergroupid, repo.repo_name
def revoke_user_group_permission(self, apiuser, 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.
"msg" : "Revoked perm for group: `<usersgroupname>` in repo: `<reponame>`",
RepoModel().revoke_user_group_permission(
repo=repo, group_name=user_group)
msg='Revoked perm for user group: `%s` in repo: `%s`' % (
user_group.users_group_name, repo.repo_name
def get_repo_group(self, apiuser, 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)
for user in repo_group.repo_group_to_perm:
for user_group in repo_group.users_group_to_perm:
data = repo_group.get_api_data()
data["members"] = members
def get_repo_groups(self, apiuser):
Returns all repository groups
for repo_group in RepoGroupModel().get_all():
result.append(repo_group.get_api_data())
def create_repo_group(self, apiuser, group_name, description=Optional(''),
parent=Optional(None),
Creates a repository group. This command can be executed only using
:param group_name:
:type group_name:
:type description:
:type owner:
:param parent:
:type parent:
:type copy_permissions:
"msg": "created new repo group `<repo_group_name>`"
"repo_group": <repogroup_object>
failed to create repo group `<repogroupid>`
if RepoGroup.get_by_group_name(group_name):
raise JSONRPCError("repo group `%s` already exist" % (group_name,))
group_description = Optional.extract(description)
parent_group = Optional.extract(parent)
if not isinstance(parent, Optional):
parent_group = get_repo_group_or_error(parent_group)
repo_group = RepoGroupModel().create(
group_name=group_name,
group_description=group_description,
parent=parent_group,
copy_permissions=copy_permissions
msg='created new repo group `%s`' % group_name,
repo_group=repo_group.get_api_data()
raise JSONRPCError('failed to create repo group `%s`' % (group_name,))
def update_repo_group(self, apiuser, repogroupid, group_name=Optional(''),
description=Optional(''),
parent=Optional(None), enable_locking=Optional(False)):
store_update(updates, group_name, 'group_name')
store_update(updates, description, 'group_description')
store_update(updates, owner, 'owner')
store_update(updates, parent, 'parent_group')
store_update(updates, enable_locking, 'enable_locking')
repo_group = RepoGroupModel().update(repo_group, updates)
msg='updated repository group ID:%s %s' % (repo_group.group_id,
repo_group.group_name),
raise JSONRPCError('failed to update repository group `%s`'
% (repogroupid,))
def delete_repo_group(self, apiuser, repogroupid):
:param repogroupid: name or id of repository group
'msg': 'deleted repo group ID:<repogroupid> <repogroupname>
'repo_group': null
"failed to delete repo group ID:<repogroupid> <repogroupname>"
RepoGroupModel().delete(repo_group)
msg='deleted repo group ID:%s %s' %
(repo_group.group_id, repo_group.group_name),
repo_group=None
raise JSONRPCError('failed to delete repo group ID:%s %s' %
(repo_group.group_id, repo_group.group_name)
def grant_user_permission_to_repo_group(self, apiuser, repogroupid, userid,
perm, apply_to_children=Optional('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 perm: (group.(none|read|write|admin))
:param apply_to_children: 'none', 'repos', 'groups', 'all'
:type apply_to_children: str
"msg" : "Granted perm: `<perm>` (recursive:<apply_to_children>) for user: `<username>` in repo group: `<repo_group_name>`",
"failed to edit permission for user: `<userid>` in repo group: `<repo_group_name>`"
# check if we have admin permission for this repo group !
if not HasRepoGroupPermissionAnyApi('group.admin')(user=apiuser,
group_name=repo_group.group_name):
raise JSONRPCError('repository group `%s` does not exist' % (repogroupid,))
perm = get_perm_or_error(perm, prefix='group.')
apply_to_children = Optional.extract(apply_to_children)
RepoGroupModel().add_permission(repo_group=repo_group,
obj=user,
obj_type="user",
perm=perm,
recursive=apply_to_children)
msg='Granted perm: `%s` (recursive:%s) for user: `%s` in repo group: `%s`' % (
perm.permission_name, apply_to_children, user.username, repo_group.name
'failed to edit permission for user: `%s` in repo group: `%s`' % (
userid, repo_group.name))
def revoke_user_permission_from_repo_group(self, apiuser, repogroupid, userid,
apply_to_children=Optional('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.
:type userid:
"msg" : "Revoked perm (recursive:<apply_to_children>) for user: `<username>` in repo group: `<repo_group_name>`",
RepoGroupModel().delete_permission(repo_group=repo_group,
msg='Revoked perm (recursive:%s) for user: `%s` in repo group: `%s`' % (
apply_to_children, user.username, repo_group.name
def grant_user_group_permission_to_repo_group(
self, apiuser, repogroupid, usergroupid, perm,
apply_to_children=Optional('none'),):
Grant permission for user group on given repository group, or update
api_key belonging to user with admin rights, or user who has admin
right to given repository group.
"msg" : "Granted perm: `<perm>` (recursive:<apply_to_children>) for user group: `<usersgroupname>` in repo group: `<repo_group_name>`",
"failed to edit permission for user group: `<usergroup>` in repo group: `<repo_group_name>`"
_perms = ('group.admin',)
if not HasRepoGroupPermissionAnyApi(*_perms)(
user=apiuser, group_name=repo_group.group_name):
'user group `%s` does not exist' % (usergroupid,))
obj=user_group,
obj_type="user_group",
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
'repo group: `%s`' % (
usergroupid, repo_group.name
def revoke_user_group_permission_from_repo_group(
self, apiuser, repogroupid, usergroupid,
executed only using api_key belonging to user with admin rights, or
"msg" : "Revoked perm (recursive:<apply_to_children>) for user group: `<usersgroupname>` in repo group: `<repo_group_name>`",
msg='Revoked perm (recursive:%s) for user group: `%s` in repo group: `%s`' % (
apply_to_children, user_group.users_group_name, repo_group.name
'failed to edit permission for user group: `%s` in repo group: `%s`' % (
def get_gist(self, apiuser, gistid):
Get given gist by id
:param gistid: id of private or public gist
:type gistid: str
gist = get_gist_or_error(gistid)
if gist.gist_owner != apiuser.user_id:
return gist.get_api_data()
def get_gists(self, apiuser, userid=Optional(OAttr('apiuser'))):
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
user_id = apiuser.user_id
user_id = get_user_or_error(userid).user_id
gists = []
_gists = Gist().query()\
.filter(or_(Gist.gist_expires == -1, Gist.gist_expires >= time.time()))\
.filter(Gist.gist_owner == user_id)\
.order_by(Gist.created_on.desc())
for gist in _gists:
gists.append(gist.get_api_data())
return gists
def create_gist(self, apiuser, files, owner=Optional(OAttr('apiuser')),
gist_type=Optional(Gist.GIST_PUBLIC), lifetime=Optional(-1),
description=Optional('')):
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
: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
"msg": "created new gist",
"gist": {}
"failed to create gist"
gist_type = Optional.extract(gist_type)
lifetime = Optional.extract(lifetime)
gist = GistModel().create(description=description,
gist_mapping=files,
gist_type=gist_type,
lifetime=lifetime)
msg='created new gist',
gist=gist.get_api_data()
raise JSONRPCError('failed to create gist')
# def update_gist(self, apiuser, gistid, files, owner=Optional(OAttr('apiuser')),
# gist_type=Optional(Gist.GIST_PUBLIC),
# gist_lifetime=Optional(-1), gist_description=Optional('')):
# gist = get_gist_or_error(gistid)
# updates = {}
def delete_gist(self, apiuser, gistid):
Deletes existing gist
:param gistid: id of gist to delete
"deleted gist ID: <gist_id>",
"gist": null
"failed to delete gist ID:<gist_id>"
GistModel().delete(gist)
msg='deleted gist ID:%s' % (gist.gist_access_id,),
gist=None
raise JSONRPCError('failed to delete gist ID:%s'
% (gist.gist_access_id,))
# -*- 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.model.repo
~~~~~~~~~~~~~~~~~~~~
Repository model 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: Jun 5, 2010
:author: marcink
:copyright: (c) 2013 RhodeCode GmbH, and others.
:license: GPLv3, see LICENSE.md for more details.
import os
import shutil
import logging
import traceback
from datetime import datetime
from sqlalchemy.orm import subqueryload
from kallithea.lib.utils import make_ui
from kallithea.lib.vcs.backends import get_backend
from kallithea.lib.utils2 import LazyProperty, safe_str, safe_unicode, \
remove_prefix, obfuscate_url_pw, get_current_authuser
from kallithea.lib.caching_query import FromCache
from kallithea.lib.hooks import log_delete_repository
from kallithea.model import BaseModel
from kallithea.model.db import Repository, UserRepoToPerm, UserGroupRepoToPerm, \
UserRepoGroupToPerm, UserGroupRepoGroupToPerm, User, Permission, \
Statistics, UserGroup, Ui, RepoGroup, RepositoryField
from kallithea.lib import helpers as h
from kallithea.lib.auth import HasRepoPermissionAny, HasUserGroupPermissionAny
from kallithea.lib.exceptions import AttachedForksError
from kallithea.model.scm import UserGroupList
class RepoModel(BaseModel):
cls = Repository
URL_SEPARATOR = Repository.url_sep()
def _get_user_group(self, users_group):
return self._get_instance(UserGroup, users_group,
callback=UserGroup.get_by_group_name)
def _get_repo_group(self, repo_group):
return self._get_instance(RepoGroup, repo_group,
callback=RepoGroup.get_by_group_name)
def _create_default_perms(self, repository, private):
# create default permission
default = 'repository.read'
def_user = User.get_default_user()
for p in def_user.user_perms:
if p.permission.permission_name.startswith('repository.'):
default = p.permission.permission_name
break
default_perm = 'repository.none' if private else default
repo_to_perm = UserRepoToPerm()
repo_to_perm.permission = Permission.get_by_key(default_perm)
repo_to_perm.repository = repository
repo_to_perm.user_id = def_user.user_id
return repo_to_perm
@LazyProperty
def repos_path(self):
Gets the repositories root path from database
q = self.sa.query(Ui).filter(Ui.ui_key == '/').one()
return q.ui_value
def get(self, repo_id, cache=False):
repo = self.sa.query(Repository) \
.filter(Repository.repo_id == repo_id)
if cache:
repo = repo.options(FromCache("sql_cache_short",
"get_repo_%s" % repo_id))
return repo.scalar()
def get_repo(self, repository):
return self._get_repo(repository)
def get_by_repo_name(self, repo_name, cache=False):
.filter(Repository.repo_name == repo_name)
"get_repo_%s" % repo_name))
def get_all_user_repos(self, user):
Gets all repositories that user have at least read access
:param user:
from kallithea.lib.auth import AuthUser
user = self._get_user(user)
repos = AuthUser(user_id=user.user_id).permissions['repositories']
access_check = lambda r: r[1] in ['repository.read',
'repository.admin']
repos = [x[0] for x in filter(access_check, repos.items())]
return Repository.query().filter(Repository.repo_name.in_(repos))
def get_users_js(self):
users = self.sa.query(User).filter(User.active == True).all()
return json.dumps([
'id': u.user_id,
'fname': h.escape(u.name),
'lname': h.escape(u.lastname),
'nname': u.username,
'gravatar_lnk': h.gravatar_url(u.email, size=28),
'gravatar_size': 14,
} for u in users]
def get_user_groups_js(self):
user_groups = self.sa.query(UserGroup) \
.filter(UserGroup.users_group_active == True) \
.options(subqueryload(UserGroup.members)) \
user_groups = UserGroupList(user_groups, perm_set=['usergroup.read',
'usergroup.write',
'usergroup.admin'])
'id': gr.users_group_id,
'grname': gr.users_group_name,
'grmembers': len(gr.members),
} for gr in user_groups]
@classmethod
def _render_datatable(cls, tmpl, *args, **kwargs):
import kallithea
from pylons import tmpl_context as c
from pylons.i18n.translation import _
_tmpl_lookup = kallithea.CONFIG['pylons.app_globals'].mako_lookup
template = _tmpl_lookup.get_template('data_table/_dt_elements.html')
tmpl = template.get_def(tmpl)
kwargs.update(dict(_=_, h=h, c=c))
return tmpl.render(*args, **kwargs)
def update_repoinfo(cls, repositories=None):
if repositories is None:
repositories = Repository.getAll()
for repo in repositories:
repo.update_changeset_cache()
def get_repos_as_dict(self, repos_list=None, admin=False, perm_check=True,
super_user_actions=False):
_render = self._render_datatable
def quick_menu(repo_name):
return _render('quick_menu', repo_name)
def repo_lnk(name, rtype, rstate, private, fork_of):
return _render('repo_name', name, rtype, rstate, private, fork_of,
short_name=not admin, admin=False)
def last_change(last_change):
return _render("last_change", last_change)
def rss_lnk(repo_name):
return _render("rss", repo_name)
def atom_lnk(repo_name):
return _render("atom", repo_name)
def last_rev(repo_name, cs_cache):
return _render('revision', repo_name, cs_cache.get('revision'),
cs_cache.get('raw_id'), cs_cache.get('author'),
cs_cache.get('message'))
def desc(desc):
return h.urlify_text(desc, truncate=60, stylize=c.visual.stylify_metatags)
def state(repo_state):
return _render("repo_state", repo_state)
def repo_actions(repo_name):
return _render('repo_actions', repo_name, super_user_actions)
def owner_actions(user_id, username):
return _render('user_name', user_id, username)
repos_data = []
for repo in repos_list:
if perm_check:
# check permission at this level
if not HasRepoPermissionAny(
'repository.read', 'repository.write',
'repository.admin'
)(repo.repo_name, 'get_repos_as_dict check'):
continue
cs_cache = repo.changeset_cache
row = {
"menu": quick_menu(repo.repo_name),
"raw_name": repo.repo_name.lower(),
"name": repo_lnk(repo.repo_name, repo.repo_type,
repo.repo_state, repo.private, repo.fork),
"last_change": last_change(repo.last_db_change),
"last_changeset": last_rev(repo.repo_name, cs_cache),
"last_rev_raw": cs_cache.get('revision'),
"desc": desc(repo.description),
"owner": h.person(repo.user),
"state": state(repo.repo_state),
"rss": rss_lnk(repo.repo_name),
"atom": atom_lnk(repo.repo_name),
if admin:
row.update({
"action": repo_actions(repo.repo_name),
"owner": owner_actions(repo.user.user_id,
h.person(repo.user))
})
repos_data.append(row)
return {
"totalRecords": len(repos_list),
"startIndex": 0,
"sort": "name",
"dir": "asc",
"records": repos_data
def _get_defaults(self, repo_name):
Gets information about repository, and returns a dict for
usage in forms
:param repo_name:
repo_info = Repository.get_by_repo_name(repo_name)
if repo_info is None:
return None
defaults = repo_info.get_dict()
group, repo_name, repo_name_full = repo_info.groups_and_repo
defaults['repo_name'] = repo_name
defaults['repo_group'] = getattr(group[-1] if group else None,
'group_id', None)
for strip, k in [(0, 'repo_type'), (1, 'repo_enable_downloads'),
(1, 'repo_description'), (1, 'repo_enable_locking'),
(1, 'repo_landing_rev'), (0, 'clone_uri'),
(1, 'repo_private'), (1, 'repo_enable_statistics')]:
attr = k
if strip:
attr = remove_prefix(k, 'repo_')
val = defaults[attr]
if k == 'repo_landing_rev':
val = ':'.join(defaults[attr])
defaults[k] = val
if k == 'clone_uri':
defaults['clone_uri_hidden'] = repo_info.clone_uri_hidden
# fill owner
if repo_info.user:
defaults.update({'user': repo_info.user.username})
replacement_user = User.query().filter(User.admin ==
True).first().username
defaults.update({'user': replacement_user})
# fill repository users
for p in repo_info.repo_to_perm:
defaults.update({'u_perm_%s' % p.user.username:
p.permission.permission_name})
# fill repository groups
for p in repo_info.users_group_to_perm:
defaults.update({'g_perm_%s' % p.users_group.users_group_name:
return defaults
def update(self, repo, **kwargs):
cur_repo = self._get_repo(repo)
org_repo_name = cur_repo.repo_name
if 'user' in kwargs:
cur_repo.user = User.get_by_username(kwargs['user'])
if 'repo_group' in kwargs:
cur_repo.group = RepoGroup.get(kwargs['repo_group'])
cur_repo.repo_name = cur_repo.get_new_name(cur_repo.just_name)
log.debug('Updating repo %s with params:%s', cur_repo, kwargs)
for k in ['repo_enable_downloads',
'repo_description',
'repo_enable_locking',
'repo_landing_rev',
'repo_private',
'repo_enable_statistics',
]:
if k in kwargs:
setattr(cur_repo, remove_prefix(k, 'repo_'), kwargs[k])
clone_uri = kwargs.get('clone_uri')
if clone_uri is not None and clone_uri != cur_repo.clone_uri_hidden:
cur_repo.clone_uri = clone_uri
new_name = cur_repo.get_new_name(kwargs['repo_name'])
cur_repo.repo_name = new_name
if 'repo_name' in kwargs:
cur_repo.repo_name = cur_repo.get_new_name(kwargs['repo_name'])
#if private flag is set, reset default permission to NONE
if kwargs.get('repo_private'):
EMPTY_PERM = 'repository.none'
RepoModel().grant_user_permission(
repo=cur_repo, user='default', perm=EMPTY_PERM
#handle extra fields
for field in filter(lambda k: k.startswith(RepositoryField.PREFIX),
kwargs):
k = RepositoryField.un_prefix_key(field)
ex_field = RepositoryField.get_by_key_name(key=k, repo=cur_repo)
if ex_field:
ex_field.field_value = kwargs[field]
self.sa.add(ex_field)
self.sa.add(cur_repo)
if org_repo_name != new_name:
if org_repo_name != cur_repo.repo_name:
# rename repository
self._rename_filesystem_repo(old=org_repo_name, new=new_name)
self._rename_filesystem_repo(old=org_repo_name, new=cur_repo.repo_name)
return cur_repo
raise
def _create_repo(self, repo_name, repo_type, description, owner,
private=False, clone_uri=None, repo_group=None,
landing_rev='rev:tip', fork_of=None,
copy_fork_permissions=False, enable_statistics=False,
enable_locking=False, enable_downloads=False,
copy_group_permissions=False, state=Repository.STATE_PENDING):
Create repository inside database with PENDING state. This should only be
executed by create() repo, with exception of importing existing repos.
from kallithea.model.scm import ScmModel
owner = self._get_user(owner)
fork_of = self._get_repo(fork_of)
repo_group = self._get_repo_group(repo_group)
repo_name = safe_unicode(repo_name)
description = safe_unicode(description)
# repo name is just a name of repository
# while repo_name_full is a full qualified name that is combined
# with name and path of group
repo_name_full = repo_name
repo_name = repo_name.split(self.URL_SEPARATOR)[-1]
new_repo = Repository()
new_repo.repo_state = state
new_repo.enable_statistics = False
new_repo.repo_name = repo_name_full
new_repo.repo_type = repo_type
new_repo.user = owner
new_repo.group = repo_group
new_repo.description = description or repo_name
new_repo.private = private
new_repo.clone_uri = clone_uri
new_repo.landing_rev = landing_rev
new_repo.enable_statistics = enable_statistics
new_repo.enable_locking = enable_locking
new_repo.enable_downloads = enable_downloads
if repo_group:
new_repo.enable_locking = repo_group.enable_locking
if fork_of:
parent_repo = fork_of
new_repo.fork = parent_repo
self.sa.add(new_repo)
if fork_of and copy_fork_permissions:
repo = fork_of
user_perms = UserRepoToPerm.query() \
.filter(UserRepoToPerm.repository == repo).all()
group_perms = UserGroupRepoToPerm.query() \
.filter(UserGroupRepoToPerm.repository == repo).all()
for perm in user_perms:
UserRepoToPerm.create(perm.user, new_repo, perm.permission)
for perm in group_perms:
UserGroupRepoToPerm.create(perm.users_group, new_repo,
perm.permission)
elif repo_group and copy_group_permissions:
user_perms = UserRepoGroupToPerm.query() \
.filter(UserRepoGroupToPerm.group == repo_group).all()
group_perms = UserGroupRepoGroupToPerm.query() \
.filter(UserGroupRepoGroupToPerm.group == repo_group).all()
perm_name = perm.permission.permission_name.replace('group.', 'repository.')
perm_obj = Permission.get_by_key(perm_name)
UserRepoToPerm.create(perm.user, new_repo, perm_obj)
UserGroupRepoToPerm.create(perm.users_group, new_repo, perm_obj)
perm_obj = self._create_default_perms(new_repo, private)
self.sa.add(perm_obj)
# now automatically start following this repository as owner
ScmModel(self.sa).toggle_following_repo(new_repo.repo_id,
owner.user_id)
# we need to flush here, in order to check if database won't
# throw any exceptions, create filesystem dirs at the very end
self.sa.flush()
return new_repo
def create(self, form_data, cur_user):
Create repository using celery tasks
:param form_data:
:param cur_user:
from kallithea.lib.celerylib import tasks, run_task
return run_task(tasks.create_repo, form_data, cur_user)
def _update_permissions(self, repo, perms_new=None, perms_updates=None,
check_perms=True):
if not perms_new:
perms_new = []
if not perms_updates:
perms_updates = []
# update permissions
for member, perm, member_type in perms_updates:
if member_type == 'user':
# this updates existing one
self.grant_user_permission(
repo=repo, user=member, perm=perm
#check if we have permissions to alter this usergroup
req_perms = (
'usergroup.read', 'usergroup.write', 'usergroup.admin')
if not check_perms or HasUserGroupPermissionAny(*req_perms)(
member):
self.grant_user_group_permission(
repo=repo, group_name=member, perm=perm
# set new permissions
for member, perm, member_type in perms_new:
def create_fork(self, form_data, cur_user):
Simple wrapper into executing celery task for fork creation
return run_task(tasks.create_repo_fork, form_data, cur_user)
def delete(self, repo, forks=None, fs_remove=True, cur_user=None):
Delete given repository, forks parameter defines what do do with
attached forks. Throws AttachedForksError if deleted repo has attached
forks
:param repo:
:param forks: str 'delete' or 'detach'
:param fs_remove: remove(archive) repo from filesystem
if not cur_user:
cur_user = getattr(get_current_authuser(), 'username', None)
repo = self._get_repo(repo)
if repo is not None:
if forks == 'detach':
for r in repo.forks:
r.fork = None
self.sa.add(r)
elif forks == 'delete':
self.delete(r, forks='delete')
elif [f for f in repo.forks]:
raise AttachedForksError()
old_repo_dict = repo.get_dict()
self.sa.delete(repo)
if fs_remove:
self._delete_filesystem_repo(repo)
log.debug('skipping removal from filesystem')
log_delete_repository(old_repo_dict,
deleted_by=cur_user)
def grant_user_permission(self, repo, user, perm):
if found
:param repo: Instance of Repository, repository_id, or repository name
:param user: Instance of User, user_id or username
:param perm: Instance of Permission, or permission_name
permission = self._get_perm(perm)
# check if we have that permission already
obj = self.sa.query(UserRepoToPerm) \
.filter(UserRepoToPerm.user == user) \
.filter(UserRepoToPerm.repository == repo) \
.scalar()
if obj is None:
# create new !
obj = UserRepoToPerm()
obj.repository = repo
obj.user = user
obj.permission = permission
self.sa.add(obj)
log.debug('Granted perm %s to %s on %s', perm, user, repo)
return obj
def revoke_user_permission(self, repo, user):
Revoke permission for user on given repository
if obj is not None:
self.sa.delete(obj)
log.debug('Revoked perm on %s on %s', repo, user)
def grant_user_group_permission(self, repo, group_name, perm):
existing one if found
:param group_name: Instance of UserGroup, users_group_id,
or user group name
group_name = self._get_user_group(group_name)
obj = self.sa.query(UserGroupRepoToPerm) \
.filter(UserGroupRepoToPerm.users_group == group_name) \
.filter(UserGroupRepoToPerm.repository == repo) \
# create new
obj = UserGroupRepoToPerm()
obj.users_group = group_name
log.debug('Granted perm %s to %s on %s', perm, group_name, repo)
def revoke_user_group_permission(self, repo, group_name):
Revoke permission for user group on given repository
log.debug('Revoked perm to %s on %s', repo, group_name)
def delete_stats(self, repo_name):
removes stats for given repo
repo = self._get_repo(repo_name)
obj = self.sa.query(Statistics) \
.filter(Statistics.repository == repo).scalar()
def _create_filesystem_repo(self, repo_name, repo_type, repo_group,
clone_uri=None, repo_store_location=None):
Makes repository on filesystem. Operation is group aware, meaning that it will create
a repository within a group, and alter the paths accordingly to the group location.
:param alias:
:param repo_store_location:
from kallithea.lib.utils import is_valid_repo, is_valid_repo_group
if '/' in repo_name:
raise ValueError('repo_name must not contain groups got `%s`' % repo_name)
if isinstance(repo_group, RepoGroup):
new_parent_path = os.sep.join(repo_group.full_path_splitted)
new_parent_path = repo_group or ''
if repo_store_location:
_paths = [repo_store_location]
_paths = [self.repos_path, new_parent_path, repo_name]
# we need to make it str for mercurial
repo_path = os.path.join(*map(lambda x: safe_str(x), _paths))
# check if this path is not a repository
if is_valid_repo(repo_path, self.repos_path):
raise Exception('This path %s is a valid repository' % repo_path)
# check if this path is a group
if is_valid_repo_group(repo_path, self.repos_path):
raise Exception('This path %s is a valid group' % repo_path)
log.info('creating repo %s in %s from url: `%s`',
repo_name, safe_unicode(repo_path),
obfuscate_url_pw(clone_uri))
backend = get_backend(repo_type)
if repo_type == 'hg':
baseui = make_ui('db', clear_session=False)
# patch and reset hooks section of UI config to not run any
# hooks on creating remote repo
for k, v in baseui.configitems('hooks'):
baseui.setconfig('hooks', k, None)
repo = backend(repo_path, create=True, src_url=clone_uri, baseui=baseui)
elif repo_type == 'git':
repo = backend(repo_path, create=True, src_url=clone_uri, bare=True)
# add kallithea hook into this repo
ScmModel().install_git_hooks(repo=repo)
raise Exception('Not supported repo_type %s expected hg/git' % repo_type)
log.debug('Created repo %s with %s backend',
safe_unicode(repo_name), safe_unicode(repo_type))
def _rename_filesystem_repo(self, old, new):
renames repository on filesystem
:param old: old name
:param new: new name
log.info('renaming repo from %s to %s', old, new)
old_path = safe_str(os.path.join(self.repos_path, old))
new_path = safe_str(os.path.join(self.repos_path, new))
if os.path.isdir(new_path):
raise Exception(
'Was trying to rename to already existing dir %s' % new_path
shutil.move(old_path, new_path)
def _delete_filesystem_repo(self, repo):
removes repo from filesystem, the removal is actually done by
renaming dir to a 'rm__*' prefix which Kallithea will skip.
It can be undeleted later by reverting the rename.
:param repo: repo object
rm_path = safe_str(os.path.join(self.repos_path, repo.repo_name))
log.info("Removing %s", rm_path)
_now = datetime.now()
_ms = str(_now.microsecond).rjust(6, '0')
_d = 'rm__%s__%s' % (_now.strftime('%Y%m%d_%H%M%S_' + _ms),
repo.just_name)
if repo.group:
args = repo.group.full_path_splitted + [_d]
_d = os.path.join(*args)
shutil.move(rm_path, safe_str(os.path.join(self.repos_path, _d)))
tests for api. run with::
KALLITHEA_WHOOSH_TEST_DISABLE=1 nosetests --with-coverage --cover-package=kallithea.controllers.api.api -x kallithea/tests/api
import random
import mock
from kallithea.tests import *
from kallithea.tests.fixture import Fixture
from kallithea.model.db import Repository, User, Setting
from kallithea.lib.utils2 import time_to_datetime
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
:param random_id:
random_id = random.randrange(1, 9999)
return random_id, json.dumps({
"id": random_id,
"api_key": apikey,
"method": method,
"args": kw
jsonify = lambda obj: json.loads(json.dumps(obj))
def crash(*args, **kwargs):
raise Exception('Total Crash !')
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=TEST_USER_ADMIN_LOGIN)
UserGroupModel().add_user_to_group(user_group=gr,
user=TEST_USER_ADMIN_LOGIN)
return gr
def make_repo_group(name=TEST_REPO_GROUP):
gr = fixture.create_repo_group(name, cur_user=TEST_USER_ADMIN_LOGIN)
class _BaseTestApi(object):
REPO = None
REPO_TYPE = None
def setup_class(cls):
cls.usr = User.get_by_username(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'
cls.TEST_USER_LOGIN = cls.test_user.username
cls.apikey_regular = cls.test_user.api_key
def teardown_class(cls):
def setUp(self):
self.maxDiff = None
make_user_group()
make_repo_group()
def tearDown(self):
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 = json.loads(given)
self.assertEqual(expected, given)
def _compare_error(self, id_, expected, given):
'error': expected,
'result': None
def test_Optional_object(self):
from kallithea.controllers.api.api import Optional
option1 = Optional(None)
self.assertEqual('<Optional:%s>' % None, repr(option1))
self.assertEqual(option1(), None)
self.assertEqual(1, Optional.extract(Optional(1)))
self.assertEqual('trololo', Optional.extract('trololo'))
def test_Optional_OAttr(self):
from kallithea.controllers.api.api import Optional, OAttr
option1 = Optional(OAttr('apiuser'))
self.assertEqual('apiuser', Optional.extract(option1))
def test_OAttr_object(self):
from kallithea.controllers.api.api import OAttr
oattr1 = OAttr('apiuser')
self.assertEqual('<OptionalAttr:apiuser>', repr(oattr1))
self.assertEqual(oattr1(), oattr1)
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')
expected = 'Missing non optional `repoid` arg in JSON DATA'
def test_api_missing_non_optional_param_args_null(self):
params = params.replace('"args": {}', '"args": null')
def test_api_missing_non_optional_param_args_bad(self):
params = params.replace('"args": {}', '"args": 1')
def test_api_args_is_null(self):
id_, params = _build_data(self.apikey, 'get_users', )
self.assertEqual(response.status, '200 OK')
def test_api_args_is_bad(self):
def test_api_args_different_args(self):
import string
expected = {
'ascii_letters': string.ascii_letters,
'ws': string.whitespace,
'printables': string.printable
id_, params = _build_data(self.apikey, 'test', args=expected)
self._compare_ok(id_, expected, response.body)
def test_api_get_users(self):
ret_all = []
_users = User.query().filter(User.username != User.DEFAULT_USER) \
.order_by(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=TEST_USER_ADMIN_LOGIN)
usr = User.get_by_username(TEST_USER_ADMIN_LOGIN)
ret['permissions'] = AuthUser(dbuser=usr).permissions
expected = ret
def test_api_get_user_that_does_not_exist(self):
userid='trololo')
expected = "user `%s` does not exist" % 'trololo'
def test_api_get_user_without_giving_userid(self):
id_, params = _build_data(self.apikey, 'get_user')
def test_api_get_user_without_giving_userid_non_admin(self):
id_, params = _build_data(self.apikey_regular, 'get_user')
usr = User.get_by_username(self.TEST_USER_LOGIN)
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)
expected = 'userid is not the same as your user'
def test_api_pull(self):
repo_name = 'test_pull'
r = fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
r.clone_uri = os.path.join(TESTS_TMP_PATH, self.REPO)
Session.add(r)
Session.commit()
id_, params = _build_data(self.apikey, 'pull',
repoid=repo_name,)
expected = {'msg': 'Pulled from `%s`' % repo_name,
'repository': repo_name}
fixture.destroy_repo(repo_name)
def test_api_pull_error(self):
repoid=self.REPO, )
expected = 'Unable to pull changes from `%s`' % self.REPO
def test_api_rescan_repos(self):
id_, params = _build_data(self.apikey, 'rescan_repos')
expected = {'added': [], 'removed': []}
@mock.patch.object(ScmModel, 'repo_scan', crash)
def test_api_rescann_error(self):
id_, params = _build_data(self.apikey, 'rescan_repos', )
expected = 'Error occurred during rescan repositories action'
def test_api_invalidate_cache(self):
repo = RepoModel().get_by_repo_name(self.REPO)
repo.scm_instance_cached() # seed cache
id_, params = _build_data(self.apikey, 'invalidate_cache',
repoid=self.REPO)
'msg': "Cache for repository `%s` was invalidated" % (self.REPO,),
'repository': self.REPO
@mock.patch.object(ScmModel, 'mark_for_invalidation', crash)
def test_api_invalidate_cache_error(self):
expected = 'Error occurred during cache invalidation action'
def test_api_invalidate_cache_regular_user_no_permission(self):
id_, params = _build_data(self.apikey_regular, 'invalidate_cache',
expected = "repository `%s` does not exist" % (self.REPO,)
def test_api_lock_repo_lock_aquire(self):
id_, params = _build_data(self.apikey, 'lock',
userid=TEST_USER_ADMIN_LOGIN,
repoid=self.REPO,
locked=True)
'repo': self.REPO, 'locked': True,
'locked_since': response.json['result']['locked_since'],
'locked_by': TEST_USER_ADMIN_LOGIN,
% (TEST_USER_ADMIN_LOGIN, self.REPO, True))
def test_api_lock_repo_lock_aquire_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, 'lock',
repoid=repo_name,
'repo': repo_name,
'locked_by': self.TEST_USER_LOGIN,
% (self.TEST_USER_LOGIN, repo_name, True))
finally:
def test_api_lock_repo_lock_aquire_non_admin_with_userid(self):
def test_api_lock_repo_lock_aquire_non_admin_not_his_repo(self):
expected = 'repository `%s` does not exist' % (self.REPO)
def test_api_lock_repo_lock_release(self):
locked=False)
'repo': self.REPO,
% (TEST_USER_ADMIN_LOGIN, self.REPO, False))
def test_api_lock_repo_lock_aquire_optional_userid(self):
time_ = response.json['result']['locked_since']
def test_api_lock_repo_lock_optional_locked(self):
% (self.REPO, TEST_USER_ADMIN_LOGIN,
def test_api_lock_repo_lock_optional_not_locked(self):
repo_name = 'api_not_locked'
repo = fixture.create_repo(repo_name, repo_type=self.REPO_TYPE,
self.assertEqual(repo.locked, [None, None])
repoid=repo.repo_id)
'msg': ('Repo `%s` not locked.' % (repo_name,))
@mock.patch.object(Repository, 'lock', crash)
def test_api_lock_error(self):
expected = 'Error occurred locking repository `%s`' % self.REPO
def test_api_get_locks_regular_user(self):
id_, params = _build_data(self.apikey_regular, 'get_locks')
expected = []
def test_api_get_locks_with_userid_regular_user(self):
id_, params = _build_data(self.apikey_regular, 'get_locks',
def test_api_get_locks(self):
id_, params = _build_data(self.apikey, 'get_locks')
def test_api_get_locks_with_one_locked_repo(self):
Repository.lock(repo, User.get_by_username(self.TEST_USER_LOGIN).user_id)
expected = [repo.get_api_data()]
def test_api_get_locks_with_one_locked_repo_for_specific_user(self):
id_, params = _build_data(self.apikey, 'get_locks',
def test_api_get_locks_with_userid(self):
userid=TEST_USER_REGULAR_LOGIN)
def test_api_create_existing_user(self):
id_, params = _build_data(self.apikey, 'create_user',
username=TEST_USER_ADMIN_LOGIN,
password='trololo')
expected = "user `%s` already exist" % TEST_USER_ADMIN_LOGIN
def test_api_create_user_with_existing_email(self):
username=TEST_USER_ADMIN_LOGIN + 'new',
email=TEST_USER_REGULAR_EMAIL,
expected = "email `%s` already exist" % TEST_USER_REGULAR_EMAIL
def test_api_create_user(self):
username = 'test_new_api_user'
email = username + "@example.com"
username=username,
email=email,
usr = User.get_by_username(username)
ret = dict(
user=jsonify(usr.get_api_data())
fixture.destroy_user(usr.user_id)
def test_api_create_user_without_password(self):
username = 'test_new_api_user_passwordless'
email=email)
def test_api_create_user_with_extern_name(self):
email=email, extern_name='internal')
@mock.patch.object(UserModel, 'create_or_update', crash)
def test_api_create_user_when_exception_happened(self):
expected = 'failed to create user `%s`' % username
def test_api_delete_user(self):
usr = UserModel().create_or_update(username=u'test_user',
password=u'qweqwe',
email=u'u232@example.com',
firstname=u'u1', lastname=u'u1')
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, )
ret = {'msg': 'deleted user ID:%s %s' % (usr_id, username),
'user': None}
@mock.patch.object(UserModel, 'delete', crash)
def test_api_delete_user_when_exception_happened(self):
ret = 'failed to delete user ID:%s %s' % (usr.user_id,
usr.username)
@parameterized.expand([('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):
kw = {name: expected,
'userid': usr.user_id}
id_, params = _build_data(self.apikey, 'update_user', **kw)
ret = {
'msg': 'updated user ID:%s %s' % (
usr.user_id, self.TEST_USER_LOGIN),
'user': jsonify(User \
.get_by_username(self.TEST_USER_LOGIN) \
.get_api_data())
def test_api_update_user_no_changed_params(self):
ret = jsonify(usr.get_api_data())
id_, params = _build_data(self.apikey, 'update_user',
usr.user_id, TEST_USER_ADMIN_LOGIN),
'user': ret
def test_api_update_user_by_user_id(self):
userid=usr.user_id)
def test_api_update_user_default_user(self):
usr = User.get_default_user()
expected = 'editing default user is forbidden'
@mock.patch.object(UserModel, 'update_user', crash)
def test_api_update_user_when_exception_happens(self):
ret = 'failed to update user `%s`' % usr.user_id
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')
id_, params = _build_data(self.apikey, 'get_repo',
ret = repo.get_api_data()
user_data = {'name': user.username, 'type': "user",
'permission': perm}
user_group_data = {'name': user_group.users_group_name,
'type': "user_group", 'permission': perm}
ret['members'] = members
ret['followers'] = followers
fixture.destroy_user_group(new_group)
@parameterized.expand([
('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)
id_, params = _build_data(self.apikey_regular, 'get_repo',
self.assertEqual(2, len(repo.repo_to_perm))
user_obj = user.user
user_data = {'name': user_obj.username, 'type': "user",
user_group_obj = user_group.users_group
user_group_data = {'name': user_group_obj.users_group_name,
RepoModel().revoke_user_permission(self.REPO, self.TEST_USER_LOGIN)
def test_api_get_repo_by_non_admin_no_permission_to_repo(self):
perm='repository.none')
def test_api_get_repo_that_doesn_not_exist(self):
repoid='no-such-repo')
ret = 'repository `%s` does not exist' % 'no-such-repo'
def test_api_get_repos(self):
id_, params = _build_data(self.apikey, 'get_repos')
for repo in RepoModel().get_all():
ret = jsonify(result)
def test_api_get_repos_non_admin(self):
id_, params = _build_data(self.apikey_regular, 'get_repos')
for repo in RepoModel().get_all_user_repos(self.TEST_USER_LOGIN):
@parameterized.expand([('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)
# we don't the actual return types here since it's tested somewhere
# else
expected = response.json['result']
def test_api_get_repo_nodes_bad_revisions(self):
rev = 'i-dont-exist'
root_path=path, )
expected = 'failed to get repo: `%s` nodes' % self.REPO
def test_api_get_repo_nodes_bad_path(self):
path = '/idontexits'
def test_api_get_repo_nodes_bad_ret_type(self):
ret_type = 'error'
expected = ('ret_type must be one of %s'
% (','.join(['files', 'dirs', 'all'])))
@parameterized.expand([('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):
id_, params = _build_data(self.apikey_regular, 'get_repo_nodes',
def test_api_create_repo(self):
repo_name = 'api-repo'
id_, params = _build_data(self.apikey, 'create_repo',
repo_name=repo_name,
owner=TEST_USER_ADMIN_LOGIN,
repo_type=self.REPO_TYPE,
repo = RepoModel().get_by_repo_name(repo_name)
self.assertNotEqual(repo, None)
'msg': 'Created new repository `%s`' % repo_name,
'success': True,
'task': None,
def test_api_create_repo_and_repo_group(self):
repo_name = 'my_gr/api-repo'
repo_type=self.REPO_TYPE,)
print params
fixture.destroy_repo_group('my_gr')
def test_api_create_repo_in_repo_group_without_permission(self):
repo_group_name = '%s/api-repo-repo' % TEST_REPO_GROUP
repo_name = '%s/api-repo' % repo_group_name
rg = fixture.create_repo_group(repo_group_name)
RepoGroupModel().grant_user_permission(repo_group_name,
self.TEST_USER_LOGIN,
'group.none')
id_, params = _build_data(self.apikey_regular, 'create_repo',
# Current result when API access control is different from Web:
# 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):
owner = 'i-dont-exist'
expected = 'user `%s` does not exist' % owner
def test_api_create_repo_dont_specify_owner(self):
def test_api_create_repo_by_non_admin(self):
def test_api_create_repo_by_non_admin_specify_owner(self):
owner=owner)
expected = 'Only Kallithea admin can specify `owner` param'
def test_api_create_repo_exists(self):
repo_name = self.REPO
expected = "repo `%s` already exist" % repo_name
@mock.patch.object(RepoModel, 'create', crash)
def test_api_create_repo_exception_occurred(self):
expected = 'failed to create repository `%s`' % repo_name
('owner', {'owner': TEST_USER_REGULAR_LOGIN}),
('description', {'description': 'new description'}),
('active', {'active': True}),
('active', {'active': False}),
('clone_uri', {'clone_uri': 'http://example.com/repo'}),
('clone_uri', {'clone_uri': None}),
('landing_rev', {'landing_rev': 'branch:master'}),
('enable_statistics', {'enable_statistics': True}),
('enable_locking', {'enable_locking': 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)
if changing_attr == 'name':
repo_name = updates['name']
repo_name = '/'.join([updates['group'], repo_name])
'msg': 'updated repo ID:%s %s' % (repo.repo_id, repo_name),
'repository': repo.get_api_data()
fixture.destroy_repo_group(updates['group'])
('description', {'description': u'new description'}),
('name', {'name': u'new_repo_name'}),
('repo_group', {'group': u'test_group_for_update'}),
def test_api_update_group_repo(self, changing_attr, updates):
group_name = u'lololo'
fixture.create_repo_group(group_name)
repo_name = u'%s/api_update_me' % group_name
repo = fixture.create_repo(repo_name, repo_group=group_name, repo_type=self.REPO_TYPE)
repo_name = u'%s/%s' % (group_name, updates['name'])
repo_name = u'/'.join([updates['group'], repo_name.rsplit('/', 1)[-1]])
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'}
expected = 'repository group `%s` does not exist' % updates['group']
def test_api_update_repo_regular_user_not_allowed(self):
updates = {'active': False}
id_, params = _build_data(self.apikey_regular, 'update_repo',
expected = 'repository `%s` does not exist' % repo_name
@mock.patch.object(RepoModel, 'update', crash)
def test_api_update_repo_exception_occurred(self):
fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
repoid=repo_name, owner=TEST_USER_ADMIN_LOGIN,)
expected = 'failed to update repo `%s`' % repo_name
def test_api_update_repo_regular_user_change_repo_name(self):
new_repo_name = 'new_repo_name'
RepoModel().grant_user_permission(repo=repo_name,
perm='repository.admin')
UserModel().revoke_perm('default', 'hg.create.repository')
UserModel().grant_perm('default', 'hg.create.none')
updates = {'name': new_repo_name}
expected = 'no permission to create (or move) repositories'
fixture.destroy_repo(new_repo_name)
def test_api_update_repo_regular_user_change_repo_name_allowed(self):
UserModel().revoke_perm('default', 'hg.create.none')
UserModel().grant_perm('default', 'hg.create.repository')
'msg': 'updated repo ID:%s %s' % (repo.repo_id, new_repo_name),
def test_api_update_repo_regular_user_change_owner(self):
updates = {'owner': TEST_USER_ADMIN_LOGIN}
def test_api_delete_repo(self):
id_, params = _build_data(self.apikey, 'delete_repo',
repoid=repo_name, )
'msg': 'Deleted repository `%s`' % repo_name,
'success': True
def test_api_delete_repo_by_non_admin(self):
id_, params = _build_data(self.apikey_regular, 'delete_repo',
def test_api_delete_repo_by_non_admin_no_permission(self):
expected = 'repository `%s` does not exist' % (repo_name)
def test_api_delete_repo_exception_occurred(self):
with mock.patch.object(RepoModel, 'delete', crash):
expected = 'failed to delete repository `%s`' % repo_name
def test_api_fork_repo(self):
fork_name = 'api-repo-fork'
id_, params = _build_data(self.apikey, 'fork_repo',
fork_name=fork_name,
'msg': 'Created fork of `%s` as `%s`' % (self.REPO,
fixture.destroy_repo(fork_name)
def test_api_fork_repo_non_admin(self):
id_, params = _build_data(self.apikey_regular, 'fork_repo',
def test_api_fork_repo_non_admin_specify_owner(self):
def test_api_fork_repo_non_admin_no_permission_to_fork(self):
@parameterized.expand([('read', 'repository.read'),
('write', 'repository.write'),
('admin', 'repository.admin')])
def test_api_fork_repo_non_admin_no_create_repo_permission(self, name, perm):
# regardless of base repository permission, forking is disallowed
# when repository creation is disabled
perm=perm)
expected = 'no permission to create repositories'
def test_api_fork_repo_unknown_owner(self):
def test_api_fork_repo_fork_exists(self):
fixture.create_fork(self.REPO, fork_name)
expected = "fork `%s` already exist" % fork_name
def test_api_fork_repo_repo_exists(self):
fork_name = self.REPO
expected = "repo `%s` already exist" % fork_name
@mock.patch.object(RepoModel, 'create_fork', crash)
def test_api_fork_repo_exception_occurred(self):
expected = 'failed to fork repository `%s` as `%s`' % (self.REPO,
def test_api_get_user_group(self):
id_, params = _build_data(self.apikey, 'get_user_group',
usergroupid=TEST_USER_GROUP)
user_group = UserGroupModel().get_group(TEST_USER_GROUP)
for user in user_group.members:
members.append(user.get_api_data())
ret = user_group.get_api_data()
def test_api_get_user_groups(self):
gr_name = 'test_user_group2'
make_user_group(gr_name)
id_, params = _build_data(self.apikey, 'get_user_groups', )
for gr_name in [TEST_USER_GROUP, 'test_user_group2']:
user_group = UserGroupModel().get_group(gr_name)
expected.append(ret)
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)
'msg': 'created new user group `%s`' % group_name,
'user_group': jsonify(UserGroupModel() \
.get_by_name(group_name) \
fixture.destroy_user_group(group_name)
def test_api_get_user_group_that_exist(self):
group_name=TEST_USER_GROUP)
expected = "user group `%s` already exist" % TEST_USER_GROUP
@mock.patch.object(UserGroupModel, 'create', crash)
def test_api_get_user_group_exception_occurred(self):
group_name = 'exception_happens'
expected = 'failed to create group `%s`' % group_name
@parameterized.expand([('group_name', {'group_name': 'new_group_name'}),
('group_name', {'group_name': 'test_group_for_update'}),
('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)
id_, params = _build_data(self.apikey, 'update_user_group',
usergroupid=gr_name, **updates)
'msg': 'updated user group ID:%s %s' % (user_group.users_group_id,
'user_group': user_group.get_api_data()
if changing_attr == 'group_name':
# switch to updated name for proper cleanup
gr_name = updates['group_name']
@mock.patch.object(UserGroupModel, 'update', crash)
def test_api_update_user_group_exception_occurred(self):
gr_name = 'test_group'
fixture.create_user_group(gr_name)
usergroupid=gr_name)
expected = 'failed to update user group `%s`' % gr_name
def test_api_add_user_to_user_group(self):
id_, params = _build_data(self.apikey, 'add_user_to_user_group',
usergroupid=gr_name,
'msg': 'added member `%s` to user group `%s`' % (
TEST_USER_ADMIN_LOGIN, gr_name),
def test_api_add_user_to_user_group_that_doesnt_exist(self):
usergroupid='false-group',
expected = 'user group `%s` does not exist' % 'false-group'
@mock.patch.object(UserGroupModel, 'add_user_to_group', crash)
def test_api_add_user_to_user_group_exception_occurred(self):
expected = 'failed to add member to user group `%s`' % 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=TEST_USER_ADMIN_LOGIN)
id_, params = _build_data(self.apikey, 'remove_user_from_user_group',
'msg': 'removed member `%s` from user group `%s`' % (
TEST_USER_ADMIN_LOGIN, gr_name
'success': True}
@mock.patch.object(UserGroupModel, 'remove_user_from_group', crash)
def test_api_remove_user_from_user_group_exception_occurred(self):
expected = 'failed to remove member from user group `%s`' % gr_name
def test_api_delete_user_group(self):
ugroup = fixture.create_user_group(gr_name)
gr_id = ugroup.users_group_id
id_, params = _build_data(self.apikey, 'delete_user_group',
'user_group': None,
'msg': 'deleted user group ID:%s %s' % (gr_id, gr_name)
if UserGroupModel().get_by_name(gr_name):
def test_api_delete_user_group_that_is_assigned(self):
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
expected = msg
def test_api_delete_user_group_exception_occurred(self):
with mock.patch.object(UserGroupModel, 'delete', crash):
expected = 'failed to delete user group ID:%s %s' % (gr_id, gr_name)
@parameterized.expand([('none', 'repository.none'),
('read', 'repository.read'),
def test_api_grant_user_permission(self, name, perm):
id_, params = _build_data(self.apikey,
'grant_user_permission',
'msg': 'Granted perm: `%s` for user: `%s` in repo: `%s`' % (
perm, TEST_USER_ADMIN_LOGIN, self.REPO
def test_api_grant_user_permission_wrong_permission(self):
perm = 'haha.no.permission'
expected = 'permission `%s` does not exist' % perm
@mock.patch.object(RepoModel, 'grant_user_permission', crash)
def test_api_grant_user_permission_exception_when_adding(self):
perm = 'repository.read'
expected = 'failed to edit permission for user: `%s` in repo: `%s`' % (
TEST_USER_ADMIN_LOGIN, self.REPO
def test_api_revoke_user_permission(self):
'revoke_user_permission',
userid=TEST_USER_ADMIN_LOGIN, )
'msg': 'Revoked perm for user: `%s` in repo: `%s`' % (
@mock.patch.object(RepoModel, 'revoke_user_permission', crash)
def test_api_revoke_user_permission_exception_when_adding(self):
def test_api_grant_user_group_permission(self, name, perm):
'grant_user_group_permission',
usergroupid=TEST_USER_GROUP,
'msg': 'Granted perm: `%s` for user group: `%s` in repo: `%s`' % (
perm, TEST_USER_GROUP, self.REPO
def test_api_grant_user_group_permission_wrong_permission(self):
@mock.patch.object(RepoModel, 'grant_user_group_permission', crash)
def test_api_grant_user_group_permission_exception_when_adding(self):
expected = 'failed to edit permission for user group: `%s` in repo: `%s`' % (
TEST_USER_GROUP, self.REPO
def test_api_revoke_user_group_permission(self):
group_name=TEST_USER_GROUP,
'revoke_user_group_permission',
usergroupid=TEST_USER_GROUP, )
'msg': 'Revoked perm for user group: `%s` in repo: `%s`' % (
@mock.patch.object(RepoModel, 'revoke_user_group_permission', crash)
def test_api_revoke_user_group_permission_exception_when_adding(self):
('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):
'grant_user_permission_to_repo_group',
repogroupid=TEST_REPO_GROUP,
perm=perm, apply_to_children=apply_to_children)
'msg': 'Granted perm: `%s` (recursive:%s) for user: `%s` in repo group: `%s`' % (
perm, apply_to_children, TEST_USER_ADMIN_LOGIN, TEST_REPO_GROUP
('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,
'group.admin')
id_, params = _build_data(self.apikey_regular,
if access_ok:
expected = 'repository group `%s` does not exist' % TEST_REPO_GROUP
def test_api_grant_user_permission_to_repo_group_wrong_permission(self):
@mock.patch.object(RepoGroupModel, 'grant_user_permission', crash)
def test_api_grant_user_permission_to_repo_group_exception_when_adding(self):
perm = 'group.read'
expected = 'failed to edit permission for user: `%s` in repo group: `%s`' % (
TEST_USER_ADMIN_LOGIN, TEST_REPO_GROUP
('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=TEST_USER_ADMIN_LOGIN,
perm='group.read',)
'revoke_user_permission_from_repo_group',
apply_to_children=apply_to_children,)
'msg': 'Revoked perm (recursive:%s) for user: `%s` in repo group: `%s`' % (
apply_to_children, TEST_USER_ADMIN_LOGIN, TEST_REPO_GROUP
('none', 'none', False, False),
('all', 'all', False, False),
('repos', 'repos', False, False),
('groups', 'groups', False, False),
# after granting admin rights
def test_api_revoke_user_permission_from_repo_group_by_regular_user(
self, name, apply_to_children, grant_admin, access_ok):
@mock.patch.object(RepoGroupModel, 'revoke_user_permission', crash)
def test_api_revoke_user_permission_from_repo_group_exception_when_adding(self):
def test_api_grant_user_group_permission_to_repo_group(self, name, perm, apply_to_children):
'grant_user_group_permission_to_repo_group',
'msg': 'Granted perm: `%s` (recursive:%s) for user group: `%s` in repo group: `%s`' % (
perm, apply_to_children, TEST_USER_GROUP, TEST_REPO_GROUP
def test_api_grant_user_group_permission_to_repo_group_by_regular_user(
def test_api_grant_user_group_permission_to_repo_group_wrong_permission(self):
@mock.patch.object(RepoGroupModel, 'grant_user_group_permission', crash)
def test_api_grant_user_group_permission_exception_when_adding_to_repo_group(self):
expected = 'failed to edit permission for user group: `%s` in repo group: `%s`' % (
TEST_USER_GROUP, TEST_REPO_GROUP
def test_api_revoke_user_group_permission_from_repo_group(self, name, apply_to_children):
RepoGroupModel().grant_user_group_permission(repo_group=TEST_REPO_GROUP,
'revoke_user_group_permission_from_repo_group',
'msg': 'Revoked perm (recursive:%s) for user group: `%s` in repo group: `%s`' % (
apply_to_children, TEST_USER_GROUP, TEST_REPO_GROUP
def test_api_revoke_user_group_permission_from_repo_group_by_regular_user(
@mock.patch.object(RepoGroupModel, 'revoke_user_group_permission', crash)
def test_api_revoke_user_group_permission_from_repo_group_exception_when_adding(self):
id_, params = _build_data(self.apikey, 'revoke_user_group_permission_from_repo_group',
usergroupid=TEST_USER_GROUP,)
def test_api_get_gist(self):
gist = fixture.create_gist()
gist_id = gist.gist_access_id
gist_created_on = gist.created_on
id_, params = _build_data(self.apikey, 'get_gist',
gistid=gist_id, )
'access_id': gist_id,
'created_on': gist_created_on,
'description': 'new-gist',
'expires': -1.0,
'gist_id': int(gist_id),
'type': 'public',
'url': 'http://localhost:80/_admin/gists/%s' % gist_id
def test_api_get_gist_that_does_not_exist(self):
id_, params = _build_data(self.apikey_regular, 'get_gist',
gistid='12345', )
expected = 'gist `%s` does not exist' % ('12345',)
def test_api_get_gist_private_gist_without_permission(self):
expected = 'gist `%s` does not exist' % gist_id
def test_api_get_gists(self):
fixture.create_gist()
id_, params = _build_data(self.apikey, 'get_gists')
expected = response.json
self.assertEqual(len(response.json['result']), 2)
#self._compare_ok(id_, expected, given=response.body)
def test_api_get_gists_regular_user(self):
# by admin
# by reg user
fixture.create_gist(owner=self.TEST_USER_LOGIN)
id_, params = _build_data(self.apikey_regular, 'get_gists')
self.assertEqual(len(response.json['result']), 3)
def test_api_get_gists_only_for_regular_user(self):
id_, params = _build_data(self.apikey, 'get_gists',
def test_api_get_gists_regular_user_with_different_userid(self):
id_, params = _build_data(self.apikey_regular, 'get_gists',
def test_api_create_gist(self):
id_, params = _build_data(self.apikey_regular, 'create_gist',
lifetime=10,
description='foobar-gist',
gist_type='public',
files={'foobar': {'content': 'foo'}})
response_json = response.json
'gist': {
'access_id': response_json['result']['gist']['access_id'],
'created_on': response_json['result']['gist']['created_on'],
'description': 'foobar-gist',
'expires': response_json['result']['gist']['expires'],
'gist_id': response_json['result']['gist']['gist_id'],
'url': response_json['result']['gist']['url']
'msg': 'created new gist'
@mock.patch.object(GistModel, 'create', crash)
def test_api_create_gist_exception_occurred(self):
files={})
expected = 'failed to create gist'
def test_api_delete_gist(self):
gist_id = fixture.create_gist().gist_access_id
id_, params = _build_data(self.apikey, 'delete_gist',
gistid=gist_id)
expected = {'gist': None, 'msg': 'deleted gist ID:%s' % gist_id}
def test_api_delete_gist_regular_user(self):
gist_id = fixture.create_gist(owner=self.TEST_USER_LOGIN).gist_access_id
id_, params = _build_data(self.apikey_regular, 'delete_gist',
def test_api_delete_gist_regular_user_no_permission(self):
expected = 'gist `%s` does not exist' % (gist_id,)
@mock.patch.object(GistModel, 'delete', crash)
def test_api_delete_gist_exception_occurred(self):
expected = 'failed to delete gist ID:%s' % (gist_id,)
def test_api_get_ip(self):
id_, params = _build_data(self.apikey, 'get_ip')
'server_ip_addr': '0.0.0.0',
'user_ips': []
def test_api_get_server_info(self):
id_, params = _build_data(self.apikey, 'get_server_info')
expected = Setting.get_server_info()
Status change: