get-perms-api-helper: show nicer message when prefix is specified to help in API usage.

rhodecode/api/tests/test_grant_user_group_permission.py

@@ -1,90 +1,90 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2019 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import mock
 import pytest
 
 from rhodecode.model.repo import RepoModel
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_error, assert_ok, crash)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestGrantUserGroupPermission(object):
     @pytest.mark.parametrize("name, perm", [
         ('none', 'repository.none'),
         ('read', 'repository.read'),
         ('write', 'repository.write'),
         ('admin', 'repository.admin')
     ])
     def test_api_grant_user_group_permission(
             self, name, perm, backend, user_util):
         user_group = user_util.create_user_group()
         id_, params = build_data(
             self.apikey,
             'grant_user_group_permission',
             repoid=backend.repo_name,
             usergroupid=user_group.users_group_name,
             perm=perm)
         response = api_call(self.app, params)
 
         ret = {
             'msg': 'Granted perm: `%s` for user group: `%s` in repo: `%s`' % (
                 perm, user_group.users_group_name, backend.repo_name
             ),
             'success': True
         }
         expected = ret
         assert_ok(id_, expected, given=response.body)
 
     def test_api_grant_user_group_permission_wrong_permission(
             self, backend, user_util):
         perm = 'haha.no.permission'
         user_group = user_util.create_user_group()
         id_, params = build_data(
             self.apikey,
             'grant_user_group_permission',
             repoid=backend.repo_name,
             usergroupid=user_group.users_group_name,
             perm=perm)
         response = api_call(self.app, params)
 
-        expected = 'permission `%s` does not exist' % (perm,)
+        expected = 'permission `%s` does not exist.' % (perm,)
         assert_error(id_, expected, given=response.body)
 
     @mock.patch.object(RepoModel, 'grant_user_group_permission', crash)
     def test_api_grant_user_group_permission_exception_when_adding(
             self, backend, user_util):
         perm = 'repository.read'
         user_group = user_util.create_user_group()
         id_, params = build_data(
             self.apikey,
             'grant_user_group_permission',
             repoid=backend.repo_name,
             usergroupid=user_group.users_group_name,
             perm=perm)
         response = api_call(self.app, params)
 
         expected = (
             'failed to edit permission for user group: `%s` in repo: `%s`' % (
                 user_group.users_group_name, backend.repo_name
             )
         )
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_grant_user_group_permission_to_repo_group.py

