Mercurial > kallithea
changeset 2653:8be70a4d72c7 beta
Implemented #354 added forking option to API
| author | Marcin Kuzminski <marcin@python-works.com> |
|---|---|
| date | Sun, 22 Jul 2012 01:16:36 +0200 |
| parents | 532ee86cf9b1 |
| children | 731e823d330e |
| files | rhodecode/controllers/api/api.py rhodecode/tests/api/api_base.py |
| diffstat | 2 files changed, 138 insertions(+), 27 deletions(-) [+] |
line wrap: on
line diff
--- a/rhodecode/controllers/api/api.py Sun Jul 22 01:15:53 2012 +0200 +++ b/rhodecode/controllers/api/api.py Sun Jul 22 01:16:36 2012 +0200 @@ -558,26 +558,48 @@ log.error(traceback.format_exc()) raise JSONRPCError('failed to create repository `%s`' % repo_name) -# @HasPermissionAnyDecorator('hg.admin') -# def fork_repo(self, apiuser, repoid, fork_name): -# repo = get_repo_or_error(repoid) -# -# try: -# form_data = dict( -# -# ) -# RepoModel().create_fork(form_data, cur_user=apiuser) -# return dict( -# msg='Created fork of `%s` as `%s`' % (repo.repo_name, -# fork_name), -# success=True -# ) -# except Exception: -# log.error(traceback.format_exc()) -# raise JSONRPCError( -# 'failed to fork repository `%s` as `%s`' % (repo.repo_name, -# fork_name) -# ) + @HasPermissionAnyDecorator('hg.admin') + def fork_repo(self, apiuser, repoid, fork_name, owner, + description=Optional(''), copy_permissions=Optional(False), + private=Optional(False), landing_rev=Optional('tip')): + repo = get_repo_or_error(repoid) + repo_name = repo.repo_name + owner = get_user_or_error(owner) + + _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)) + + try: + # create structure of groups and return the last group + 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, + ) + RepoModel().create_fork(form_data, cur_user=owner) + return dict( + msg='Created fork of `%s` as `%s`' % (repo.repo_name, + fork_name), + success=True # cannot return the repo data here since fork + # cann be done async + ) + except Exception: + log.error(traceback.format_exc()) + raise JSONRPCError( + 'failed to fork repository `%s` as `%s`' % (repo_name, + fork_name) + ) @HasPermissionAnyDecorator('hg.admin') def delete_repo(self, apiuser, repoid):
--- a/rhodecode/tests/api/api_base.py Sun Jul 22 01:15:53 2012 +0200 +++ b/rhodecode/tests/api/api_base.py Sun Jul 22 01:16:36 2012 +0200 @@ -68,6 +68,15 @@ return r +def create_fork(fork_name, fork_type, fork_of): + fork = RepoModel(Session())._get_repo(fork_of) + r = create_repo(fork_name, fork_type) + r.fork = fork + Session().add(r) + Session().commit() + return r + + def destroy_repo(repo_name): RepoModel().delete(repo_name) Session().commit() @@ -571,18 +580,98 @@ def test_api_delete_repo_exception_occurred(self): repo_name = 'api_delete_me' create_repo(repo_name, self.REPO_TYPE) - with mock.patch.object(RepoModel, 'delete', crash): - id_, params = _build_data(self.apikey, 'delete_repo', - repoid=repo_name,) + try: + with mock.patch.object(RepoModel, 'delete', crash): + id_, params = _build_data(self.apikey, 'delete_repo', + repoid=repo_name,) + response = self.app.post(API_URL, content_type='application/json', + params=params) + + expected = 'failed to delete repository `%s`' % repo_name + self._compare_error(id_, expected, given=response.body) + finally: + destroy_repo(repo_name) + + def test_api_fork_repo(self): + fork_name = 'api-repo-fork' + id_, params = _build_data(self.apikey, 'fork_repo', + repoid=self.REPO, + fork_name=fork_name, + owner=TEST_USER_ADMIN_LOGIN, + ) + response = self.app.post(API_URL, content_type='application/json', + params=params) + + ret = { + 'msg': 'Created fork of `%s` as `%s`' % (self.REPO, + fork_name), + 'success': True + } + expected = ret + self._compare_ok(id_, expected, given=response.body) + destroy_repo(fork_name) + + def test_api_fork_repo_unknown_owner(self): + fork_name = 'api-repo-fork' + owner = 'i-dont-exist' + id_, params = _build_data(self.apikey, 'fork_repo', + repoid=self.REPO, + fork_name=fork_name, + owner=owner, + ) + response = self.app.post(API_URL, content_type='application/json', + params=params) + expected = 'user `%s` does not exist' % owner + self._compare_error(id_, expected, given=response.body) + + def test_api_fork_repo_fork_exists(self): + fork_name = 'api-repo-fork' + create_fork(fork_name, self.REPO_TYPE, self.REPO) + + try: + fork_name = 'api-repo-fork' + + id_, params = _build_data(self.apikey, 'fork_repo', + repoid=self.REPO, + fork_name=fork_name, + owner=TEST_USER_ADMIN_LOGIN, + ) response = self.app.post(API_URL, content_type='application/json', params=params) - expected = 'failed to delete repository `%s`' % repo_name + expected = "fork `%s` already exist" % fork_name self._compare_error(id_, expected, given=response.body) - destroy_repo(repo_name) + finally: + destroy_repo(fork_name) + + def test_api_fork_repo_repo_exists(self): + fork_name = self.REPO + + id_, params = _build_data(self.apikey, 'fork_repo', + repoid=self.REPO, + fork_name=fork_name, + owner=TEST_USER_ADMIN_LOGIN, + ) + response = self.app.post(API_URL, content_type='application/json', + params=params) - def test_api_fork_repo(self): - self.failIf(False, 'TODO:') + expected = "repo `%s` already exist" % fork_name + self._compare_error(id_, expected, given=response.body) + + @mock.patch.object(RepoModel, 'create_fork', crash) + def test_api_fork_repo_exception_occurred(self): + fork_name = 'api-repo-fork' + id_, params = _build_data(self.apikey, 'fork_repo', + repoid=self.REPO, + fork_name=fork_name, + owner=TEST_USER_ADMIN_LOGIN, + ) + response = self.app.post(API_URL, content_type='application/json', + params=params) + + expected = 'failed to fork repository `%s` as `%s`' % (self.REPO, + fork_name) + self._compare_error(id_, expected, given=response.body) def test_api_get_users_group(self): id_, params = _build_data(self.apikey, 'get_users_group',