@@ -1,174 +1,173 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2019 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import mock
 import pytest
 
 from rhodecode.model.user import UserModel
 from rhodecode.model.repo_group import RepoGroupModel
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_error, assert_ok, crash)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestGrantUserGroupPermissionFromRepoGroup(object):
     @pytest.mark.parametrize("name, perm, apply_to_children", [
         ('none', 'group.none', 'none'),
         ('read', 'group.read', 'none'),
         ('write', 'group.write', 'none'),
         ('admin', 'group.admin', 'none'),
 
         ('none', 'group.none', 'all'),
         ('read', 'group.read', 'all'),
         ('write', 'group.write', 'all'),
         ('admin', 'group.admin', 'all'),
 
         ('none', 'group.none', 'repos'),
         ('read', 'group.read', 'repos'),
         ('write', 'group.write', 'repos'),
         ('admin', 'group.admin', 'repos'),
 
         ('none', 'group.none', 'groups'),
         ('read', 'group.read', 'groups'),
         ('write', 'group.write', 'groups'),
         ('admin', 'group.admin', 'groups'),
     ])
     def test_api_grant_user_group_permission_to_repo_group(
             self, name, perm, apply_to_children, user_util):
         user_group = user_util.create_user_group()
         repo_group = user_util.create_repo_group()
         user_util.create_repo(parent=repo_group)
 
         id_, params = build_data(
             self.apikey,
             'grant_user_group_permission_to_repo_group',
             repogroupid=repo_group.name,
             usergroupid=user_group.users_group_name,
             perm=perm,
             apply_to_children=apply_to_children,)
         response = api_call(self.app, params)
 
         ret = {
             'msg': (
                 'Granted perm: `%s` (recursive:%s) for user group: `%s`'
                 ' in repo group: `%s`' % (
                     perm, apply_to_children, user_group.users_group_name,
                     repo_group.name
                 )
             ),
             'success': True
         }
         expected = ret
         try:
             assert_ok(id_, expected, given=response.body)
         finally:
             RepoGroupModel().revoke_user_group_permission(
                 repo_group.group_id, user_group.users_group_id)
 
     @pytest.mark.parametrize(
         "name, perm, apply_to_children, grant_admin, access_ok", [
             ('none_fails', 'group.none', 'none', False, False),
             ('read_fails', 'group.read', 'none', False, False),
             ('write_fails', 'group.write', 'none', False, False),
             ('admin_fails', 'group.admin', 'none', False, False),
 
             # with granted perms
             ('none_ok', 'group.none', 'none', True, True),
             ('read_ok', 'group.read', 'none', True, True),
             ('write_ok', 'group.write', 'none', True, True),
             ('admin_ok', 'group.admin', 'none', True, True),
         ]
     )
     def test_api_grant_user_group_permission_to_repo_group_by_regular_user(
             self, name, perm, apply_to_children, grant_admin, access_ok,
             user_util):
         user = UserModel().get_by_username(self.TEST_USER_LOGIN)
         user_group = user_util.create_user_group()
         repo_group = user_util.create_repo_group()
         if grant_admin:
             user_util.grant_user_permission_to_repo_group(
                 repo_group, user, 'group.admin')
 
         id_, params = build_data(
             self.apikey_regular,
             'grant_user_group_permission_to_repo_group',
             repogroupid=repo_group.name,
             usergroupid=user_group.users_group_name,
             perm=perm,
             apply_to_children=apply_to_children,)
         response = api_call(self.app, params)
         if access_ok:
             ret = {
                 'msg': (
                     'Granted perm: `%s` (recursive:%s) for user group: `%s`'
                     ' in repo group: `%s`' % (
                         perm, apply_to_children, user_group.users_group_name,
                         repo_group.name
                     )
                 ),
                 'success': True
             }
             expected = ret
             try:
                 assert_ok(id_, expected, given=response.body)
             finally:
                 RepoGroupModel().revoke_user_group_permission(
                     repo_group.group_id, user_group.users_group_id)
         else:
-            expected = 'repository group `%s` does not exist' % (
-                repo_group.name,)
+            expected = 'repository group `%s` does not exist' % (repo_group.name,)
             assert_error(id_, expected, given=response.body)
 
     def test_api_grant_user_group_permission_to_repo_group_wrong_permission(
             self, user_util):
         user_group = user_util.create_user_group()
         repo_group = user_util.create_repo_group()
         perm = 'haha.no.permission'
         id_, params = build_data(
             self.apikey,
             'grant_user_group_permission_to_repo_group',
             repogroupid=repo_group.name,
             usergroupid=user_group.users_group_name,
             perm=perm)
         response = api_call(self.app, params)
 
-        expected = 'permission `%s` does not exist' % (perm,)
+        expected = 'permission `%s` does not exist. Permission should start with prefix: `group.`' % (perm,)
         assert_error(id_, expected, given=response.body)
 
     @mock.patch.object(RepoGroupModel, 'grant_user_group_permission', crash)
     def test_api_grant_user_group_permission_exception_when_adding_2(
             self, user_util):
         user_group = user_util.create_user_group()
         repo_group = user_util.create_repo_group()
         perm = 'group.read'
         id_, params = build_data(
             self.apikey,
             'grant_user_group_permission_to_repo_group',
             repogroupid=repo_group.name,
             usergroupid=user_group.users_group_name,
             perm=perm)
         response = api_call(self.app, params)
 
         expected = (
             'failed to edit permission for user group: `%s`'
             ' in repo group: `%s`' % (
                 user_group.users_group_name, repo_group.name)
         )
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_grant_user_permission.py

@@ -1,87 +1,87 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2019 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import mock
 import pytest
 
 from rhodecode.model.repo import RepoModel
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_error, assert_ok, crash)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestGrantUserPermission(object):
     @pytest.mark.parametrize("name, perm", [
         ('none', 'repository.none'),
         ('read', 'repository.read'),
         ('write', 'repository.write'),
         ('admin', 'repository.admin')
     ])
     def test_api_grant_user_permission(self, name, perm, backend, user_util):
         user = user_util.create_user()
         id_, params = build_data(
             self.apikey,
             'grant_user_permission',
             repoid=backend.repo_name,
             userid=user.username,
             perm=perm)
         response = api_call(self.app, params)
 
         ret = {
             'msg': 'Granted perm: `%s` for user: `%s` in repo: `%s`' % (
                 perm, user.username, backend.repo_name
             ),
             'success': True
         }
         expected = ret
         assert_ok(id_, expected, given=response.body)
 
     def test_api_grant_user_permission_wrong_permission(
             self, backend, user_util):
         user = user_util.create_user()
         perm = 'haha.no.permission'
         id_, params = build_data(
             self.apikey,
             'grant_user_permission',
             repoid=backend.repo_name,
             userid=user.username,
             perm=perm)
         response = api_call(self.app, params)
 
-        expected = 'permission `%s` does not exist' % (perm,)
+        expected = 'permission `%s` does not exist.' % (perm,)
         assert_error(id_, expected, given=response.body)
 
     @mock.patch.object(RepoModel, 'grant_user_permission', crash)
     def test_api_grant_user_permission_exception_when_adding(
             self, backend, user_util):
         user = user_util.create_user()
         perm = 'repository.read'
         id_, params = build_data(
             self.apikey,
             'grant_user_permission',
             repoid=backend.repo_name,
             userid=user.username,
             perm=perm)
         response = api_call(self.app, params)
 
         expected = 'failed to edit permission for user: `%s` in repo: `%s`' % (
             user.username, backend.repo_name
         )
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_grant_user_permission_to_repo_group.py

@@ -1,157 +1,157 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2019 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import mock
 import pytest
 
 from rhodecode.model.user import UserModel
 from rhodecode.model.repo_group import RepoGroupModel
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_error, assert_ok, crash)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestGrantUserPermissionFromRepoGroup(object):
     @pytest.mark.parametrize("name, perm, apply_to_children", [
         ('none', 'group.none', 'none'),
         ('read', 'group.read', 'none'),
         ('write', 'group.write', 'none'),
         ('admin', 'group.admin', 'none'),
 
         ('none', 'group.none', 'all'),
         ('read', 'group.read', 'all'),
         ('write', 'group.write', 'all'),
         ('admin', 'group.admin', 'all'),
 
         ('none', 'group.none', 'repos'),
         ('read', 'group.read', 'repos'),
         ('write', 'group.write', 'repos'),
         ('admin', 'group.admin', 'repos'),
 
         ('none', 'group.none', 'groups'),
         ('read', 'group.read', 'groups'),
         ('write', 'group.write', 'groups'),
         ('admin', 'group.admin', 'groups'),
     ])
     def test_api_grant_user_permission_to_repo_group(
             self, name, perm, apply_to_children, user_util):
         user = user_util.create_user()
         repo_group = user_util.create_repo_group()
         id_, params = build_data(
             self.apikey, 'grant_user_permission_to_repo_group',
             repogroupid=repo_group.name, userid=user.username,
             perm=perm, apply_to_children=apply_to_children)
         response = api_call(self.app, params)
 
         ret = {
             'msg': (
                 'Granted perm: `%s` (recursive:%s) for user: `%s`'
                 ' in repo group: `%s`' % (
                     perm, apply_to_children, user.username, repo_group.name
                 )
             ),
             'success': True
         }
         expected = ret
         assert_ok(id_, expected, given=response.body)
 
     @pytest.mark.parametrize(
         "name, perm, apply_to_children, grant_admin, access_ok", [
             ('none_fails', 'group.none', 'none', False, False),
             ('read_fails', 'group.read', 'none', False, False),
             ('write_fails', 'group.write', 'none', False, False),
             ('admin_fails', 'group.admin', 'none', False, False),
 
             # with granted perms
             ('none_ok', 'group.none', 'none', True, True),
             ('read_ok', 'group.read', 'none', True, True),
             ('write_ok', 'group.write', 'none', True, True),
             ('admin_ok', 'group.admin', 'none', True, True),
         ]
     )
     def test_api_grant_user_permission_to_repo_group_by_regular_user(
             self, name, perm, apply_to_children, grant_admin, access_ok,
             user_util):
         user = user_util.create_user()
         repo_group = user_util.create_repo_group()
 
         if grant_admin:
             test_user = UserModel().get_by_username(self.TEST_USER_LOGIN)
             user_util.grant_user_permission_to_repo_group(
                 repo_group, test_user, 'group.admin')
 
         id_, params = build_data(
             self.apikey_regular, 'grant_user_permission_to_repo_group',
             repogroupid=repo_group.name, userid=user.username,
             perm=perm, apply_to_children=apply_to_children)
         response = api_call(self.app, params)
         if access_ok:
             ret = {
                 'msg': (
                     'Granted perm: `%s` (recursive:%s) for user: `%s`'
                     ' in repo group: `%s`' % (
                         perm, apply_to_children, user.username, repo_group.name
                     )
                 ),
                 'success': True
             }
             expected = ret
             assert_ok(id_, expected, given=response.body)
         else:
             expected = 'repository group `%s` does not exist' % (
                 repo_group.name, )
             assert_error(id_, expected, given=response.body)
 
     def test_api_grant_user_permission_to_repo_group_wrong_permission(
             self, user_util):
         user = user_util.create_user()
         repo_group = user_util.create_repo_group()
         perm = 'haha.no.permission'
         id_, params = build_data(
             self.apikey,
             'grant_user_permission_to_repo_group',
             repogroupid=repo_group.name,
             userid=user.username,
             perm=perm)
         response = api_call(self.app, params)
 
-        expected = 'permission `%s` does not exist' % (perm,)
+        expected = 'permission `%s` does not exist. Permission should start with prefix: `group.`' % (perm,)
         assert_error(id_, expected, given=response.body)
 
     @mock.patch.object(RepoGroupModel, 'grant_user_permission', crash)
     def test_api_grant_user_permission_to_repo_group_exception_when_adding(
             self, user_util):
         user = user_util.create_user()
         repo_group = user_util.create_repo_group()
         perm = 'group.read'
         id_, params = build_data(
             self.apikey,
             'grant_user_permission_to_repo_group',
             repogroupid=repo_group.name,
             userid=user.username,
             perm=perm)
         response = api_call(self.app, params)
 
         expected = (
             'failed to edit permission for user: `%s` in repo group: `%s`' % (
                 user.username, repo_group.name
             )
         )
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_grant_user_permission_to_user_group.py

@@ -1,156 +1,156 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2019 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import mock
 import pytest
 
 from rhodecode.model.user import UserModel
 from rhodecode.model.user_group import UserGroupModel
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_error, assert_ok, crash)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestGrantUserPermissionFromUserGroup(object):
     @pytest.mark.parametrize("name, perm", [
         ('none', 'usergroup.none'),
         ('read', 'usergroup.read'),
         ('write', 'usergroup.write'),
         ('admin', 'usergroup.admin'),
 
         ('none', 'usergroup.none'),
         ('read', 'usergroup.read'),
         ('write', 'usergroup.write'),
         ('admin', 'usergroup.admin'),
 
         ('none', 'usergroup.none'),
         ('read', 'usergroup.read'),
         ('write', 'usergroup.write'),
         ('admin', 'usergroup.admin'),
 
         ('none', 'usergroup.none'),
         ('read', 'usergroup.read'),
         ('write', 'usergroup.write'),
         ('admin', 'usergroup.admin'),
     ])
     def test_api_grant_user_permission_to_user_group(
             self, name, perm, user_util):
         user = user_util.create_user()
         group = user_util.create_user_group()
         id_, params = build_data(
             self.apikey,
             'grant_user_permission_to_user_group',
             usergroupid=group.users_group_name,
             userid=user.username,
             perm=perm)
         response = api_call(self.app, params)
 
         ret = {
             'msg': 'Granted perm: `%s` for user: `%s` in user group: `%s`' % (
                 perm, user.username, group.users_group_name
             ),
             'success': True
         }
         expected = ret
         assert_ok(id_, expected, given=response.body)
 
     @pytest.mark.parametrize("name, perm, grant_admin, access_ok", [
         ('none_fails', 'usergroup.none', False, False),
         ('read_fails', 'usergroup.read', False, False),
         ('write_fails', 'usergroup.write', False, False),
         ('admin_fails', 'usergroup.admin', False, False),
 
         # with granted perms
         ('none_ok', 'usergroup.none', True, True),
         ('read_ok', 'usergroup.read', True, True),
         ('write_ok', 'usergroup.write', True, True),
         ('admin_ok', 'usergroup.admin', True, True),
     ])
     def test_api_grant_user_permission_to_user_group_by_regular_user(
             self, name, perm, grant_admin, access_ok, user_util):
         api_user = UserModel().get_by_username(self.TEST_USER_LOGIN)
         user = user_util.create_user()
         group = user_util.create_user_group()
         # grant the user ability to at least read the group
         permission = 'usergroup.admin' if grant_admin else 'usergroup.read'
         user_util.grant_user_permission_to_user_group(
             group, api_user, permission)
 
         id_, params = build_data(
             self.apikey_regular,
             'grant_user_permission_to_user_group',
             usergroupid=group.users_group_name,
             userid=user.username,
             perm=perm)
         response = api_call(self.app, params)
 
         if access_ok:
             ret = {
                 'msg': (
                     'Granted perm: `%s` for user: `%s` in user group: `%s`' % (
                         perm, user.username, group.users_group_name
                     )
                 ),
                 'success': True
             }
             expected = ret
             assert_ok(id_, expected, given=response.body)
         else:
             expected = 'user group `%s` does not exist' % (
                 group.users_group_name)
             assert_error(id_, expected, given=response.body)
 
     def test_api_grant_user_permission_to_user_group_wrong_permission(
             self, user_util):
         user = user_util.create_user()
         group = user_util.create_user_group()
         perm = 'haha.no.permission'
         id_, params = build_data(
             self.apikey,
             'grant_user_permission_to_user_group',
             usergroupid=group.users_group_name,
             userid=user.username,
             perm=perm)
         response = api_call(self.app, params)
 
-        expected = 'permission `%s` does not exist' % perm
+        expected = 'permission `%s` does not exist. Permission should start with prefix: `usergroup.`' % perm
         assert_error(id_, expected, given=response.body)
 
     def test_api_grant_user_permission_to_user_group_exception_when_adding(
             self, user_util):
         user = user_util.create_user()
         group = user_util.create_user_group()
 
         perm = 'usergroup.read'
         id_, params = build_data(
             self.apikey,
             'grant_user_permission_to_user_group',
             usergroupid=group.users_group_name,
             userid=user.username,
             perm=perm)
         with mock.patch.object(UserGroupModel, 'grant_user_permission', crash):
             response = api_call(self.app, params)
 
         expected = (
             'failed to edit permission for user: `%s` in user group: `%s`' % (
                 user.username, group.users_group_name
             )
         )
         assert_error(id_, expected, given=response.body)

rhodecode/api/utils.py

@@ -1,449 +1,453 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2014-2019 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 """
 JSON RPC utils
 """
 
 import collections
 import logging
 
 from rhodecode.api.exc import JSONRPCError
 from rhodecode.lib.auth import (
     HasPermissionAnyApi, HasRepoPermissionAnyApi, HasRepoGroupPermissionAnyApi)
 from rhodecode.lib.utils import safe_unicode
 from rhodecode.lib.vcs.exceptions import RepositoryError
 from rhodecode.lib.view_utils import get_commit_from_ref_name
 from rhodecode.lib.utils2 import str2bool
 
 log = logging.getLogger(__name__)
 
 
 class OAttr(object):
     """
     Special Option that defines other attribute, and can default to them
 
     Example::
 
         def test(apiuser, userid=Optional(OAttr('apiuser')):
             user = Optional.extract(userid, evaluate_locals=local())
             #if we pass in userid, we get it, else it will default to apiuser
             #attribute
     """
 
     def __init__(self, attr_name):
         self.attr_name = attr_name
 
     def __repr__(self):
         return '<OptionalAttr:%s>' % self.attr_name
 
     def __call__(self):
         return self
 
 
 class Optional(object):
     """
     Defines an optional parameter::
 
         param = param.getval() if isinstance(param, Optional) else param
         param = param() if isinstance(param, Optional) else param
 
     is equivalent of::
 
         param = Optional.extract(param)
 
     """
 
     def __init__(self, type_):
         self.type_ = type_
 
     def __repr__(self):
         return '<Optional:%s>' % self.type_.__repr__()
 
     def __call__(self):
         return self.getval()
 
     def getval(self, evaluate_locals=None):
         """
         returns value from this Optional instance
         """
         if isinstance(self.type_, OAttr):
             param_name = self.type_.attr_name
             if evaluate_locals:
                 return evaluate_locals[param_name]
             # use params name
             return param_name
         return self.type_
 
     @classmethod
     def extract(cls, val, evaluate_locals=None, binary=None):
         """
         Extracts value from Optional() instance
 
         :param val:
         :return: original value if it's not Optional instance else
             value of instance
         """
         if isinstance(val, cls):
             val = val.getval(evaluate_locals)
 
         if binary:
             val = str2bool(val)
 
         return val
 
 
 def parse_args(cli_args, key_prefix=''):
     from rhodecode.lib.utils2 import (escape_split)
     kwargs = collections.defaultdict(dict)
     for el in escape_split(cli_args, ','):
         kv = escape_split(el, '=', 1)
         if len(kv) == 2:
             k, v = kv
             kwargs[key_prefix + k] = v
     return kwargs
 
 
 def get_origin(obj):
     """
     Get origin of permission from object.
 
     :param obj:
     """
     origin = 'permission'
 
     if getattr(obj, 'owner_row', '') and getattr(obj, 'admin_row', ''):
         # admin and owner case, maybe we should use dual string ?
         origin = 'owner'
     elif getattr(obj, 'owner_row', ''):
         origin = 'owner'
     elif getattr(obj, 'admin_row', ''):
         origin = 'super-admin'
     return origin
 
 
 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 has_superadmin_permission(apiuser):
     """
     Return True if apiuser is admin or return False
 
     :param apiuser:
     """
     if HasPermissionAnyApi('hg.admin')(user=apiuser):
         return True
     return False
 
 
 def validate_repo_permissions(apiuser, repoid, repo, perms):
     """
     Raise JsonRPCError if apiuser is not authorized or return True
 
     :param apiuser:
     :param repoid:
     :param repo:
     :param perms:
     """
     if not HasRepoPermissionAnyApi(*perms)(
             user=apiuser, repo_name=repo.repo_name):
         raise JSONRPCError(
             'repository `%s` does not exist' % repoid)
 
     return True
 
 
 def validate_repo_group_permissions(apiuser, repogroupid, repo_group, perms):
     """
     Raise JsonRPCError if apiuser is not authorized or return True
 
     :param apiuser:
     :param repogroupid: just the id of repository group
     :param repo_group: instance of repo_group
     :param perms:
     """
     if not HasRepoGroupPermissionAnyApi(*perms)(
             user=apiuser, group_name=repo_group.group_name):
         raise JSONRPCError(
             'repository group `%s` does not exist' % repogroupid)
 
     return True
 
 
 def validate_set_owner_permissions(apiuser, owner):
     if isinstance(owner, Optional):
         owner = get_user_or_error(apiuser.user_id)
     else:
         if has_superadmin_permission(apiuser):
             owner = get_user_or_error(owner)
         else:
             # forbid setting owner for non-admins
             raise JSONRPCError(
                 'Only RhodeCode super-admin can specify `owner` param')
     return owner
 
 
 def get_user_or_error(userid):
     """
     Get user by id or name or return JsonRPCError if not found
 
     :param userid:
     """
     from rhodecode.model.user import UserModel
     user_model = UserModel()
 
     if isinstance(userid, (int, long)):
         try:
             user = user_model.get_user(userid)
         except ValueError:
             user = None
     else:
         user = user_model.get_by_username(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:
     """
     from rhodecode.model.repo import RepoModel
     repo_model = RepoModel()
 
     if isinstance(repoid, (int, long)):
         try:
             repo = repo_model.get_repo(repoid)
         except ValueError:
             repo = None
     else:
         repo = repo_model.get_by_repo_name(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:
     """
     from rhodecode.model.repo_group import RepoGroupModel
     repo_group_model = RepoGroupModel()
 
     if isinstance(repogroupid, (int, long)):
         try:
             repo_group = repo_group_model._get_repo_group(repogroupid)
         except ValueError:
             repo_group = None
     else:
         repo_group = repo_group_model.get_by_group_name(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:
     """
     from rhodecode.model.user_group import UserGroupModel
     user_group_model = UserGroupModel()
 
     if isinstance(usergroupid, (int, long)):
         try:
             user_group = user_group_model.get_group(usergroupid)
         except ValueError:
             user_group = None
     else:
         user_group = user_group_model.get_by_name(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:
     """
     from rhodecode.model.permission import PermissionModel
 
     perm = PermissionModel.cls.get_by_key(permid)
     if perm is None:
-        raise JSONRPCError('permission `%s` does not exist' % (permid,))
+        msg = 'permission `{}` does not exist.'.format(permid)
+        if prefix:
+            msg += ' Permission should start with prefix: `{}`'.format(prefix)
+        raise JSONRPCError(msg)
+
     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:
     """
     from rhodecode.model.gist import GistModel
 
     gist = GistModel.cls.get_by_access_id(gistid)
     if gist is None:
         raise JSONRPCError('gist `%s` does not exist' % (gistid,))
     return gist
 
 
 def get_pull_request_or_error(pullrequestid):
     """
     Get pull request by id or return JsonRPCError if not found
 
     :param pullrequestid:
     """
     from rhodecode.model.pull_request import PullRequestModel
 
     try:
         pull_request = PullRequestModel().get(int(pullrequestid))
     except ValueError:
         raise JSONRPCError('pullrequestid must be an integer')
     if not pull_request:
         raise JSONRPCError('pull request `%s` does not exist' % (
             pullrequestid,))
     return pull_request
 
 
 def build_commit_data(commit, detail_level):
     parsed_diff = []
     if detail_level == 'extended':
         for f in commit.added:
             parsed_diff.append(_get_commit_dict(filename=f.path, op='A'))
         for f in commit.changed:
             parsed_diff.append(_get_commit_dict(filename=f.path, op='M'))
         for f in commit.removed:
             parsed_diff.append(_get_commit_dict(filename=f.path, op='D'))
 
     elif detail_level == 'full':
         from rhodecode.lib.diffs import DiffProcessor
         diff_processor = DiffProcessor(commit.diff())
         for dp in diff_processor.prepare():
             del dp['stats']['ops']
             _stats = dp['stats']
             parsed_diff.append(_get_commit_dict(
                 filename=dp['filename'], op=dp['operation'],
                 new_revision=dp['new_revision'],
                 old_revision=dp['old_revision'],
                 raw_diff=dp['raw_diff'], stats=_stats))
 
     return parsed_diff
 
 
 def get_commit_or_error(ref, repo):
     try:
         ref_type, _, ref_hash = ref.split(':')
     except ValueError:
         raise JSONRPCError(
             'Ref `{ref}` given in a wrong format. Please check the API'
             ' documentation for more details'.format(ref=ref))
     try:
         # TODO: dan: refactor this to use repo.scm_instance().get_commit()
         # once get_commit supports ref_types
         return get_commit_from_ref_name(repo, ref_hash)
     except RepositoryError:
         raise JSONRPCError('Ref `{ref}` does not exist'.format(ref=ref))
 
 
 def _get_ref_hash(repo, type_, name):
     vcs_repo = repo.scm_instance()
     if type_ in ['branch'] and vcs_repo.alias in ('hg', 'git'):
         return vcs_repo.branches[name]
     elif type_ in ['bookmark', 'book'] and vcs_repo.alias == 'hg':
         return vcs_repo.bookmarks[name]
     else:
         raise ValueError()
 
 
 def resolve_ref_or_error(ref, repo, allowed_ref_types=None):
     allowed_ref_types = allowed_ref_types or ['bookmark', 'book', 'tag', 'branch']
 
     def _parse_ref(type_, name, hash_=None):
         return type_, name, hash_
 
     try:
         ref_type, ref_name, ref_hash = _parse_ref(*ref.split(':'))
     except TypeError:
         raise JSONRPCError(
             'Ref `{ref}` given in a wrong format. Please check the API'
             ' documentation for more details'.format(ref=ref))
 
     if ref_type not in allowed_ref_types:
         raise JSONRPCError(
             'Ref `{ref}` type is not allowed. '
             'Only:{allowed_refs} are possible.'.format(
                 ref=ref, allowed_refs=allowed_ref_types))
 
     try:
         ref_hash = ref_hash or _get_ref_hash(repo, ref_type, ref_name)
     except (KeyError, ValueError):
         raise JSONRPCError(
             'The specified value:{type}:`{name}` does not exist, or is not allowed.'.format(
                 type=ref_type, name=ref_name))
 
     return ':'.join([ref_type, ref_name, ref_hash])
 
 
 def _get_commit_dict(
         filename, op, new_revision=None, old_revision=None,
         raw_diff=None, stats=None):
     if stats is None:
         stats = {
             "added": None,
             "binary": None,
             "deleted": None
         }
     return {
         "filename": safe_unicode(filename),
         "op": op,
 
         # extra details
         "new_revision": new_revision,
         "old_revision": old_revision,
 
         "raw_diff": raw_diff,
         "stats": stats
     }