api: make all tests for API pass

rhodecode/api/tests/test_comment_pull_request.py

@@ -1,390 +1,380 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 
 from rhodecode.model.comment import CommentsModel
 from rhodecode.model.db import UserLog, User, ChangesetComment
 from rhodecode.model.pull_request import PullRequestModel
 from rhodecode.tests import TEST_USER_ADMIN_LOGIN
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_error, assert_ok)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestCommentPullRequest(object):
     finalizers = []
 
     def teardown_method(self, method):
         if self.finalizers:
             for finalizer in self.finalizers:
                 finalizer()
             self.finalizers = []
 
     @pytest.mark.backends("git", "hg")
     def test_api_comment_pull_request(self, pr_util, no_notifications):
         pull_request = pr_util.create_pull_request()
         pull_request_id = pull_request.pull_request_id
         author = pull_request.user_id
         repo = pull_request.target_repo.repo_id
         id_, params = build_data(
             self.apikey, 'comment_pull_request',
             repoid=pull_request.target_repo.repo_name,
             pullrequestid=pull_request.pull_request_id,
             message='test message')
         response = api_call(self.app, params)
         pull_request = PullRequestModel().get(pull_request.pull_request_id)
 
         comments = CommentsModel().get_comments(
             pull_request.target_repo.repo_id, pull_request=pull_request)
 
         expected = {
             'pull_request_id': pull_request.pull_request_id,
             'comment_id': comments[-1].comment_id,
             'status': {'given': None, 'was_changed': None}
         }
         assert_ok(id_, expected, response.body)
 
         journal = UserLog.query()\
             .filter(UserLog.user_id == author)\
             .filter(UserLog.repository_id == repo) \
             .order_by(UserLog.user_log_id.asc()) \
             .all()
         assert journal[-1].action == 'repo.pull_request.comment.create'
 
     @pytest.mark.backends("git", "hg")
     def test_api_comment_pull_request_with_extra_recipients(self, pr_util, user_util):
         pull_request = pr_util.create_pull_request()
 
         user1 = user_util.create_user()
         user1_id = user1.user_id
         user2 = user_util.create_user()
         user2_id = user2.user_id
 
         id_, params = build_data(
             self.apikey, 'comment_pull_request',
             repoid=pull_request.target_repo.repo_name,
             pullrequestid=pull_request.pull_request_id,
             message='test message',
             extra_recipients=[user1.user_id, user2.username]
         )
         response = api_call(self.app, params)
         pull_request = PullRequestModel().get(pull_request.pull_request_id)
 
         comments = CommentsModel().get_comments(
             pull_request.target_repo.repo_id, pull_request=pull_request)
 
         expected = {
             'pull_request_id': pull_request.pull_request_id,
             'comment_id': comments[-1].comment_id,
             'status': {'given': None, 'was_changed': None}
         }
         assert_ok(id_, expected, response.body)
         # check user1/user2 inbox for notification
         user1 = User.get(user1_id)
         assert 1 == len(user1.notifications)
         assert 'test message' in user1.notifications[0].notification.body
 
         user2 = User.get(user2_id)
         assert 1 == len(user2.notifications)
         assert 'test message' in user2.notifications[0].notification.body
 
     @pytest.mark.backends("git", "hg")
     def test_api_comment_pull_request_change_status(
             self, pr_util, no_notifications):
         pull_request = pr_util.create_pull_request()
         pull_request_id = pull_request.pull_request_id
         id_, params = build_data(
             self.apikey, 'comment_pull_request',
             repoid=pull_request.target_repo.repo_name,
             pullrequestid=pull_request.pull_request_id,
             status='rejected')
         response = api_call(self.app, params)
         pull_request = PullRequestModel().get(pull_request_id)
 
         comments = CommentsModel().get_comments(
             pull_request.target_repo.repo_id, pull_request=pull_request)
         expected = {
             'pull_request_id': pull_request.pull_request_id,
             'comment_id': comments[-1].comment_id,
             'status':  {'given': 'rejected', 'was_changed': True}
         }
         assert_ok(id_, expected, response.body)
 
     @pytest.mark.backends("git", "hg")
-    def test_api_comment_pull_request_change_status_with_specific_commit_id(
+    def test_api_comment_pull_request_change_status_with_specific_commit_id_and_test_commit(
             self, pr_util, no_notifications):
         pull_request = pr_util.create_pull_request()
         pull_request_id = pull_request.pull_request_id
         latest_commit_id = 'test_commit'
+
         # inject additional revision, to fail test the status change on
         # non-latest commit
         pull_request.revisions = pull_request.revisions + ['test_commit']
 
         id_, params = build_data(
             self.apikey, 'comment_pull_request',
+            message='test-change-of-status-not-allowed',
             repoid=pull_request.target_repo.repo_name,
             pullrequestid=pull_request.pull_request_id,
             status='approved', commit_id=latest_commit_id)
         response = api_call(self.app, params)
         pull_request = PullRequestModel().get(pull_request_id)
+        comments = CommentsModel().get_comments(
+            pull_request.target_repo.repo_id, pull_request=pull_request)
 
         expected = {
             'pull_request_id': pull_request.pull_request_id,
-            'comment_id': None,
+            'comment_id': comments[-1].comment_id,
             'status':  {'given': 'approved', 'was_changed': False}
         }
         assert_ok(id_, expected, response.body)
 
     @pytest.mark.backends("git", "hg")
     def test_api_comment_pull_request_change_status_with_specific_commit_id(
             self, pr_util, no_notifications):
         pull_request = pr_util.create_pull_request()
         pull_request_id = pull_request.pull_request_id
         latest_commit_id = pull_request.revisions[0]
 
         id_, params = build_data(
             self.apikey, 'comment_pull_request',
             repoid=pull_request.target_repo.repo_name,
             pullrequestid=pull_request.pull_request_id,
             status='approved', commit_id=latest_commit_id)
         response = api_call(self.app, params)
         pull_request = PullRequestModel().get(pull_request_id)
 
         comments = CommentsModel().get_comments(
             pull_request.target_repo.repo_id, pull_request=pull_request)
         expected = {
             'pull_request_id': pull_request.pull_request_id,
             'comment_id': comments[-1].comment_id,
             'status':  {'given': 'approved', 'was_changed': True}
         }
         assert_ok(id_, expected, response.body)
 
     @pytest.mark.backends("git", "hg")
     def test_api_comment_pull_request_missing_params_error(self, pr_util):
         pull_request = pr_util.create_pull_request()
         pull_request_id = pull_request.pull_request_id
         pull_request_repo = pull_request.target_repo.repo_name
         id_, params = build_data(
             self.apikey, 'comment_pull_request',
             repoid=pull_request_repo,
             pullrequestid=pull_request_id)
         response = api_call(self.app, params)
 
         expected = 'Both message and status parameters are missing. At least one is required.'
         assert_error(id_, expected, given=response.body)
 
     @pytest.mark.backends("git", "hg")
     def test_api_comment_pull_request_unknown_status_error(self, pr_util):
         pull_request = pr_util.create_pull_request()
         pull_request_id = pull_request.pull_request_id
         pull_request_repo = pull_request.target_repo.repo_name
         id_, params = build_data(
             self.apikey, 'comment_pull_request',
             repoid=pull_request_repo,
             pullrequestid=pull_request_id,
             status='42')
         response = api_call(self.app, params)
 
         expected = 'Unknown comment status: `42`'
         assert_error(id_, expected, given=response.body)
 
     @pytest.mark.backends("git", "hg")
     def test_api_comment_pull_request_repo_error(self, pr_util):
         pull_request = pr_util.create_pull_request()
         id_, params = build_data(
             self.apikey, 'comment_pull_request',
             repoid=666, pullrequestid=pull_request.pull_request_id)
         response = api_call(self.app, params)
 
         expected = 'repository `666` does not exist'
         assert_error(id_, expected, given=response.body)
 
     @pytest.mark.backends("git", "hg")
     def test_api_comment_pull_request_non_admin_with_userid_error(self, pr_util):
         pull_request = pr_util.create_pull_request()
         id_, params = build_data(
             self.apikey_regular, 'comment_pull_request',
             repoid=pull_request.target_repo.repo_name,
             pullrequestid=pull_request.pull_request_id,
             userid=TEST_USER_ADMIN_LOGIN)
         response = api_call(self.app, params)
 
         expected = 'userid is not the same as your user'
         assert_error(id_, expected, given=response.body)
 
     @pytest.mark.backends("git", "hg")
-    def test_api_comment_pull_request_non_admin_with_userid_error(self, pr_util):
-        pull_request = pr_util.create_pull_request()
-        id_, params = build_data(
-            self.apikey_regular, 'comment_pull_request',
-            repoid=pull_request.target_repo.repo_name,
-            pullrequestid=pull_request.pull_request_id,
-            userid=TEST_USER_ADMIN_LOGIN)
-        response = api_call(self.app, params)
-
-        expected = 'userid is not the same as your user'
-        assert_error(id_, expected, given=response.body)
-
-    @pytest.mark.backends("git", "hg")
     def test_api_comment_pull_request_wrong_commit_id_error(self, pr_util):
         pull_request = pr_util.create_pull_request()
         id_, params = build_data(
             self.apikey_regular, 'comment_pull_request',
             repoid=pull_request.target_repo.repo_name,
             status='approved',
             pullrequestid=pull_request.pull_request_id,
             commit_id='XXX')
         response = api_call(self.app, params)
 
         expected = 'Invalid commit_id `XXX` for this pull request.'
         assert_error(id_, expected, given=response.body)
 
     @pytest.mark.backends("git", "hg")
     def test_api_edit_comment(self, pr_util):
         pull_request = pr_util.create_pull_request()
 
         id_, params = build_data(
             self.apikey,
             'comment_pull_request',
             repoid=pull_request.target_repo.repo_name,
             pullrequestid=pull_request.pull_request_id,
             message='test message',
         )
         response = api_call(self.app, params)
         json_response = response.json
         comment_id = json_response['result']['comment_id']
 
         message_after_edit = 'just message'
         id_, params = build_data(
             self.apikey,
             'edit_comment',
             comment_id=comment_id,
             message=message_after_edit,
             version=0,
         )
         response = api_call(self.app, params)
         json_response = response.json
         assert json_response['result']['version'] == 1
 
         text_form_db = ChangesetComment.get(comment_id).text
         assert message_after_edit == text_form_db
 
     @pytest.mark.backends("git", "hg")
-    def test_api_edit_comment_wrong_version(self, pr_util):
+    def test_api_edit_comment_wrong_version_mismatch(self, pr_util):
         pull_request = pr_util.create_pull_request()
 
         id_, params = build_data(
             self.apikey, 'comment_pull_request',
             repoid=pull_request.target_repo.repo_name,
             pullrequestid=pull_request.pull_request_id,
             message='test message')
         response = api_call(self.app, params)
         json_response = response.json
         comment_id = json_response['result']['comment_id']
 
         message_after_edit = 'just message'
         id_, params = build_data(
-            self.apikey_regular,
+            self.apikey,
             'edit_comment',
             comment_id=comment_id,
             message=message_after_edit,
             version=1,
         )
         response = api_call(self.app, params)
         expected = 'comment ({}) version ({}) mismatch'.format(comment_id, 1)
         assert_error(id_, expected, given=response.body)
 
     @pytest.mark.backends("git", "hg")
     def test_api_edit_comment_wrong_version(self, pr_util):
         pull_request = pr_util.create_pull_request()
 
         id_, params = build_data(
             self.apikey, 'comment_pull_request',
             repoid=pull_request.target_repo.repo_name,
             pullrequestid=pull_request.pull_request_id,
             message='test message')
         response = api_call(self.app, params)
         json_response = response.json
         comment_id = json_response['result']['comment_id']
 
         id_, params = build_data(
             self.apikey,
             'edit_comment',
             comment_id=comment_id,
             message='',
             version=0,
         )
         response = api_call(self.app, params)
-        expected = "comment ({}) can't be changed with empty string".format(comment_id, 1)
+        expected = f"comment ({comment_id}) can't be changed with empty string"
         assert_error(id_, expected, given=response.body)
 
     @pytest.mark.backends("git", "hg")
     def test_api_edit_comment_wrong_user_set_by_non_admin(self, pr_util):
         pull_request = pr_util.create_pull_request()
         pull_request_id = pull_request.pull_request_id
         id_, params = build_data(
             self.apikey,
             'comment_pull_request',
             repoid=pull_request.target_repo.repo_name,
             pullrequestid=pull_request_id,
             message='test message'
         )
         response = api_call(self.app, params)
         json_response = response.json
         comment_id = json_response['result']['comment_id']
 
         id_, params = build_data(
             self.apikey_regular,
             'edit_comment',
             comment_id=comment_id,
             message='just message',
             version=0,
             userid=TEST_USER_ADMIN_LOGIN
         )
         response = api_call(self.app, params)
         expected = 'userid is not the same as your user'
         assert_error(id_, expected, given=response.body)
 
     @pytest.mark.backends("git", "hg")
     def test_api_edit_comment_wrong_user_with_permissions_to_edit_comment(self, pr_util):
         pull_request = pr_util.create_pull_request()
         pull_request_id = pull_request.pull_request_id
         id_, params = build_data(
             self.apikey,
             'comment_pull_request',
             repoid=pull_request.target_repo.repo_name,
             pullrequestid=pull_request_id,
             message='test message'
         )
         response = api_call(self.app, params)
         json_response = response.json
         comment_id = json_response['result']['comment_id']
 
         id_, params = build_data(
             self.apikey_regular,
             'edit_comment',
             comment_id=comment_id,
             message='just message',
             version=0,
         )
         response = api_call(self.app, params)
         expected = "you don't have access to edit this comment"
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_create_gist.py

@@ -1,102 +1,101 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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.db import Gist
 from rhodecode.model.gist import GistModel
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_error, assert_ok, crash)
 from rhodecode.tests.fixture import Fixture
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestApiCreateGist(object):
     @pytest.mark.parametrize("lifetime, gist_type, gist_acl_level", [
         (10, Gist.GIST_PUBLIC, Gist.ACL_LEVEL_PUBLIC),
         (20, Gist.GIST_PUBLIC, Gist.ACL_LEVEL_PRIVATE),
         (40, Gist.GIST_PRIVATE, Gist.ACL_LEVEL_PUBLIC),
         (80, Gist.GIST_PRIVATE, Gist.ACL_LEVEL_PRIVATE),
     ])
     def test_api_create_gist(self, lifetime, gist_type, gist_acl_level):
         id_, params = build_data(
             self.apikey_regular, 'create_gist',
             lifetime=lifetime,
             description='foobar-gist',
             gist_type=gist_type,
             acl_level=gist_acl_level,
             files={'foobar_ąć': {'content': 'foo'}})
         response = api_call(self.app, params)
         response_json = response.json
         gist = response_json['result']['gist']
         expected = {
             'gist': {
                 'access_id': gist['access_id'],
                 'created_on': gist['created_on'],
                 'modified_at': gist['modified_at'],
                 'description': 'foobar-gist',
                 'expires': gist['expires'],
                 'gist_id': gist['gist_id'],
                 'type': gist_type,
                 'url': gist['url'],
                 # content is empty since we don't show it here
                 'content': None,
                 'acl_level': gist_acl_level,
             },
             'msg': 'created new gist'
         }
         try:
             assert_ok(id_, expected, given=response.body)
         finally:
             Fixture().destroy_gists()
 
     @pytest.mark.parametrize("expected, lifetime, gist_type, gist_acl_level, files", [
         ({'gist_type': '"ups" is not one of private, public'},
          10, 'ups', Gist.ACL_LEVEL_PUBLIC, {'f': {'content': 'f'}}),
 
         ({'lifetime': '-120 is less than minimum value -1'},
          -120, Gist.GIST_PUBLIC, Gist.ACL_LEVEL_PUBLIC, {'f': {'content': 'f'}}),
 
         ({'0.content': 'Required'},
          10, Gist.GIST_PUBLIC, Gist.ACL_LEVEL_PUBLIC, {'f': {'x': 'f'}}),
     ])
     def test_api_try_create_gist(
             self, expected, lifetime, gist_type, gist_acl_level, files):
         id_, params = build_data(
             self.apikey_regular, 'create_gist',
             lifetime=lifetime,
             description='foobar-gist',
             gist_type=gist_type,
             acl_level=gist_acl_level,
             files=files)
         response = api_call(self.app, params)
 
         try:
             assert_error(id_, expected, given=response.body)
         finally:
             Fixture().destroy_gists()
 
     @mock.patch.object(GistModel, 'create', crash)
     def test_api_create_gist_exception_occurred(self):
         id_, params = build_data(self.apikey_regular, 'create_gist', files={})
         response = api_call(self.app, params)
         expected = 'failed to create gist'
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_create_pull_request.py

@@ -1,368 +1,367 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 
 from rhodecode.model.db import User
 from rhodecode.model.pull_request import PullRequestModel
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.user import UserModel
 from rhodecode.tests import TEST_USER_ADMIN_LOGIN, TEST_USER_REGULAR_LOGIN
 from rhodecode.api.tests.utils import build_data, api_call, assert_error
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestCreatePullRequestApi(object):
     finalizers = []
 
     def teardown_method(self, method):
         if self.finalizers:
             for finalizer in self.finalizers:
                 finalizer()
             self.finalizers = []
 
     def test_create_with_wrong_data(self):
         required_data = {
             'source_repo': 'tests/source_repo',
             'target_repo': 'tests/target_repo',
             'source_ref': 'branch:default:initial',
             'target_ref': 'branch:default:new-feature',
         }
         for key in required_data:
             data = required_data.copy()
             data.pop(key)
             id_, params = build_data(
                 self.apikey, 'create_pull_request', **data)
             response = api_call(self.app, params)
 
             expected = 'Missing non optional `{}` arg in JSON DATA'.format(key)
             assert_error(id_, expected, given=response.body)
 
     @pytest.mark.backends("git", "hg")
     @pytest.mark.parametrize('source_ref', [
         'bookmarg:default:initial'
     ])
     def test_create_with_wrong_refs_data(self, backend, source_ref):
 
         data = self._prepare_data(backend)
         data['source_ref'] = source_ref
 
         id_, params = build_data(
             self.apikey_regular, 'create_pull_request', **data)
 
         response = api_call(self.app, params)
 
         expected = "Ref `{}` type is not allowed. " \
                    "Only:['bookmark', 'book', 'tag', 'branch'] " \
                    "are possible.".format(source_ref)
         assert_error(id_, expected, given=response.body)
 
     @pytest.mark.backends("git", "hg")
     def test_create_with_correct_data(self, backend):
         data = self._prepare_data(backend)
         RepoModel().revoke_user_permission(
             self.source.repo_name, User.DEFAULT_USER)
         id_, params = build_data(
             self.apikey_regular, 'create_pull_request', **data)
         response = api_call(self.app, params)
         expected_message = "Created new pull request `{title}`".format(
             title=data['title'])
         result = response.json
         assert result['error'] is None
         assert result['result']['msg'] == expected_message
         pull_request_id = result['result']['pull_request_id']
         pull_request = PullRequestModel().get(pull_request_id)
         assert pull_request.title == data['title']
         assert pull_request.description == data['description']
         assert pull_request.source_ref == data['source_ref']
         assert pull_request.target_ref == data['target_ref']
         assert pull_request.source_repo.repo_name == data['source_repo']
         assert pull_request.target_repo.repo_name == data['target_repo']
         assert pull_request.revisions == [self.commit_ids['change']]
         assert len(pull_request.reviewers) == 1
 
     @pytest.mark.backends("git", "hg")
     def test_create_with_empty_description(self, backend):
         data = self._prepare_data(backend)
         data.pop('description')
         id_, params = build_data(
             self.apikey_regular, 'create_pull_request', **data)
         response = api_call(self.app, params)
         expected_message = "Created new pull request `{title}`".format(
             title=data['title'])
         result = response.json
         assert result['error'] is None
         assert result['result']['msg'] == expected_message
         pull_request_id = result['result']['pull_request_id']
         pull_request = PullRequestModel().get(pull_request_id)
         assert pull_request.description == ''
 
     @pytest.mark.backends("git", "hg")
     def test_create_with_empty_title(self, backend):
         data = self._prepare_data(backend)
         data.pop('title')
         id_, params = build_data(
             self.apikey_regular, 'create_pull_request', **data)
         response = api_call(self.app, params)
         result = response.json
         pull_request_id = result['result']['pull_request_id']
         pull_request = PullRequestModel().get(pull_request_id)
         data['ref'] = backend.default_branch_name
         title = '{source_repo}#{ref} to {target_repo}'.format(**data)
         assert pull_request.title == title
 
     @pytest.mark.backends("git", "hg")
     def test_create_with_reviewers_specified_by_names(
             self, backend, no_notifications):
         data = self._prepare_data(backend)
         reviewers = [
             {'username': TEST_USER_REGULAR_LOGIN,
              'reasons': ['{} added manually'.format(TEST_USER_REGULAR_LOGIN)]},
             {'username': TEST_USER_ADMIN_LOGIN,
              'reasons': ['{} added manually'.format(TEST_USER_ADMIN_LOGIN)],
              'mandatory': True},
         ]
         data['reviewers'] = reviewers
 
         id_, params = build_data(
             self.apikey_regular, 'create_pull_request', **data)
         response = api_call(self.app, params)
 
         expected_message = "Created new pull request `{title}`".format(
             title=data['title'])
         result = response.json
         assert result['error'] is None
         assert result['result']['msg'] == expected_message
         pull_request_id = result['result']['pull_request_id']
         pull_request = PullRequestModel().get(pull_request_id)
 
         actual_reviewers = []
         for rev in pull_request.reviewers:
             entry = {
                 'username': rev.user.username,
                 'reasons': rev.reasons,
              }
             if rev.mandatory:
                 entry['mandatory'] = rev.mandatory
             actual_reviewers.append(entry)
 
         owner_username = pull_request.target_repo.user.username
         for spec_reviewer in reviewers[::]:
             # default reviewer will be added who is an owner of the repo
             # this get's overridden by a add owner to reviewers rule
             if spec_reviewer['username'] == owner_username:
                 spec_reviewer['reasons'] = [u'Default reviewer', u'Repository owner']
                 # since owner is more important, we don't inherit mandatory flag
                 del spec_reviewer['mandatory']
 
         assert sorted(actual_reviewers, key=lambda e: e['username']) \
                == sorted(reviewers, key=lambda e: e['username'])
 
     @pytest.mark.backends("git", "hg")
     def test_create_with_reviewers_specified_by_ids(
             self, backend, no_notifications):
         data = self._prepare_data(backend)
         reviewers = [
             {'username': UserModel().get_by_username(
                 TEST_USER_REGULAR_LOGIN).user_id,
              'reasons': ['added manually']},
             {'username': UserModel().get_by_username(
                 TEST_USER_ADMIN_LOGIN).user_id,
              'reasons': ['added manually']},
         ]
 
         data['reviewers'] = reviewers
         id_, params = build_data(
             self.apikey_regular, 'create_pull_request', **data)
         response = api_call(self.app, params)
 
         expected_message = "Created new pull request `{title}`".format(
             title=data['title'])
         result = response.json
         assert result['error'] is None
         assert result['result']['msg'] == expected_message
         pull_request_id = result['result']['pull_request_id']
         pull_request = PullRequestModel().get(pull_request_id)
 
         actual_reviewers = []
         for rev in pull_request.reviewers:
             entry = {
                 'username': rev.user.user_id,
                 'reasons': rev.reasons,
              }
             if rev.mandatory:
                 entry['mandatory'] = rev.mandatory
             actual_reviewers.append(entry)
 
         owner_user_id = pull_request.target_repo.user.user_id
         for spec_reviewer in reviewers[::]:
             # default reviewer will be added who is an owner of the repo
             # this get's overridden by a add owner to reviewers rule
             if spec_reviewer['username'] == owner_user_id:
                 spec_reviewer['reasons'] = [u'Default reviewer', u'Repository owner']
 
         assert sorted(actual_reviewers, key=lambda e: e['username']) \
                == sorted(reviewers, key=lambda e: e['username'])
 
     @pytest.mark.backends("git", "hg")
     def test_create_fails_when_the_reviewer_is_not_found(self, backend):
         data = self._prepare_data(backend)
         data['reviewers'] = [{'username': 'somebody'}]
         id_, params = build_data(
             self.apikey_regular, 'create_pull_request', **data)
         response = api_call(self.app, params)
         expected_message = 'user `somebody` does not exist'
         assert_error(id_, expected_message, given=response.body)
 
     @pytest.mark.backends("git", "hg")
     def test_cannot_create_with_reviewers_in_wrong_format(self, backend):
         data = self._prepare_data(backend)
         reviewers = ','.join([TEST_USER_REGULAR_LOGIN, TEST_USER_ADMIN_LOGIN])
         data['reviewers'] = reviewers
         id_, params = build_data(
             self.apikey_regular, 'create_pull_request', **data)
         response = api_call(self.app, params)
         expected_message = {u'': '"test_regular,test_admin" is not iterable'}
         assert_error(id_, expected_message, given=response.body)
 
     @pytest.mark.backends("git", "hg")
     def test_create_with_no_commit_hashes(self, backend):
         data = self._prepare_data(backend)
         expected_source_ref = data['source_ref']
         expected_target_ref = data['target_ref']
         data['source_ref'] = 'branch:{}'.format(backend.default_branch_name)
         data['target_ref'] = 'branch:{}'.format(backend.default_branch_name)
         id_, params = build_data(
             self.apikey_regular, 'create_pull_request', **data)
         response = api_call(self.app, params)
         expected_message = "Created new pull request `{title}`".format(
             title=data['title'])
         result = response.json
         assert result['result']['msg'] == expected_message
         pull_request_id = result['result']['pull_request_id']
         pull_request = PullRequestModel().get(pull_request_id)
         assert pull_request.source_ref == expected_source_ref
         assert pull_request.target_ref == expected_target_ref
 
     @pytest.mark.backends("git", "hg")
     @pytest.mark.parametrize("data_key", ["source_repo", "target_repo"])
     def test_create_fails_with_wrong_repo(self, backend, data_key):
         repo_name = 'fake-repo'
         data = self._prepare_data(backend)
         data[data_key] = repo_name
         id_, params = build_data(
             self.apikey_regular, 'create_pull_request', **data)
         response = api_call(self.app, params)
         expected_message = 'repository `{}` does not exist'.format(repo_name)
         assert_error(id_, expected_message, given=response.body)
 
     @pytest.mark.backends("git", "hg")
     @pytest.mark.parametrize("data_key", ["source_ref", "target_ref"])
     def test_create_fails_with_non_existing_branch(self, backend, data_key):
         branch_name = 'test-branch'
         data = self._prepare_data(backend)
         data[data_key] = "branch:{}".format(branch_name)
         id_, params = build_data(
             self.apikey_regular, 'create_pull_request', **data)
         response = api_call(self.app, params)
         expected_message = 'The specified value:{type}:`{name}` ' \
                            'does not exist, or is not allowed.'.format(type='branch',
                                                                        name=branch_name)
         assert_error(id_, expected_message, given=response.body)
 
     @pytest.mark.backends("git", "hg")
     @pytest.mark.parametrize("data_key", ["source_ref", "target_ref"])
     def test_create_fails_with_ref_in_a_wrong_format(self, backend, data_key):
         data = self._prepare_data(backend)
         ref = 'stange-ref'
         data[data_key] = ref
         id_, params = build_data(
             self.apikey_regular, 'create_pull_request', **data)
         response = api_call(self.app, params)
         expected_message = (
             'Ref `{ref}` given in a wrong format. Please check the API'
             ' documentation for more details'.format(ref=ref))
         assert_error(id_, expected_message, given=response.body)
 
     @pytest.mark.backends("git", "hg")
     @pytest.mark.parametrize("data_key", ["source_ref", "target_ref"])
     def test_create_fails_with_non_existing_ref(self, backend, data_key):
         commit_id = 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa10'
         ref = self._get_full_ref(backend, commit_id)
         data = self._prepare_data(backend)
         data[data_key] = ref
         id_, params = build_data(
             self.apikey_regular, 'create_pull_request', **data)
         response = api_call(self.app, params)
         expected_message = 'Ref `{}` does not exist'.format(ref)
         assert_error(id_, expected_message, given=response.body)
 
     @pytest.mark.backends("git", "hg")
     def test_create_fails_when_no_revisions(self, backend):
         data = self._prepare_data(backend, source_head='initial')
         id_, params = build_data(
             self.apikey_regular, 'create_pull_request', **data)
         response = api_call(self.app, params)
         expected_message = 'no commits found for merge between specified references'
         assert_error(id_, expected_message, given=response.body)
 
     @pytest.mark.backends("git", "hg")
     def test_create_fails_when_no_permissions(self, backend):
         data = self._prepare_data(backend)
         RepoModel().revoke_user_permission(
             self.source.repo_name, self.test_user)
         RepoModel().revoke_user_permission(
             self.source.repo_name, User.DEFAULT_USER)
 
         id_, params = build_data(
             self.apikey_regular, 'create_pull_request', **data)
         response = api_call(self.app, params)
         expected_message = 'repository `{}` does not exist'.format(
             self.source.repo_name)
         assert_error(id_, expected_message, given=response.body)
 
     def _prepare_data(
             self, backend, source_head='change', target_head='initial'):
         commits = [
             {'message': 'initial'},
             {'message': 'change'},
             {'message': 'new-feature', 'parents': ['initial']},
         ]
         self.commit_ids = backend.create_master_repo(commits)
         self.source = backend.create_repo(heads=[source_head])
         self.target = backend.create_repo(heads=[target_head])
 
         data = {
             'source_repo': self.source.repo_name,
             'target_repo': self.target.repo_name,
             'source_ref': self._get_full_ref(
                 backend, self.commit_ids[source_head]),
             'target_ref': self._get_full_ref(
                 backend, self.commit_ids[target_head]),
             'title': 'Test PR 1',
             'description': 'Test'
         }
         RepoModel().grant_user_permission(
             self.source.repo_name, self.TEST_USER_LOGIN, 'repository.read')
         return data
 
     def _get_full_ref(self, backend, commit_id):
         return 'branch:{branch}:{commit_id}'.format(
             branch=backend.default_branch_name, commit_id=commit_id)

rhodecode/api/tests/test_create_repo.py

@@ -1,350 +1,348 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 json
-
 import mock
 import pytest
 
-from rhodecode.lib.utils2 import safe_unicode
 from rhodecode.lib.vcs import settings
 from rhodecode.model.meta import Session
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.user import UserModel
 from rhodecode.tests import TEST_USER_ADMIN_LOGIN
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_ok, assert_error, crash)
 from rhodecode.tests.fixture import Fixture
+from rhodecode.lib.ext_json import json
+from rhodecode.lib.str_utils import safe_str
 
 
 fixture = Fixture()
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestCreateRepo(object):
 
     @pytest.mark.parametrize('given, expected_name, expected_exc', [
         ('api repo-1', 'api-repo-1', False),
         ('api-repo 1-ąć', 'api-repo-1-ąć', False),
         (u'unicode-ąć', u'unicode-ąć', False),
         ('some repo v1.2', 'some-repo-v1.2', False),
         ('v2.0', 'v2.0', False),
     ])
     def test_api_create_repo(self, backend, given, expected_name, expected_exc):
 
         id_, params = build_data(
             self.apikey,
             'create_repo',
             repo_name=given,
             owner=TEST_USER_ADMIN_LOGIN,
             repo_type=backend.alias,
         )
         response = api_call(self.app, params)
 
         ret = {
             'msg': 'Created new repository `%s`' % (expected_name,),
             'success': True,
             'task': None,
         }
         expected = ret
         assert_ok(id_, expected, given=response.body)
 
-        repo = RepoModel().get_by_repo_name(safe_unicode(expected_name))
+        repo = RepoModel().get_by_repo_name(safe_str(expected_name))
         assert repo is not None
 
         id_, params = build_data(self.apikey, 'get_repo', repoid=expected_name)
         response = api_call(self.app, params)
         body = json.loads(response.body)
 
         assert body['result']['enable_downloads'] is False
         assert body['result']['enable_locking'] is False
         assert body['result']['enable_statistics'] is False
 
-        fixture.destroy_repo(safe_unicode(expected_name))
+        fixture.destroy_repo(safe_str(expected_name))
 
     def test_api_create_restricted_repo_type(self, backend):
         repo_name = 'api-repo-type-{0}'.format(backend.alias)
         id_, params = build_data(
             self.apikey,
             'create_repo',
             repo_name=repo_name,
             owner=TEST_USER_ADMIN_LOGIN,
             repo_type=backend.alias,
         )
         git_backend = settings.BACKENDS['git']
         with mock.patch(
                 'rhodecode.lib.vcs.settings.BACKENDS', {'git': git_backend}):
             response = api_call(self.app, params)
 
         repo = RepoModel().get_by_repo_name(repo_name)
 
         if backend.alias == 'git':
             assert repo is not None
             expected = {
                 'msg': 'Created new repository `{0}`'.format(repo_name,),
                 'success': True,
                 'task': None,
             }
             assert_ok(id_, expected, given=response.body)
         else:
             assert repo is None
 
         fixture.destroy_repo(repo_name)
 
     def test_api_create_repo_with_booleans(self, backend):
         repo_name = 'api-repo-2'
         id_, params = build_data(
             self.apikey,
             'create_repo',
             repo_name=repo_name,
             owner=TEST_USER_ADMIN_LOGIN,
             repo_type=backend.alias,
             enable_statistics=True,
             enable_locking=True,
             enable_downloads=True
         )
         response = api_call(self.app, params)
 
         repo = RepoModel().get_by_repo_name(repo_name)
 
         assert repo is not None
         ret = {
             'msg': 'Created new repository `%s`' % (repo_name,),
             'success': True,
             'task': None,
         }
         expected = ret
         assert_ok(id_, expected, given=response.body)
 
         id_, params = build_data(self.apikey, 'get_repo', repoid=repo_name)
         response = api_call(self.app, params)
         body = json.loads(response.body)
 
         assert body['result']['enable_downloads'] is True
         assert body['result']['enable_locking'] is True
         assert body['result']['enable_statistics'] is True
 
         fixture.destroy_repo(repo_name)
 
     def test_api_create_repo_in_group(self, backend):
         repo_group_name = 'my_gr'
         # create the parent
         fixture.create_repo_group(repo_group_name)
 
         repo_name = '%s/api-repo-gr' % (repo_group_name,)
         id_, params = build_data(
             self.apikey, 'create_repo',
             repo_name=repo_name,
             owner=TEST_USER_ADMIN_LOGIN,
             repo_type=backend.alias,)
         response = api_call(self.app, params)
         repo = RepoModel().get_by_repo_name(repo_name)
         assert repo is not None
         assert repo.group is not None
 
         ret = {
             'msg': 'Created new repository `%s`' % (repo_name,),
             'success': True,
             'task': None,
         }
         expected = ret
         assert_ok(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
         fixture.destroy_repo_group(repo_group_name)
 
     def test_create_repo_in_group_that_doesnt_exist(self, backend, user_util):
         repo_group_name = 'fake_group'
 
         repo_name = '%s/api-repo-gr' % (repo_group_name,)
         id_, params = build_data(
             self.apikey, 'create_repo',
             repo_name=repo_name,
             owner=TEST_USER_ADMIN_LOGIN,
             repo_type=backend.alias,)
         response = api_call(self.app, params)
 
         expected = {'repo_group': 'Repository group `{}` does not exist'.format(
             repo_group_name)}
         assert_error(id_, expected, given=response.body)
 
     def test_api_create_repo_unknown_owner(self, backend):
         repo_name = 'api-repo-2'
         owner = 'i-dont-exist'
         id_, params = build_data(
             self.apikey, 'create_repo',
             repo_name=repo_name,
             owner=owner,
             repo_type=backend.alias)
         response = api_call(self.app, params)
         expected = 'user `%s` does not exist' % (owner,)
         assert_error(id_, expected, given=response.body)
 
     def test_api_create_repo_dont_specify_owner(self, backend):
         repo_name = 'api-repo-3'
         id_, params = build_data(
             self.apikey, 'create_repo',
             repo_name=repo_name,
             repo_type=backend.alias)
         response = api_call(self.app, params)
 
         repo = RepoModel().get_by_repo_name(repo_name)
         assert repo is not None
         ret = {
             'msg': 'Created new repository `%s`' % (repo_name,),
             'success': True,
             'task': None,
         }
         expected = ret
         assert_ok(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
 
     def test_api_create_repo_by_non_admin(self, backend):
         repo_name = 'api-repo-4'
         id_, params = build_data(
             self.apikey_regular, 'create_repo',
             repo_name=repo_name,
             repo_type=backend.alias)
         response = api_call(self.app, params)
 
         repo = RepoModel().get_by_repo_name(repo_name)
         assert repo is not None
         ret = {
             'msg': 'Created new repository `%s`' % (repo_name,),
             'success': True,
             'task': None,
         }
         expected = ret
         assert_ok(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
 
     def test_api_create_repo_by_non_admin_specify_owner(self, backend):
         repo_name = 'api-repo-5'
         owner = 'i-dont-exist'
         id_, params = build_data(
             self.apikey_regular, 'create_repo',
             repo_name=repo_name,
             repo_type=backend.alias,
             owner=owner)
         response = api_call(self.app, params)
 
         expected = 'Only RhodeCode super-admin can specify `owner` param'
         assert_error(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
 
     def test_api_create_repo_by_non_admin_no_parent_group_perms(self, backend):
         repo_group_name = 'no-access'
         fixture.create_repo_group(repo_group_name)
         repo_name = 'no-access/api-repo'
 
         id_, params = build_data(
             self.apikey_regular, 'create_repo',
             repo_name=repo_name,
             repo_type=backend.alias)
         response = api_call(self.app, params)
 
         expected = {'repo_group': 'Repository group `{}` does not exist'.format(
             repo_group_name)}
         assert_error(id_, expected, given=response.body)
         fixture.destroy_repo_group(repo_group_name)
         fixture.destroy_repo(repo_name)
 
     def test_api_create_repo_non_admin_no_permission_to_create_to_root_level(
             self, backend, user_util):
 
         regular_user = user_util.create_user()
         regular_user_api_key = regular_user.api_key
 
         usr = UserModel().get_by_username(regular_user.username)
         usr.inherit_default_permissions = False
         Session().add(usr)
 
         repo_name = backend.new_repo_name()
         id_, params = build_data(
             regular_user_api_key, 'create_repo',
             repo_name=repo_name,
             repo_type=backend.alias)
         response = api_call(self.app, params)
         expected = {
             "repo_name": "You do not have the permission to "
                          "store repositories in the root location."}
         assert_error(id_, expected, given=response.body)
 
     def test_api_create_repo_exists(self, backend):
         repo_name = backend.repo_name
         id_, params = build_data(
             self.apikey, 'create_repo',
             repo_name=repo_name,
             owner=TEST_USER_ADMIN_LOGIN,
             repo_type=backend.alias,)
         response = api_call(self.app, params)
         expected = {
             'unique_repo_name': 'Repository with name `{}` already exists'.format(
                 repo_name)}
         assert_error(id_, expected, given=response.body)
 
     @mock.patch.object(RepoModel, 'create', crash)
     def test_api_create_repo_exception_occurred(self, backend):
         repo_name = 'api-repo-6'
         id_, params = build_data(
             self.apikey, 'create_repo',
             repo_name=repo_name,
             owner=TEST_USER_ADMIN_LOGIN,
             repo_type=backend.alias,)
         response = api_call(self.app, params)
         expected = 'failed to create repository `%s`' % (repo_name,)
         assert_error(id_, expected, given=response.body)
 
     @pytest.mark.parametrize('parent_group, dirty_name, expected_name', [
         (None, 'foo bar x', 'foo-bar-x'),
         ('foo', '/foo//bar x', 'foo/bar-x'),
         ('foo-bar', 'foo-bar //bar x', 'foo-bar/bar-x'),
     ])
     def test_create_repo_with_extra_slashes_in_name(
             self, backend, parent_group, dirty_name, expected_name):
 
         if parent_group:
             gr = fixture.create_repo_group(parent_group)
             assert gr.group_name == parent_group
 
         id_, params = build_data(
             self.apikey, 'create_repo',
             repo_name=dirty_name,
             repo_type=backend.alias,
             owner=TEST_USER_ADMIN_LOGIN,)
         response = api_call(self.app, params)
         expected ={
            "msg": "Created new repository `{}`".format(expected_name),
            "task": None,
            "success": True
         }
         assert_ok(id_, expected, response.body)
 
         repo = RepoModel().get_by_repo_name(expected_name)
         assert repo is not None
 
         expected = {
             'msg': 'Created new repository `%s`' % (expected_name,),
             'success': True,
             'task': None,
         }
         assert_ok(id_, expected, given=response.body)
         fixture.destroy_repo(expected_name)
         if parent_group:
             fixture.destroy_repo_group(parent_group)

rhodecode/api/tests/test_create_repo_group.py

@@ -1,289 +1,288 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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.meta import Session
 from rhodecode.model.repo_group import RepoGroupModel
 from rhodecode.model.user import UserModel
 from rhodecode.tests import TEST_USER_ADMIN_LOGIN
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_ok, assert_error, crash)
 from rhodecode.tests.fixture import Fixture
 
 
 fixture = Fixture()
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestCreateRepoGroup(object):
     def test_api_create_repo_group(self):
         repo_group_name = 'api-repo-group'
 
         repo_group = RepoGroupModel.cls.get_by_group_name(repo_group_name)
         assert repo_group is None
 
         id_, params = build_data(
             self.apikey, 'create_repo_group',
             group_name=repo_group_name,
             owner=TEST_USER_ADMIN_LOGIN,)
         response = api_call(self.app, params)
 
         repo_group = RepoGroupModel.cls.get_by_group_name(repo_group_name)
         assert repo_group is not None
         ret = {
             'msg': 'Created new repo group `%s`' % (repo_group_name,),
             'repo_group': repo_group.get_api_data()
         }
         expected = ret
         try:
             assert_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo_group(repo_group_name)
 
     def test_api_create_repo_group_in_another_group(self):
         repo_group_name = 'api-repo-group'
 
         repo_group = RepoGroupModel.cls.get_by_group_name(repo_group_name)
         assert repo_group is None
         # create the parent
         fixture.create_repo_group(repo_group_name)
 
         full_repo_group_name = repo_group_name+'/'+repo_group_name
         id_, params = build_data(
             self.apikey, 'create_repo_group',
             group_name=full_repo_group_name,
             owner=TEST_USER_ADMIN_LOGIN,
             copy_permissions=True)
         response = api_call(self.app, params)
 
         repo_group = RepoGroupModel.cls.get_by_group_name(full_repo_group_name)
         assert repo_group is not None
         ret = {
             'msg': 'Created new repo group `%s`' % (full_repo_group_name,),
             'repo_group': repo_group.get_api_data()
         }
         expected = ret
         try:
             assert_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo_group(full_repo_group_name)
             fixture.destroy_repo_group(repo_group_name)
 
     def test_api_create_repo_group_in_another_group_not_existing(self):
         repo_group_name = 'api-repo-group-no'
 
         repo_group = RepoGroupModel.cls.get_by_group_name(repo_group_name)
         assert repo_group is None
 
         full_repo_group_name = repo_group_name+'/'+repo_group_name
         id_, params = build_data(
             self.apikey, 'create_repo_group',
             group_name=full_repo_group_name,
             owner=TEST_USER_ADMIN_LOGIN,
             copy_permissions=True)
         response = api_call(self.app, params)
         expected = {
             'repo_group':
                 'Parent repository group `{}` does not exist'.format(
                     repo_group_name)}
         assert_error(id_, expected, given=response.body)
 
     def test_api_create_repo_group_that_exists(self):
         repo_group_name = 'api-repo-group'
 
         repo_group = RepoGroupModel.cls.get_by_group_name(repo_group_name)
         assert repo_group is None
 
         fixture.create_repo_group(repo_group_name)
         id_, params = build_data(
             self.apikey, 'create_repo_group',
             group_name=repo_group_name,
             owner=TEST_USER_ADMIN_LOGIN,)
         response = api_call(self.app, params)
         expected = {
             'unique_repo_group_name':
                 'Repository group with name `{}` already exists'.format(
                     repo_group_name)}
         try:
             assert_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo_group(repo_group_name)
 
     def test_api_create_repo_group_regular_user_wit_root_location_perms(
             self, user_util):
         regular_user = user_util.create_user()
         regular_user_api_key = regular_user.api_key
 
         repo_group_name = 'api-repo-group-by-regular-user'
 
         usr = UserModel().get_by_username(regular_user.username)
         usr.inherit_default_permissions = False
         Session().add(usr)
 
         UserModel().grant_perm(
             regular_user.username, 'hg.repogroup.create.true')
         Session().commit()
 
         repo_group = RepoGroupModel.cls.get_by_group_name(repo_group_name)
         assert repo_group is None
 
         id_, params = build_data(
             regular_user_api_key, 'create_repo_group',
             group_name=repo_group_name)
         response = api_call(self.app, params)
 
         repo_group = RepoGroupModel.cls.get_by_group_name(repo_group_name)
         assert repo_group is not None
         expected = {
             'msg': 'Created new repo group `%s`' % (repo_group_name,),
             'repo_group': repo_group.get_api_data()
         }
         try:
             assert_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo_group(repo_group_name)
 
     def test_api_create_repo_group_regular_user_with_admin_perms_to_parent(
             self, user_util):
 
         repo_group_name = 'api-repo-group-parent'
 
         repo_group = RepoGroupModel.cls.get_by_group_name(repo_group_name)
         assert repo_group is None
         # create the parent
         fixture.create_repo_group(repo_group_name)
 
         # user perms
         regular_user = user_util.create_user()
         regular_user_api_key = regular_user.api_key
 
         usr = UserModel().get_by_username(regular_user.username)
         usr.inherit_default_permissions = False
         Session().add(usr)
 
         RepoGroupModel().grant_user_permission(
             repo_group_name, regular_user.username, 'group.admin')
         Session().commit()
 
         full_repo_group_name = repo_group_name + '/' + repo_group_name
         id_, params = build_data(
             regular_user_api_key, 'create_repo_group',
             group_name=full_repo_group_name)
         response = api_call(self.app, params)
 
         repo_group = RepoGroupModel.cls.get_by_group_name(full_repo_group_name)
         assert repo_group is not None
         expected = {
             'msg': 'Created new repo group `{}`'.format(full_repo_group_name),
             'repo_group': repo_group.get_api_data()
         }
         try:
             assert_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo_group(full_repo_group_name)
             fixture.destroy_repo_group(repo_group_name)
 
     def test_api_create_repo_group_regular_user_no_permission_to_create_to_root_level(self):
         repo_group_name = 'api-repo-group'
 
         id_, params = build_data(
             self.apikey_regular, 'create_repo_group',
             group_name=repo_group_name)
         response = api_call(self.app, params)
 
         expected = {
             'repo_group':
                 u'You do not have the permission to store '
                 u'repository groups in the root location.'}
         assert_error(id_, expected, given=response.body)
 
     def test_api_create_repo_group_regular_user_no_parent_group_perms(self):
         repo_group_name = 'api-repo-group-regular-user'
 
         repo_group = RepoGroupModel.cls.get_by_group_name(repo_group_name)
         assert repo_group is None
         # create the parent
         fixture.create_repo_group(repo_group_name)
 
         full_repo_group_name = repo_group_name+'/'+repo_group_name
 
         id_, params = build_data(
             self.apikey_regular, 'create_repo_group',
             group_name=full_repo_group_name)
         response = api_call(self.app, params)
 
         expected = {
             'repo_group':
                 u"You do not have the permissions to store "
                 u"repository groups inside repository group `{}`".format(repo_group_name)}
         try:
             assert_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo_group(repo_group_name)
 
     def test_api_create_repo_group_regular_user_no_permission_to_specify_owner(
             self):
         repo_group_name = 'api-repo-group'
 
         id_, params = build_data(
             self.apikey_regular, 'create_repo_group',
             group_name=repo_group_name,
             owner=TEST_USER_ADMIN_LOGIN,)
         response = api_call(self.app, params)
 
         expected = "Only RhodeCode super-admin can specify `owner` param"
         assert_error(id_, expected, given=response.body)
 
     @mock.patch.object(RepoGroupModel, 'create', crash)
     def test_api_create_repo_group_exception_occurred(self):
         repo_group_name = 'api-repo-group'
 
         repo_group = RepoGroupModel.cls.get_by_group_name(repo_group_name)
         assert repo_group is None
 
         id_, params = build_data(
             self.apikey, 'create_repo_group',
             group_name=repo_group_name,
             owner=TEST_USER_ADMIN_LOGIN,)
         response = api_call(self.app, params)
         expected = 'failed to create repo group `%s`' % (repo_group_name,)
         assert_error(id_, expected, given=response.body)
 
     def test_create_group_with_extra_slashes_in_name(self, user_util):
         existing_repo_group = user_util.create_repo_group()
         dirty_group_name = '//{}//group2//'.format(
             existing_repo_group.group_name)
         cleaned_group_name = '{}/group2'.format(
             existing_repo_group.group_name)
 
         id_, params = build_data(
             self.apikey, 'create_repo_group',
             group_name=dirty_group_name,
             owner=TEST_USER_ADMIN_LOGIN,)
         response = api_call(self.app, params)
         repo_group = RepoGroupModel.cls.get_by_group_name(cleaned_group_name)
         expected = {
             'msg': 'Created new repo group `%s`' % (cleaned_group_name,),
             'repo_group': repo_group.get_api_data()
         }
         assert_ok(id_, expected, given=response.body)
         fixture.destroy_repo_group(cleaned_group_name)

rhodecode/api/tests/test_create_user.py

@@ -1,207 +1,206 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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.lib.auth import check_password
 from rhodecode.model.user import UserModel
 from rhodecode.tests import (
     TEST_USER_ADMIN_LOGIN, TEST_USER_REGULAR_EMAIL)
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_ok, assert_error, jsonify, crash)
 from rhodecode.tests.fixture import Fixture
 from rhodecode.model.db import RepoGroup
 
 
 # TODO: mikhail: remove fixture from here
 fixture = Fixture()
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestCreateUser(object):
     def test_api_create_existing_user(self):
         id_, params = build_data(
             self.apikey, 'create_user',
             username=TEST_USER_ADMIN_LOGIN,
             email='[e-mail]',
             password='trololo')
         response = api_call(self.app, params)
 
         expected = "user `%s` already exist" % (TEST_USER_ADMIN_LOGIN,)
         assert_error(id_, expected, given=response.body)
 
     def test_api_create_user_with_existing_email(self):
         id_, params = build_data(
             self.apikey, 'create_user',
             username=TEST_USER_ADMIN_LOGIN + 'new',
             email=TEST_USER_REGULAR_EMAIL,
             password='trololo')
         response = api_call(self.app, params)
 
         expected = "email `%s` already exist" % (TEST_USER_REGULAR_EMAIL,)
         assert_error(id_, expected, given=response.body)
 
     def test_api_create_user_with_wrong_username(self):
         bad_username = '<> HELLO WORLD <>'
         id_, params = build_data(
             self.apikey, 'create_user',
             username=bad_username,
             email='[e-mail]',
             password='trololo')
         response = api_call(self.app, params)
 
         expected = {'username':
                         "Username may only contain alphanumeric characters "
                         "underscores, periods or dashes and must begin with "
                         "alphanumeric character or underscore"}
         assert_error(id_, expected, given=response.body)
 
     def test_api_create_user(self):
         username = 'test_new_api_user'
         email = username + "@foo.com"
 
         id_, params = build_data(
             self.apikey, 'create_user',
             username=username,
             email=email,
             description='CTO of Things',
             password='example')
         response = api_call(self.app, params)
 
         usr = UserModel().get_by_username(username)
         ret = {
             'msg': 'created new user `%s`' % (username,),
             'user': jsonify(usr.get_api_data(include_secrets=True)),
         }
         try:
             expected = ret
             assert check_password('example', usr.password)
             assert_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_user(usr.user_id)
 
     def test_api_create_user_without_password(self):
         username = 'test_new_api_user_passwordless'
         email = username + "@foo.com"
 
         id_, params = build_data(
             self.apikey, 'create_user',
             username=username,
             email=email)
         response = api_call(self.app, params)
 
         usr = UserModel().get_by_username(username)
         ret = {
             'msg': 'created new user `%s`' % (username,),
             'user': jsonify(usr.get_api_data(include_secrets=True)),
         }
         try:
             expected = ret
             assert_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_user(usr.user_id)
 
     def test_api_create_user_with_extern_name(self):
         username = 'test_new_api_user_passwordless'
         email = username + "@foo.com"
 
         id_, params = build_data(
             self.apikey, 'create_user',
             username=username,
             email=email, extern_name='rhodecode')
         response = api_call(self.app, params)
 
         usr = UserModel().get_by_username(username)
         ret = {
             'msg': 'created new user `%s`' % (username,),
             'user': jsonify(usr.get_api_data(include_secrets=True)),
         }
         try:
             expected = ret
             assert_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_user(usr.user_id)
 
     def test_api_create_user_with_password_change(self):
         username = 'test_new_api_user_password_change'
         email = username + "@foo.com"
 
         id_, params = build_data(
             self.apikey, 'create_user',
             username=username,
             email=email, extern_name='rhodecode',
             force_password_change=True)
         response = api_call(self.app, params)
 
         usr = UserModel().get_by_username(username)
         ret = {
             'msg': 'created new user `%s`' % (username,),
             'user': jsonify(usr.get_api_data(include_secrets=True)),
         }
         try:
             expected = ret
             assert_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_user(usr.user_id)
 
     def test_api_create_user_with_personal_repo_group(self):
         username = 'test_new_api_user_personal_group'
         email = username + "@foo.com"
 
         id_, params = build_data(
             self.apikey, 'create_user',
             username=username,
             email=email, extern_name='rhodecode',
             create_personal_repo_group=True)
         response = api_call(self.app, params)
 
         usr = UserModel().get_by_username(username)
         ret = {
             'msg': 'created new user `%s`' % (username,),
             'user': jsonify(usr.get_api_data(include_secrets=True)),
         }
 
         personal_group = RepoGroup.get_by_group_name(username)
         assert personal_group
         assert personal_group.personal == True
         assert personal_group.user.username == username
 
         try:
             expected = ret
             assert_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo_group(username)
             fixture.destroy_user(usr.user_id)
 
     @mock.patch.object(UserModel, 'create_or_update', crash)
     def test_api_create_user_when_exception_happened(self):
 
         username = 'test_new_api_user'
         email = username + "@foo.com"
 
         id_, params = build_data(
             self.apikey, 'create_user',
             username=username,
             email=email,
             password='trololo')
         response = api_call(self.app, params)
         expected = 'failed to create user `%s`' % (username,)
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_create_user_group.py

@@ -1,127 +1,126 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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.meta import Session
 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, jsonify)
 from rhodecode.tests.fixture import Fixture
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestCreateUserGroup(object):
     fixture = Fixture()
 
     def test_api_create_user_group(self):
         group_name = 'some_new_group'
         id_, params = build_data(
             self.apikey, 'create_user_group', group_name=group_name)
         response = api_call(self.app, params)
 
         ret = {
             'msg': 'created new user group `%s`' % (group_name,),
             'user_group': jsonify(
                 UserGroupModel()
                 .get_by_name(group_name)
                 .get_api_data()
             )
         }
         expected = ret
         assert_ok(id_, expected, given=response.body)
         self.fixture.destroy_user_group(group_name)
 
     def test_api_create_user_group_regular_user(self):
         group_name = 'some_new_group'
 
         usr = UserModel().get_by_username(self.TEST_USER_LOGIN)
         usr.inherit_default_permissions = False
         Session().add(usr)
         UserModel().grant_perm(
             self.TEST_USER_LOGIN, 'hg.usergroup.create.true')
         Session().commit()
 
         id_, params = build_data(
             self.apikey_regular, 'create_user_group', group_name=group_name)
         response = api_call(self.app, params)
 
         expected = {
             'msg': 'created new user group `%s`' % (group_name,),
             'user_group': jsonify(
                 UserGroupModel()
                 .get_by_name(group_name)
                 .get_api_data()
             )
         }
         try:
             assert_ok(id_, expected, given=response.body)
         finally:
             self.fixture.destroy_user_group(group_name)
             UserModel().revoke_perm(
                 self.TEST_USER_LOGIN, 'hg.usergroup.create.true')
             usr = UserModel().get_by_username(self.TEST_USER_LOGIN)
             usr.inherit_default_permissions = True
             Session().add(usr)
             Session().commit()
 
     def test_api_create_user_group_regular_user_no_permission(self):
         group_name = 'some_new_group'
         id_, params = build_data(
             self.apikey_regular, 'create_user_group', group_name=group_name)
         response = api_call(self.app, params)
         expected = "Access was denied to this resource."
         assert_error(id_, expected, given=response.body)
 
     def test_api_create_user_group_that_exist(self, user_util):
         group = user_util.create_user_group()
         group_name = group.users_group_name
 
         id_, params = build_data(
             self.apikey, 'create_user_group', group_name=group_name)
         response = api_call(self.app, params)
 
         expected = "user group `%s` already exist" % (group_name,)
         assert_error(id_, expected, given=response.body)
 
     @mock.patch.object(UserGroupModel, 'create', crash)
     def test_api_create_user_group_exception_occurred(self):
         group_name = 'exception_happens'
         id_, params = build_data(
             self.apikey, 'create_user_group', group_name=group_name)
         response = api_call(self.app, params)
 
         expected = 'failed to create group `%s`' % (group_name,)
         assert_error(id_, expected, given=response.body)
 
     def test_api_create_user_group_with_wrong_name(self, user_util):
 
         group_name = 'wrong NAME <>'
         id_, params = build_data(
             self.apikey, 'create_user_group', group_name=group_name)
         response = api_call(self.app, params)
 
         expected = {"user_group_name":
                         "Allowed in name are letters, numbers, and `-`, `_`, "
                         "`.` Name must start with a letter or number. "
                         "Got `{}`".format(group_name)}
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_delete_gist.py

@@ -1,61 +1,60 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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.gist import GistModel
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_error, assert_ok, crash)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestApiDeleteGist(object):
     def test_api_delete_gist(self, gist_util):
         gist_id = gist_util.create_gist().gist_access_id
         id_, params = build_data(self.apikey, 'delete_gist', gistid=gist_id)
         response = api_call(self.app, params)
         expected = {'gist': None, 'msg': 'deleted gist ID:%s' % (gist_id,)}
         assert_ok(id_, expected, given=response.body)
 
     def test_api_delete_gist_regular_user(self, gist_util):
         gist_id = gist_util.create_gist(
             owner=self.TEST_USER_LOGIN).gist_access_id
         id_, params = build_data(
             self.apikey_regular, 'delete_gist', gistid=gist_id)
         response = api_call(self.app, params)
         expected = {'gist': None, 'msg': 'deleted gist ID:%s' % (gist_id,)}
         assert_ok(id_, expected, given=response.body)
 
     def test_api_delete_gist_regular_user_no_permission(self, gist_util):
         gist_id = gist_util.create_gist().gist_access_id
         id_, params = build_data(
             self.apikey_regular, 'delete_gist', gistid=gist_id)
         response = api_call(self.app, params)
         expected = 'gist `%s` does not exist' % (gist_id,)
         assert_error(id_, expected, given=response.body)
 
     @mock.patch.object(GistModel, 'delete', crash)
     def test_api_delete_gist_exception_occurred(self, gist_util):
         gist_id = gist_util.create_gist().gist_access_id
         id_, params = build_data(self.apikey, 'delete_gist', gistid=gist_id)
         response = api_call(self.app, params)
         expected = 'failed to delete gist ID:%s' % (gist_id,)
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_delete_repo.py

@@ -1,74 +1,73 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 TestApiDeleteRepo(object):
     def test_api_delete_repo(self, backend):
         repo = backend.create_repo()
         repo_name = repo.repo_name
         id_, params = build_data(
             self.apikey, 'delete_repo', repoid=repo.repo_name, )
         response = api_call(self.app, params)
 
         expected = {
             'msg': 'Deleted repository `%s`' % (repo_name,),
             'success': True
         }
         assert_ok(id_, expected, given=response.body)
 
     def test_api_delete_repo_by_non_admin(self, backend, user_regular):
         repo = backend.create_repo(cur_user=user_regular.username)
         repo_name = repo.repo_name
         id_, params = build_data(
             user_regular.api_key, 'delete_repo', repoid=repo.repo_name, )
         response = api_call(self.app, params)
 
         expected = {
             'msg': 'Deleted repository `%s`' % (repo_name,),
             'success': True
         }
         assert_ok(id_, expected, given=response.body)
 
     def test_api_delete_repo_by_non_admin_no_permission(self, backend):
         repo = backend.create_repo()
         repo_name = repo.repo_name
         id_, params = build_data(
             self.apikey_regular, 'delete_repo', repoid=repo.repo_name, )
         response = api_call(self.app, params)
         expected = 'repository `%s` does not exist' % (repo_name)
         assert_error(id_, expected, given=response.body)
 
     def test_api_delete_repo_exception_occurred(self, backend):
         repo = backend.create_repo()
         repo_name = repo.repo_name
         id_, params = build_data(
             self.apikey, 'delete_repo', repoid=repo.repo_name, )
         with mock.patch.object(RepoModel, 'delete', crash):
             response = api_call(self.app, params)
         expected = 'failed to delete repository `%s`' % (repo_name,)
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_delete_repo_group.py

@@ -1,86 +1,85 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 
 from rhodecode.model.repo_group import RepoGroupModel
 from rhodecode.model.user import UserModel
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_error, assert_ok)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestApiDeleteRepoGroup(object):
     def test_api_delete_repo_group(self, user_util):
         repo_group = user_util.create_repo_group(auto_cleanup=False)
         repo_group_name = repo_group.group_name
         repo_group_id = repo_group.group_id
         id_, params = build_data(
             self.apikey, 'delete_repo_group', repogroupid=repo_group_name, )
         response = api_call(self.app, params)
 
         ret = {
             'msg': 'deleted repo group ID:%s %s' % (
                 repo_group_id, repo_group_name
             ),
             'repo_group': None
         }
         expected = ret
         assert_ok(id_, expected, given=response.body)
         gr = RepoGroupModel()._get_repo_group(repo_group_name)
         assert gr is None
 
     def test_api_delete_repo_group_regular_user(self, user_util):
         repo_group = user_util.create_repo_group(auto_cleanup=False)
         repo_group_name = repo_group.group_name
         repo_group_id = repo_group.group_id
 
         user = UserModel().get_by_username(self.TEST_USER_LOGIN)
         user_util.grant_user_permission_to_repo_group(
             repo_group, user, 'group.admin')
 
         id_, params = build_data(
             self.apikey, 'delete_repo_group', repogroupid=repo_group_name, )
         response = api_call(self.app, params)
 
         ret = {
             'msg': 'deleted repo group ID:%s %s' % (
                 repo_group_id, repo_group_name
             ),
             'repo_group': None
         }
         expected = ret
         assert_ok(id_, expected, given=response.body)
         gr = RepoGroupModel()._get_repo_group(repo_group_name)
         assert gr is None
 
     def test_api_delete_repo_group_regular_user_no_permission(self, user_util):
         repo_group = user_util.create_repo_group()
         repo_group_name = repo_group.group_name
 
         id_, params = build_data(
             self.apikey_regular, 'delete_repo_group',
             repogroupid=repo_group_name, )
         response = api_call(self.app, params)
 
         expected = 'repository group `%s` does not exist' % (
             repo_group_name,)
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_delete_user.py

@@ -1,57 +1,56 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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.api.tests.utils import (
     build_data, api_call, assert_ok, assert_error, crash)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestDeleteUser(object):
     def test_api_delete_user(self, user_util):
         usr = user_util.create_user(auto_cleanup=False)
 
         username = usr.username
         usr_id = usr.user_id
 
         id_, params = build_data(self.apikey, 'delete_user', userid=username)
         response = api_call(self.app, params)
 
         ret = {'msg': 'deleted user ID:%s %s' % (usr_id, username),
                'user': None}
         expected = ret
         assert_ok(id_, expected, given=response.body)
 
     @mock.patch.object(UserModel, 'delete', crash)
     def test_api_delete_user_when_exception_happened(self, user_util):
         usr = user_util.create_user()
         username = usr.username
 
         id_, params = build_data(
             self.apikey, 'delete_user', userid=username, )
         response = api_call(self.app, params)
         ret = 'failed to delete user ID:%s %s' % (usr.user_id,
                                                   usr.username)
         expected = ret
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_delete_user_group.py

@@ -1,106 +1,105 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 TestDeleteUserGroup(object):
     def test_api_delete_user_group(self, user_util):
         user_group = user_util.create_user_group(auto_cleanup=False)
         group_name = user_group.users_group_name
         group_id = user_group.users_group_id
         id_, params = build_data(
             self.apikey, 'delete_user_group', usergroupid=group_name)
         response = api_call(self.app, params)
 
         expected = {
             'user_group': None,
             'msg': 'deleted user group ID:%s %s' % (group_id, group_name)
         }
         assert_ok(id_, expected, given=response.body)
 
     def test_api_delete_user_group_regular_user(self, user_util):
         ugroup = user_util.create_user_group(auto_cleanup=False)
         group_name = ugroup.users_group_name
         group_id = ugroup.users_group_id
         user = UserModel().get_by_username(self.TEST_USER_LOGIN)
 
         user_util.grant_user_permission_to_user_group(
             ugroup, user, 'usergroup.admin')
 
         id_, params = build_data(
             self.apikey_regular, 'delete_user_group', usergroupid=group_name)
         response = api_call(self.app, params)
 
         expected = {
             'user_group': None,
             'msg': 'deleted user group ID:%s %s' % (group_id, group_name)
         }
         assert_ok(id_, expected, given=response.body)
 
     def test_api_delete_user_group_regular_user_no_permission(self, user_util):
         user_group = user_util.create_user_group()
         group_name = user_group.users_group_name
 
         id_, params = build_data(
             self.apikey_regular, 'delete_user_group', usergroupid=group_name)
         response = api_call(self.app, params)
 
         expected = 'user group `%s` does not exist' % (group_name)
         assert_error(id_, expected, given=response.body)
 
     def test_api_delete_user_group_that_is_assigned(self, backend, user_util):
         ugroup = user_util.create_user_group()
         group_name = ugroup.users_group_name
         repo = backend.create_repo()
 
         ugr_to_perm = user_util.grant_user_group_permission_to_repo(
             repo, ugroup, 'repository.write')
         msg = 'UserGroup assigned to %s' % (ugr_to_perm.repository)
 
         id_, params = build_data(
             self.apikey, 'delete_user_group',
             usergroupid=group_name)
         response = api_call(self.app, params)
 
         expected = msg
         assert_error(id_, expected, given=response.body)
 
     def test_api_delete_user_group_exception_occurred(self, user_util):
         ugroup = user_util.create_user_group()
         group_name = ugroup.users_group_name
         group_id = ugroup.users_group_id
         id_, params = build_data(
             self.apikey, 'delete_user_group',
             usergroupid=group_name)
 
         with mock.patch.object(UserGroupModel, 'delete', crash):
             response = api_call(self.app, params)
             expected = 'failed to delete user group ID:%s %s' % (
                 group_id, group_name)
             assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_deprecated_api.py

@@ -1,79 +1,78 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 
 from rhodecode.api.views import deprecated_api
 from rhodecode.lib.ext_json import json
 from rhodecode.api.tests.utils import (
     build_data, api_call)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestCommitComment(object):
     def test_deprecated_message_in_docstring(self):
         docstring = deprecated_api.changeset_comment.__doc__
         assert '.. deprecated:: 3.4.0' in docstring
         assert 'Please use method `comment_commit` instead.' in docstring
 
     def test_deprecated_message_in_retvalue(self):
 
         id_, params = build_data(
             self.apikey, 'show_ip')
         response = api_call(self.app, params)
 
         expected = {
             'id': id_,
             'error': None,
             'result': json.loads(response.body)['result'],
             'DEPRECATION_WARNING':
                 'DEPRECATED METHOD Please use method `get_ip` instead.'
         }
         assert expected == json.loads(response.body)
 
     # def test_calls_comment_commit(self, backend, no_notifications):
     #     data = {
     #         'repoid': backend.repo_name,
     #         'status': ChangesetStatus.STATUS_APPROVED,
     #         'message': 'Approved',
     #         'revision': 'tip'
     #     }
     #     with patch.object(repo_api, 'changeset_commit') as comment_mock:
     #         id_, params = build_data(self.apikey, 'comment_commit', **data)
     #         api_call(self.app, params)
     #
     #     _, call_args = comment_mock.call_args
     #     data['commit_id'] = data.pop('revision')
     #     for key in data:
     #         assert call_args[key] == data[key]
 
     # def test_warning_log_contains_deprecation_message(self):
     #     api = self.SampleApi()
     #     with patch.object(utils, 'log') as log_mock:
     #         api.api_method()
     #
     #     assert log_mock.warning.call_count == 1
     #     call_args = log_mock.warning.call_args[0]
     #     assert (
     #         call_args[0] ==
     #         'DEPRECATED API CALL on function %s, please use `%s` instead')
     #     assert call_args[1].__name__ == 'api_method'
     #     assert call_args[2] == 'new_method'
  No newline at end of file

rhodecode/api/tests/test_fork_repo.py

@@ -1,279 +1,278 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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.meta import Session
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.repo_group import RepoGroupModel
 from rhodecode.model.user import UserModel
 from rhodecode.tests import TEST_USER_ADMIN_LOGIN
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_error, assert_ok, crash)
 from rhodecode.tests.fixture import Fixture
 
 
 fixture = Fixture()
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestApiForkRepo(object):
     def test_api_fork_repo(self, backend):
         source_name = backend['minimal'].repo_name
         fork_name = backend.new_repo_name()
 
         id_, params = build_data(
             self.apikey, 'fork_repo',
             repoid=source_name,
             fork_name=fork_name,
             owner=TEST_USER_ADMIN_LOGIN)
         response = api_call(self.app, params)
 
         expected = {
             'msg': 'Created fork of `%s` as `%s`' % (source_name, fork_name),
             'success': True,
             'task': None,
         }
         try:
             assert_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(fork_name)
 
     def test_api_fork_repo_into_group(self, backend, user_util):
         source_name = backend['minimal'].repo_name
         repo_group = user_util.create_repo_group()
         fork_name = '%s/api-repo-fork' % repo_group.group_name
         id_, params = build_data(
             self.apikey, 'fork_repo',
             repoid=source_name,
             fork_name=fork_name,
             owner=TEST_USER_ADMIN_LOGIN)
         response = api_call(self.app, params)
 
         ret = {
             'msg': 'Created fork of `%s` as `%s`' % (source_name, fork_name),
             'success': True,
             'task': None,
         }
         expected = ret
         try:
             assert_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(fork_name)
 
     def test_api_fork_repo_non_admin(self, backend):
         source_name = backend['minimal'].repo_name
         fork_name = backend.new_repo_name()
 
         id_, params = build_data(
             self.apikey_regular, 'fork_repo',
             repoid=source_name,
             fork_name=fork_name)
         response = api_call(self.app, params)
 
         expected = {
             'msg': 'Created fork of `%s` as `%s`' % (source_name, fork_name),
             'success': True,
             'task': None,
         }
         try:
             assert_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(fork_name)
 
     def test_api_fork_repo_non_admin_into_group_no_permission(self, backend, user_util):
         source_name = backend['minimal'].repo_name
         repo_group = user_util.create_repo_group()
         repo_group_name = repo_group.group_name
         fork_name = '%s/api-repo-fork' % repo_group_name
 
         id_, params = build_data(
             self.apikey_regular, 'fork_repo',
             repoid=source_name,
             fork_name=fork_name)
         response = api_call(self.app, params)
 
         expected = {
             'repo_group': 'Repository group `{}` does not exist'.format(
                 repo_group_name)}
         try:
             assert_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(fork_name)
 
     def test_api_fork_repo_non_admin_into_group(self, backend, user_util):
         source_name = backend['minimal'].repo_name
         repo_group = user_util.create_repo_group()
         fork_name = '%s/api-repo-fork' % repo_group.group_name
 
         RepoGroupModel().grant_user_permission(
             repo_group, self.TEST_USER_LOGIN, 'group.admin')
         Session().commit()
 
         id_, params = build_data(
             self.apikey_regular, 'fork_repo',
             repoid=source_name,
             fork_name=fork_name)
         response = api_call(self.app, params)
 
         expected = {
             'msg': 'Created fork of `%s` as `%s`' % (source_name, fork_name),
             'success': True,
             'task': None,
         }
         try:
             assert_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(fork_name)
 
     def test_api_fork_repo_non_admin_specify_owner(self, backend):
         source_name = backend['minimal'].repo_name
         fork_name = backend.new_repo_name()
         id_, params = build_data(
             self.apikey_regular, 'fork_repo',
             repoid=source_name,
             fork_name=fork_name,
             owner=TEST_USER_ADMIN_LOGIN)
         response = api_call(self.app, params)
         expected = 'Only RhodeCode super-admin can specify `owner` param'
         assert_error(id_, expected, given=response.body)
 
     def test_api_fork_repo_non_admin_no_permission_of_source_repo(
             self, backend):
         source_name = backend['minimal'].repo_name
         RepoModel().grant_user_permission(repo=source_name,
                                           user=self.TEST_USER_LOGIN,
                                           perm='repository.none')
         fork_name = backend.new_repo_name()
         id_, params = build_data(
             self.apikey_regular, 'fork_repo',
             repoid=backend.repo_name,
             fork_name=fork_name)
         response = api_call(self.app, params)
         expected = 'repository `%s` does not exist' % (backend.repo_name)
         assert_error(id_, expected, given=response.body)
 
     def test_api_fork_repo_non_admin_no_permission_to_fork_to_root_level(
             self, backend, user_util):
 
         regular_user = user_util.create_user()
         regular_user_api_key = regular_user.api_key
         usr = UserModel().get_by_username(regular_user.username)
         usr.inherit_default_permissions = False
         Session().add(usr)
         UserModel().grant_perm(regular_user.username, 'hg.fork.repository')
 
         source_name = backend['minimal'].repo_name
         fork_name = backend.new_repo_name()
         id_, params = build_data(
             regular_user_api_key, 'fork_repo',
             repoid=source_name,
             fork_name=fork_name)
         response = api_call(self.app, params)
         expected = {
             "repo_name": "You do not have the permission to "
                          "store repositories in the root location."}
         assert_error(id_, expected, given=response.body)
 
     def test_api_fork_repo_non_admin_no_permission_to_fork(
             self, backend, user_util):
 
         regular_user = user_util.create_user()
         regular_user_api_key = regular_user.api_key
         usr = UserModel().get_by_username(regular_user.username)
         usr.inherit_default_permissions = False
         Session().add(usr)
 
         source_name = backend['minimal'].repo_name
         fork_name = backend.new_repo_name()
         id_, params = build_data(
             regular_user_api_key, 'fork_repo',
             repoid=source_name,
             fork_name=fork_name)
         response = api_call(self.app, params)
 
         expected = "Access was denied to this resource."
         assert_error(id_, expected, given=response.body)
 
     def test_api_fork_repo_unknown_owner(self, backend):
         source_name = backend['minimal'].repo_name
         fork_name = backend.new_repo_name()
         owner = 'i-dont-exist'
         id_, params = build_data(
             self.apikey, 'fork_repo',
             repoid=source_name,
             fork_name=fork_name,
             owner=owner)
         response = api_call(self.app, params)
         expected = 'user `%s` does not exist' % (owner,)
         assert_error(id_, expected, given=response.body)
 
     def test_api_fork_repo_fork_exists(self, backend):
         source_name = backend['minimal'].repo_name
         fork_name = backend.new_repo_name()
         fork_repo = fixture.create_fork(source_name, fork_name)
 
         id_, params = build_data(
             self.apikey, 'fork_repo',
             repoid=source_name,
             fork_name=fork_name,
             owner=TEST_USER_ADMIN_LOGIN)
         response = api_call(self.app, params)
 
         try:
             expected = {
                 'unique_repo_name': 'Repository with name `{}` already exists'.format(
                     fork_name)}
             assert_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(fork_repo.repo_name)
 
     def test_api_fork_repo_repo_exists(self, backend):
         source_name = backend['minimal'].repo_name
         fork_name = source_name
 
         id_, params = build_data(
             self.apikey, 'fork_repo',
             repoid=source_name,
             fork_name=fork_name,
             owner=TEST_USER_ADMIN_LOGIN)
         response = api_call(self.app, params)
 
         expected = {
             'unique_repo_name': 'Repository with name `{}` already exists'.format(
                 fork_name)}
         assert_error(id_, expected, given=response.body)
 
     @mock.patch.object(RepoModel, 'create_fork', crash)
     def test_api_fork_repo_exception_occurred(self, backend):
         source_name = backend['minimal'].repo_name
         fork_name = backend.new_repo_name()
         id_, params = build_data(
             self.apikey, 'fork_repo',
             repoid=source_name,
             fork_name=fork_name,
             owner=TEST_USER_ADMIN_LOGIN)
         response = api_call(self.app, params)
 
         expected = 'failed to fork repository `%s` as `%s`' % (source_name,
                                                                fork_name)
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_fts_search.py

@@ -1,115 +1,114 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 from rhodecode.tests import HG_REPO
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_error, assert_ok)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestApiSearch(object):
 
     @pytest.mark.parametrize("sort_dir", [
         "asc",
         "desc",
     ])
     @pytest.mark.parametrize("sort", [
         "xxx",
         "author_email",
         "date",
         "message",
     ])
     @pytest.mark.parametrize("query, expected_hits, expected_paths", [
         ('todo', 23, [
             'vcs/backends/hg/inmemory.py',
             'vcs/tests/test_git.py']),
         ('extension:rst installation', 6, [
             'docs/index.rst',
             'docs/installation.rst']),
         ('def repo', 87, [
             'vcs/tests/test_git.py',
             'vcs/tests/test_changesets.py']),
         ('repository:%s def test' % HG_REPO, 18, [
             'vcs/tests/test_git.py',
             'vcs/tests/test_changesets.py']),
         ('"def main"', 9, [
             'vcs/__init__.py',
             'vcs/tests/__init__.py',
             'vcs/utils/progressbar.py']),
         ('owner:test_admin', 358, [
             'vcs/tests/base.py',
             'MANIFEST.in',
             'vcs/utils/termcolors.py',
             'docs/theme/ADC/static/documentation.png']),
         ('owner:test_admin def main', 72, [
             'vcs/__init__.py',
             'vcs/tests/test_utils_filesize.py',
             'vcs/tests/test_cli.py']),
         ('owner:michał test', 0, []),
     ])
     def test_search_content_results(self, sort_dir, sort, query, expected_hits, expected_paths):
         id_, params = build_data(
             self.apikey_regular, 'search',
             search_query=query,
             search_sort='{}:{}'.format(sort_dir, sort),
             search_type='content')
 
         response = api_call(self.app, params)
         json_response = response.json
 
         assert json_response['result']['item_count'] == expected_hits
         paths = [x['f_path'] for x in json_response['result']['results']]
 
         for expected_path in expected_paths:
             assert expected_path in paths
 
     @pytest.mark.parametrize("sort_dir", [
         "asc",
         "desc",
     ])
     @pytest.mark.parametrize("sort", [
         "xxx",
         "date",
         "file",
         "size",
     ])
     @pytest.mark.parametrize("query, expected_hits, expected_paths", [
         ('readme.rst', 3, []),
         ('test*', 75, []),
         ('*model*', 1, []),
         ('extension:rst', 48, []),
         ('extension:rst api', 24, []),
     ])
     def test_search_file_paths(self, sort_dir, sort, query, expected_hits, expected_paths):
         id_, params = build_data(
             self.apikey_regular, 'search',
             search_query=query,
             search_sort='{}:{}'.format(sort_dir, sort),
             search_type='path')
 
         response = api_call(self.app, params)
         json_response = response.json
 
         assert json_response['result']['item_count'] == expected_hits
         paths = [x['f_path'] for x in json_response['result']['results']]
 
         for expected_path in expected_paths:
             assert expected_path in paths

rhodecode/api/tests/test_get_gist.py

@@ -1,101 +1,101 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 
+from rhodecode.lib.str_utils import safe_bytes
 from rhodecode.model.db import Gist
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_error, assert_ok)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestApiGetGist(object):
     def test_api_get_gist(self, gist_util, http_host_only_stub):
         gist = gist_util.create_gist()
         gist_id = gist.gist_access_id
         gist_created_on = gist.created_on
         gist_modified_at = gist.modified_at
         id_, params = build_data(
             self.apikey, 'get_gist', gistid=gist_id, )
         response = api_call(self.app, params)
 
         expected = {
             'access_id': gist_id,
             'created_on': gist_created_on,
             'modified_at': gist_modified_at,
             'description': 'new-gist',
             'expires': -1.0,
             'gist_id': int(gist_id),
             'type': 'public',
             'url': 'http://%s/_admin/gists/%s' % (http_host_only_stub, gist_id,),
             'acl_level': Gist.ACL_LEVEL_PUBLIC,
             'content': None,
         }
 
         assert_ok(id_, expected, given=response.body)
 
     def test_api_get_gist_with_content(self, gist_util, http_host_only_stub):
         mapping = {
-            u'filename1.txt': {'content': u'hello world'},
-            u'filename1ą.txt': {'content': u'hello worldę'}
+            b'filename1.txt': {'content': b'hello world'},
+            safe_bytes('filename1ą.txt'): {'content': safe_bytes('hello worldę')}
         }
         gist = gist_util.create_gist(gist_mapping=mapping)
         gist_id = gist.gist_access_id
         gist_created_on = gist.created_on
         gist_modified_at = gist.modified_at
         id_, params = build_data(
             self.apikey, 'get_gist', gistid=gist_id, content=True)
         response = api_call(self.app, params)
 
         expected = {
             'access_id': gist_id,
             'created_on': gist_created_on,
             'modified_at': gist_modified_at,
             'description': 'new-gist',
             'expires': -1.0,
             'gist_id': int(gist_id),
             'type': 'public',
             'url': 'http://%s/_admin/gists/%s' % (http_host_only_stub, gist_id,),
             'acl_level': Gist.ACL_LEVEL_PUBLIC,
             'content': {
                 u'filename1.txt': u'hello world',
                 u'filename1ą.txt': u'hello worldę'
             },
         }
 
         assert_ok(id_, expected, given=response.body)
 
     def test_api_get_gist_not_existing(self):
         id_, params = build_data(
             self.apikey_regular, 'get_gist', gistid='12345', )
         response = api_call(self.app, params)
         expected = 'gist `%s` does not exist' % ('12345',)
         assert_error(id_, expected, given=response.body)
 
     def test_api_get_gist_private_gist_without_permission(self, gist_util):
         gist = gist_util.create_gist()
         gist_id = gist.gist_access_id
         id_, params = build_data(
             self.apikey_regular, 'get_gist', gistid=gist_id, )
         response = api_call(self.app, params)
 
         expected = 'gist `%s` does not exist' % (gist_id,)
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_get_gists.py

@@ -1,74 +1,73 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 
 from rhodecode.tests import TEST_USER_ADMIN_LOGIN
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_error)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestApiGetGist(object):
     def test_api_get_gists(self, gist_util):
         gist_util.create_gist()
         gist_util.create_gist()
 
         id_, params = build_data(self.apikey, 'get_gists')
         response = api_call(self.app, params)
         assert len(response.json['result']) == 2
 
     def test_api_get_gists_regular_user(self, gist_util):
         # by admin
         gist_util.create_gist()
         gist_util.create_gist()
 
         # by reg user
         gist_util.create_gist(owner=self.TEST_USER_LOGIN)
         gist_util.create_gist(owner=self.TEST_USER_LOGIN)
         gist_util.create_gist(owner=self.TEST_USER_LOGIN)
 
         id_, params = build_data(self.apikey_regular, 'get_gists')
         response = api_call(self.app, params)
         assert len(response.json['result']) == 3
 
     def test_api_get_gists_only_for_regular_user(self, gist_util):
         # by admin
         gist_util.create_gist()
         gist_util.create_gist()
 
         # by reg user
         gist_util.create_gist(owner=self.TEST_USER_LOGIN)
         gist_util.create_gist(owner=self.TEST_USER_LOGIN)
         gist_util.create_gist(owner=self.TEST_USER_LOGIN)
 
         id_, params = build_data(
             self.apikey, 'get_gists', userid=self.TEST_USER_LOGIN)
         response = api_call(self.app, params)
         assert len(response.json['result']) == 3
 
     def test_api_get_gists_regular_user_with_different_userid(self):
         id_, params = build_data(
             self.apikey_regular, 'get_gists',
             userid=TEST_USER_ADMIN_LOGIN)
         response = api_call(self.app, params)
         expected = 'userid is not the same as your user'
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_get_ip.py

@@ -1,36 +1,35 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 
 from rhodecode.api.tests.utils import build_data, api_call, assert_ok
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestGetIp(object):
     def test_api_get_ip(self):
         id_, params = build_data(self.apikey, 'get_ip')
         response = api_call(self.app, params)
         expected = {
             'server_ip_addr': '0.0.0.0',
             'user_ips': []
         }
         assert_ok(id_, expected, given=response.body)

rhodecode/api/tests/test_get_locks.py

@@ -1,91 +1,90 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 
 from rhodecode.model.db import Repository, User
 from rhodecode.tests import TEST_USER_ADMIN_LOGIN, TEST_USER_REGULAR_LOGIN
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_ok, assert_error)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestGetLocks(object):
     def test_api_get_user_locks_regular_user(self):
         id_, params = build_data(self.apikey_regular, 'get_user_locks')
         response = api_call(self.app, params)
         expected = []
         assert_ok(id_, expected, given=response.body)
 
     def test_api_get_user_locks_with_userid_regular_user(self):
         id_, params = build_data(
             self.apikey_regular, 'get_user_locks', userid=TEST_USER_ADMIN_LOGIN)
         response = api_call(self.app, params)
         expected = 'userid is not the same as your user'
         assert_error(id_, expected, given=response.body)
 
     def test_api_get_user_locks(self):
         id_, params = build_data(self.apikey, 'get_user_locks')
         response = api_call(self.app, params)
         expected = []
         assert_ok(id_, expected, given=response.body)
 
     @pytest.mark.parametrize("apikey_attr, expect_secrets", [
         ('apikey', True),
         ('apikey_regular', False),
     ])
     def test_api_get_user_locks_with_one_locked_repo(
             self, apikey_attr, expect_secrets, backend):
 
         repo = backend.create_repo(cur_user=self.TEST_USER_LOGIN)
         Repository.lock(
             repo, User.get_by_username(self.TEST_USER_LOGIN).user_id)
 
         apikey = getattr(self, apikey_attr)
 
         id_, params = build_data(apikey, 'get_user_locks')
         if apikey_attr == 'apikey':
             # super-admin should call in specific user
             id_, params = build_data(apikey, 'get_user_locks',
                                      userid=self.TEST_USER_LOGIN)
 
         response = api_call(self.app, params)
         expected = [repo.get_api_data(include_secrets=expect_secrets)]
         assert_ok(id_, expected, given=response.body)
 
     def test_api_get_user_locks_with_one_locked_repo_for_specific_user(
             self, backend):
         repo = backend.create_repo(cur_user=self.TEST_USER_LOGIN)
 
         Repository.lock(repo, User.get_by_username(
             self.TEST_USER_LOGIN).user_id)
         id_, params = build_data(
             self.apikey, 'get_user_locks', userid=self.TEST_USER_LOGIN)
         response = api_call(self.app, params)
         expected = [repo.get_api_data(include_secrets=True)]
         assert_ok(id_, expected, given=response.body)
 
     def test_api_get_user_locks_with_userid(self):
         id_, params = build_data(
             self.apikey, 'get_user_locks', userid=TEST_USER_REGULAR_LOGIN)
         response = api_call(self.app, params)
         expected = []
         assert_ok(id_, expected, given=response.body)

rhodecode/api/tests/test_get_method.py

@@ -1,63 +1,62 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 
 from rhodecode.api.tests.utils import build_data, api_call, assert_ok
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestGetMethod(object):
     def test_get_methods_no_matches(self):
         id_, params = build_data(self.apikey, 'get_method', pattern='hello')
         response = api_call(self.app, params)
 
         expected = []
         assert_ok(id_, expected, given=response.body)
 
     def test_get_methods(self):
         id_, params = build_data(self.apikey, 'get_method', pattern='*comment*')
         response = api_call(self.app, params)
 
         expected = [
             'changeset_comment', 'comment_pull_request', 'get_pull_request_comments',
             'comment_commit', 'edit_comment', 'get_comment', 'get_repo_comments'
         ]
         assert_ok(id_, expected, given=response.body)
 
     def test_get_methods_on_single_match(self):
         id_, params = build_data(self.apikey, 'get_method',
                                  pattern='*comment_commit*')
         response = api_call(self.app, params)
 
         expected = ['comment_commit',
                     {'apiuser': '<RequiredType>',
-                     'comment_type': "<Optional:u'note'>",
+                     'comment_type': "<Optional:'note'>",
                      'commit_id': '<RequiredType>',
                      'extra_recipients': '<Optional:[]>',
                      'message': '<RequiredType>',
                      'repoid': '<RequiredType>',
                      'request': '<RequiredType>',
                      'resolves_comment_id': '<Optional:None>',
                      'status': '<Optional:None>',
                      'userid': '<Optional:<OptionalAttr:apiuser>>',
                      'send_email': '<Optional:True>'}]
         assert_ok(id_, expected, given=response.body)

rhodecode/api/tests/test_get_pull_request.py

@@ -1,143 +1,143 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 import urlobject
 
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_error, assert_ok)
 from rhodecode.lib import helpers as h
-from rhodecode.lib.utils2 import safe_unicode
+from rhodecode.lib.str_utils import safe_str
+
 
 pytestmark = pytest.mark.backends("git", "hg")
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestGetPullRequest(object):
 
     def test_api_get_pull_request(self, pr_util, http_host_only_stub):
         from rhodecode.model.pull_request import PullRequestModel
         pull_request = pr_util.create_pull_request(mergeable=True)
         id_, params = build_data(
             self.apikey, 'get_pull_request',
             pullrequestid=pull_request.pull_request_id, merge_state=True)
 
         response = api_call(self.app, params)
 
         assert response.status == '200 OK'
 
         url_obj = urlobject.URLObject(
             h.route_url(
                 'pullrequest_show',
                 repo_name=pull_request.target_repo.repo_name,
                 pull_request_id=pull_request.pull_request_id))
 
-        pr_url = safe_unicode(
+        pr_url = safe_str(
             url_obj.with_netloc(http_host_only_stub))
-        source_url = safe_unicode(
+        source_url = safe_str(
             pull_request.source_repo.clone_url().with_netloc(http_host_only_stub))
-        target_url = safe_unicode(
+        target_url = safe_str(
             pull_request.target_repo.clone_url().with_netloc(http_host_only_stub))
-        shadow_url = safe_unicode(
+        shadow_url = safe_str(
             PullRequestModel().get_shadow_clone_url(pull_request))
 
         expected = {
             'pull_request_id': pull_request.pull_request_id,
             'url': pr_url,
             'title': pull_request.title,
             'description': pull_request.description,
             'status': pull_request.status,
             'state': pull_request.pull_request_state,
             'created_on': pull_request.created_on,
             'updated_on': pull_request.updated_on,
             'commit_ids': pull_request.revisions,
             'review_status': pull_request.calculated_review_status(),
             'mergeable': {
                 'status': True,
                 'message': 'This pull request can be automatically merged.',
             },
             'source': {
                 'clone_url': source_url,
                 'repository': pull_request.source_repo.repo_name,
                 'reference': {
                     'name': pull_request.source_ref_parts.name,
                     'type': pull_request.source_ref_parts.type,
                     'commit_id': pull_request.source_ref_parts.commit_id,
                 },
             },
             'target': {
                 'clone_url': target_url,
                 'repository': pull_request.target_repo.repo_name,
                 'reference': {
                     'name': pull_request.target_ref_parts.name,
                     'type': pull_request.target_ref_parts.type,
                     'commit_id': pull_request.target_ref_parts.commit_id,
                 },
             },
             'merge': {
                 'clone_url': shadow_url,
                 'reference': {
                     'name': pull_request.shadow_merge_ref.name,
                     'type': pull_request.shadow_merge_ref.type,
                     'commit_id': pull_request.shadow_merge_ref.commit_id,
                 },
             },
             'author': pull_request.author.get_api_data(include_secrets=False,
                                                        details='basic'),
             'reviewers': [
                 {
                     'user': reviewer.get_api_data(include_secrets=False,
                                                   details='basic'),
                     'reasons': reasons,
                     'review_status': st[0][1].status if st else 'not_reviewed',
                 }
                 for obj, reviewer, reasons, mandatory, st in
                 pull_request.reviewers_statuses()
             ]
         }
         assert_ok(id_, expected, response.body)
 
     def test_api_get_pull_request_repo_error(self, pr_util):
         pull_request = pr_util.create_pull_request()
         id_, params = build_data(
             self.apikey, 'get_pull_request',
             repoid=666, pullrequestid=pull_request.pull_request_id)
         response = api_call(self.app, params)
 
         expected = 'repository `666` does not exist'
         assert_error(id_, expected, given=response.body)
 
     def test_api_get_pull_request_pull_request_error(self):
         id_, params = build_data(
             self.apikey, 'get_pull_request', pullrequestid=666)
         response = api_call(self.app, params)
 
         expected = 'pull request `666` does not exist'
         assert_error(id_, expected, given=response.body)
 
     def test_api_get_pull_request_pull_request_error_just_pr_id(self):
         id_, params = build_data(
             self.apikey, 'get_pull_request',
             pullrequestid=666)
         response = api_call(self.app, params)
 
         expected = 'pull request `666` does not exist'
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_get_pull_request_comments.py

@@ -1,87 +1,83 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
-import urlobject
 
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_error, assert_ok)
-from rhodecode.lib import helpers as h
-from rhodecode.lib.utils2 import safe_unicode
 
 pytestmark = pytest.mark.backends("git", "hg")
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestGetPullRequestComments(object):
 
     def test_api_get_pull_request_comments(self, pr_util, http_host_only_stub):
         from rhodecode.model.pull_request import PullRequestModel
 
         pull_request = pr_util.create_pull_request(mergeable=True)
         id_, params = build_data(
             self.apikey, 'get_pull_request_comments',
             pullrequestid=pull_request.pull_request_id)
 
         response = api_call(self.app, params)
 
         assert response.status == '200 OK'
         resp_date = response.json['result'][0]['comment_created_on']
         resp_comment_id = response.json['result'][0]['comment_id']
 
         expected = [
             {'comment_author': {'active': True,
                                 'full_name_or_username': 'RhodeCode Admin',
                                 'username': 'test_admin'},
              'comment_created_on': resp_date,
              'comment_f_path': None,
              'comment_id': resp_comment_id,
              'comment_lineno': None,
              'comment_status': {'status': 'under_review',
                                 'status_lbl': 'Under Review'},
              'comment_text': 'Auto status change to |new_status|\n\n.. |new_status| replace:: *"Under Review"*',
              'comment_type': 'note',
              'comment_resolved_by': None,
              'pull_request_version': None,
              'comment_last_version': 0,
              'comment_commit_id': None,
              'comment_pull_request_id': pull_request.pull_request_id
              }
         ]
         assert_ok(id_, expected, response.body)
 
     def test_api_get_pull_request_comments_repo_error(self, pr_util):
         pull_request = pr_util.create_pull_request()
         id_, params = build_data(
             self.apikey, 'get_pull_request_comments',
             repoid=666, pullrequestid=pull_request.pull_request_id)
         response = api_call(self.app, params)
 
         expected = 'repository `666` does not exist'
         assert_error(id_, expected, given=response.body)
 
     def test_api_get_pull_request_comments_pull_request_error(self):
         id_, params = build_data(
             self.apikey, 'get_pull_request_comments', pullrequestid=666)
         response = api_call(self.app, params)
 
         expected = 'pull request `666` does not exist'
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_get_pull_requests.py

@@ -1,82 +1,81 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 
 from rhodecode.model.meta import Session
 from rhodecode.model.pull_request import PullRequestModel
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_error)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestGetPullRequest(object):
 
     @pytest.mark.backends("git", "hg")
     def test_api_get_pull_requests(self, pr_util):
         pull_request = pr_util.create_pull_request()
         pull_request_2 = PullRequestModel().create(
             created_by=pull_request.author,
             source_repo=pull_request.source_repo,
             source_ref=pull_request.source_ref,
             target_repo=pull_request.target_repo,
             target_ref=pull_request.target_ref,
             revisions=pull_request.revisions,
             reviewers=(),
             observers=(),
             title=pull_request.title,
             description=pull_request.description,
         )
         Session().commit()
         id_, params = build_data(
             self.apikey, 'get_pull_requests',
             repoid=pull_request.target_repo.repo_name)
         response = api_call(self.app, params)
         assert response.status == '200 OK'
         assert len(response.json['result']) == 2
 
         PullRequestModel().close_pull_request(
             pull_request_2, pull_request_2.author)
         Session().commit()
 
         id_, params = build_data(
             self.apikey, 'get_pull_requests',
             repoid=pull_request.target_repo.repo_name,
             status='new')
         response = api_call(self.app, params)
         assert response.status == '200 OK'
         assert len(response.json['result']) == 1
 
         id_, params = build_data(
             self.apikey, 'get_pull_requests',
             repoid=pull_request.target_repo.repo_name,
             status='closed')
         response = api_call(self.app, params)
         assert response.status == '200 OK'
         assert len(response.json['result']) == 1
 
     @pytest.mark.backends("git", "hg")
     def test_api_get_pull_requests_repo_error(self):
         id_, params = build_data(self.apikey, 'get_pull_requests', repoid=666)
         response = api_call(self.app, params)
 
         expected = 'repository `666` does not exist'
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_get_repo.py

@@ -1,143 +1,142 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 
 from rhodecode.model.meta import Session
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.user import UserModel
 from rhodecode.tests import TEST_USER_ADMIN_LOGIN
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_ok, assert_error, expected_permissions)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestGetRepo(object):
     @pytest.mark.parametrize("apikey_attr, expect_secrets", [
         ('apikey', True),
         ('apikey_regular', False),
     ])
     @pytest.mark.parametrize("cache_param", [
         True,
         False,
         None,
     ])
     def test_api_get_repo(
             self, apikey_attr, expect_secrets, cache_param, backend,
             user_util):
         repo = backend.create_repo()
         repo_id = repo.repo_id
         usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
         group = user_util.create_user_group(members=[usr])
         user_util.grant_user_group_permission_to_repo(
             repo=repo, user_group=group, permission_name='repository.read')
         Session().commit()
         kwargs = {
             'repoid': repo.repo_name,
         }
         if cache_param is not None:
             kwargs['cache'] = cache_param
 
         apikey = getattr(self, apikey_attr)
         id_, params = build_data(apikey, 'get_repo', **kwargs)
         response = api_call(self.app, params)
 
         ret = repo.get_api_data()
 
         permissions = expected_permissions(repo)
 
         followers = []
 
         repo = RepoModel().get(repo_id)
         for user in repo.followers:
             followers.append(user.user.get_api_data(
                 include_secrets=expect_secrets))
 
         ret['permissions'] = permissions
         ret['followers'] = followers
 
         expected = ret
 
         assert_ok(id_, expected, given=response.body)
 
     @pytest.mark.parametrize("grant_perm", [
         'repository.admin',
         'repository.write',
         'repository.read',
     ])
     def test_api_get_repo_by_non_admin(self, grant_perm, backend):
         # TODO: Depending on which tests are running before this one, we
         # start with a different number of permissions in the database.
         repo = RepoModel().get_by_repo_name(backend.repo_name)
         repo_id = repo.repo_id
         permission_count = len(repo.repo_to_perm)
 
         RepoModel().grant_user_permission(repo=backend.repo_name,
                                           user=self.TEST_USER_LOGIN,
                                           perm=grant_perm)
         Session().commit()
         id_, params = build_data(
             self.apikey_regular, 'get_repo', repoid=backend.repo_name)
         response = api_call(self.app, params)
 
         repo = RepoModel().get_by_repo_name(backend.repo_name)
         ret = repo.get_api_data()
 
         assert permission_count + 1, len(repo.repo_to_perm)
 
         permissions = expected_permissions(repo)
 
         followers = []
 
         repo = RepoModel().get(repo_id)
         for user in repo.followers:
             followers.append(user.user.get_api_data())
 
         ret['permissions'] = permissions
         ret['followers'] = followers
 
         expected = ret
         try:
             assert_ok(id_, expected, given=response.body)
         finally:
             RepoModel().revoke_user_permission(
                 backend.repo_name, self.TEST_USER_LOGIN)
 
     def test_api_get_repo_by_non_admin_no_permission_to_repo(self, backend):
         RepoModel().grant_user_permission(repo=backend.repo_name,
                                           user=self.TEST_USER_LOGIN,
                                           perm='repository.none')
 
         id_, params = build_data(
             self.apikey_regular, 'get_repo', repoid=backend.repo_name)
         response = api_call(self.app, params)
 
         expected = 'repository `%s` does not exist' % (backend.repo_name)
         assert_error(id_, expected, given=response.body)
 
     def test_api_get_repo_not_existing(self):
         id_, params = build_data(
             self.apikey, 'get_repo', repoid='no-such-repo')
         response = api_call(self.app, params)
 
         ret = 'repository `%s` does not exist' % 'no-such-repo'
         expected = ret
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_get_repo_changeset.py

@@ -1,141 +1,140 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 
 from rhodecode.api.tests.utils import build_data, api_call, assert_error
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestGetRepoChangeset(object):
     @pytest.mark.parametrize("details", ['basic', 'extended', 'full'])
     def test_get_repo_changeset(self, details, backend):
         commit = backend.repo.get_commit(commit_idx=0)
         __, params = build_data(
             self.apikey, 'get_repo_changeset',
             repoid=backend.repo_name, revision=commit.raw_id,
             details=details,
         )
         response = api_call(self.app, params)
         result = response.json['result']
         assert result['revision'] == 0
         assert result['raw_id'] == commit.raw_id
 
         if details == 'full':
             assert result['refs']['bookmarks'] == getattr(
                 commit, 'bookmarks', [])
             branches = [commit.branch] if commit.branch else []
             assert result['refs']['branches'] == branches
             assert result['refs']['tags'] == commit.tags
 
     @pytest.mark.parametrize("details", ['basic', 'extended', 'full'])
     def test_get_repo_changeset_bad_type(self, details, backend):
         id_, params = build_data(
             self.apikey, 'get_repo_changeset',
             repoid=backend.repo_name, revision=0,
             details=details,
         )
         response = api_call(self.app, params)
-        expected = "commit_id must be a string value got <type 'int'> instead"
+        expected = "commit_id must be a string value got <class 'int'> instead"
         assert_error(id_, expected, given=response.body)
 
     @pytest.mark.parametrize("details", ['basic', 'extended', 'full'])
     def test_get_repo_changesets(self, details, backend):
         limit = 2
         commit = backend.repo.get_commit(commit_idx=0)
         __, params = build_data(
             self.apikey, 'get_repo_changesets',
             repoid=backend.repo_name, start_rev=commit.raw_id, limit=limit,
             details=details,
         )
         response = api_call(self.app, params)
         result = response.json['result']
         assert result
         assert len(result) == limit
         for x in range(limit):
             assert result[x]['revision'] == x
 
         if details == 'full':
             for x in range(limit):
                 assert 'bookmarks' in result[x]['refs']
                 assert 'branches' in result[x]['refs']
                 assert 'tags' in result[x]['refs']
 
     @pytest.mark.parametrize("details", ['basic', 'extended', 'full'])
     @pytest.mark.parametrize("start_rev, expected_revision", [
         ("0", 0),
         ("10", 10),
         ("20", 20),
         ])
     @pytest.mark.backends("hg", "git")
     def test_get_repo_changesets_commit_range(
             self, details, backend, start_rev, expected_revision):
         limit = 10
         __, params = build_data(
             self.apikey, 'get_repo_changesets',
             repoid=backend.repo_name, start_rev=start_rev, limit=limit,
             details=details,
         )
         response = api_call(self.app, params)
         result = response.json['result']
         assert result
         assert len(result) == limit
         for i in range(limit):
             assert result[i]['revision'] == int(expected_revision) + i
 
     @pytest.mark.parametrize("details", ['basic', 'extended', 'full'])
     @pytest.mark.parametrize("start_rev, expected_revision", [
         ("0", 0),
         ("10", 9),
         ("20", 19),
         ])
     def test_get_repo_changesets_commit_range_svn(
             self, details, backend_svn, start_rev, expected_revision):
 
         # TODO: johbo: SVN showed a problem here: The parameter "start_rev"
         # in our API allows to pass in a "Commit ID" as well as a
         # "Commit Index". In the case of Subversion it is not possible to
         # distinguish these cases. As a workaround we implemented this
         # behavior which gives a preference to see it as a "Commit ID".
 
         limit = 10
         __, params = build_data(
             self.apikey, 'get_repo_changesets',
             repoid=backend_svn.repo_name, start_rev=start_rev, limit=limit,
             details=details,
         )
         response = api_call(self.app, params)
         result = response.json['result']
         assert result
         assert len(result) == limit
         for i in range(limit):
             assert result[i]['revision'] == int(expected_revision) + i
 
     @pytest.mark.parametrize("details", ['basic', 'extended', 'full'])
     def test_get_repo_changesets_bad_type(self, details, backend):
         id_, params = build_data(
             self.apikey, 'get_repo_changesets',
             repoid=backend.repo_name, start_rev=0, limit=2,
             details=details,
         )
         response = api_call(self.app, params)
-        expected = "commit_id must be a string value got <type 'int'> instead"
+        expected = "commit_id must be a string value got <class 'int'> instead"
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_get_repo_comments.py

@@ -1,142 +1,141 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 
 from rhodecode.model.db import User, ChangesetComment
 from rhodecode.model.meta import Session
 from rhodecode.model.comment import CommentsModel
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_error, assert_call_ok)
 
 
 @pytest.fixture()
 def make_repo_comments_factory(request):
 
     class Make(object):
 
         def make_comments(self, repo):
             user = User.get_first_super_admin()
             commit = repo.scm_instance()[0]
 
             commit_id = commit.raw_id
             file_0 = commit.affected_files[0]
             comments = []
 
             # general
             comment = CommentsModel().create(
                 text='General Comment', repo=repo, user=user, commit_id=commit_id,
                 comment_type=ChangesetComment.COMMENT_TYPE_NOTE, send_email=False)
             comments.append(comment)
 
             # inline
             comment = CommentsModel().create(
                 text='Inline Comment', repo=repo, user=user, commit_id=commit_id,
                 f_path=file_0, line_no='n1',
                 comment_type=ChangesetComment.COMMENT_TYPE_NOTE, send_email=False)
             comments.append(comment)
 
             # todo
             comment = CommentsModel().create(
                 text='INLINE TODO Comment', repo=repo, user=user, commit_id=commit_id,
                 f_path=file_0, line_no='n1',
                 comment_type=ChangesetComment.COMMENT_TYPE_TODO, send_email=False)
             comments.append(comment)
 
             return comments
 
     return Make()
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestGetRepo(object):
 
     @pytest.mark.parametrize('filters, expected_count', [
         ({}, 3),
         ({'comment_type': ChangesetComment.COMMENT_TYPE_NOTE}, 2),
         ({'comment_type': ChangesetComment.COMMENT_TYPE_TODO}, 1),
         ({'commit_id': 'FILLED DYNAMIC'}, 3),
     ])
     def test_api_get_repo_comments(self, backend, user_util,
                                    make_repo_comments_factory, filters, expected_count):
         commits = [{'message': 'A'}, {'message': 'B'}]
         repo = backend.create_repo(commits=commits)
         make_repo_comments_factory.make_comments(repo)
 
         api_call_params = {'repoid': repo.repo_name,}
         api_call_params.update(filters)
 
         if 'commit_id' in api_call_params:
             commit = repo.scm_instance()[0]
             commit_id = commit.raw_id
             api_call_params['commit_id'] = commit_id
 
         id_, params = build_data(self.apikey, 'get_repo_comments', **api_call_params)
         response = api_call(self.app, params)
         result = assert_call_ok(id_, given=response.body)
 
         assert len(result) == expected_count
 
     def test_api_get_repo_comments_wrong_comment_type(
             self, make_repo_comments_factory, backend_hg):
         commits = [{'message': 'A'}, {'message': 'B'}]
         repo = backend_hg.create_repo(commits=commits)
         make_repo_comments_factory.make_comments(repo)
 
         api_call_params = {'repoid': repo.repo_name}
         api_call_params.update({'comment_type': 'bogus'})
 
         expected = 'comment_type must be one of `{}` got {}'.format(
                     ChangesetComment.COMMENT_TYPES, 'bogus')
         id_, params = build_data(self.apikey, 'get_repo_comments', **api_call_params)
         response = api_call(self.app, params)
         assert_error(id_, expected, given=response.body)
 
     def test_api_get_comment(self, make_repo_comments_factory, backend_hg):
         commits = [{'message': 'A'}, {'message': 'B'}]
         repo = backend_hg.create_repo(commits=commits)
 
         comments = make_repo_comments_factory.make_comments(repo)
         comment_ids = [x.comment_id for x in comments]
         Session().commit()
 
         for comment_id in comment_ids:
             id_, params = build_data(self.apikey, 'get_comment',
                                      **{'comment_id': comment_id})
             response = api_call(self.app, params)
             result = assert_call_ok(id_, given=response.body)
             assert result['comment_id'] == comment_id
 
     def test_api_get_comment_no_access(self, make_repo_comments_factory, backend_hg, user_util):
         commits = [{'message': 'A'}, {'message': 'B'}]
         repo = backend_hg.create_repo(commits=commits)
         comments = make_repo_comments_factory.make_comments(repo)
         comment_id = comments[0].comment_id
 
         test_user = user_util.create_user()
         user_util.grant_user_permission_to_repo(repo, test_user, 'repository.none')
 
         id_, params = build_data(test_user.api_key, 'get_comment',
                                  **{'comment_id': comment_id})
         response = api_call(self.app, params)
         assert_error(id_,
                      expected='comment `{}` does not exist'.format(comment_id),
                      given=response.body)

rhodecode/api/tests/test_get_repo_group.py

@@ -1,55 +1,54 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 
 from rhodecode.model.repo_group import RepoGroupModel
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_ok, assert_error, expected_permissions)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestApiGetRepoGroup(object):
     def test_api_get_repo_group(self, user_util):
         repo_group = user_util.create_repo_group()
         repo_group_name = repo_group.group_name
 
         id_, params = build_data(
             self.apikey, 'get_repo_group', repogroupid=repo_group_name)
         response = api_call(self.app, params)
 
         repo_group = RepoGroupModel()._get_repo_group(repo_group_name)
         ret = repo_group.get_api_data()
 
         permissions = expected_permissions(repo_group)
 
         ret['permissions'] = permissions
         expected = ret
         assert_ok(id_, expected, given=response.body)
 
     def test_api_get_repo_group_not_existing(self):
         id_, params = build_data(
             self.apikey, 'get_repo_group', repogroupid='no-such-repo-group')
         response = api_call(self.app, params)
 
         ret = 'repository group `%s` does not exist' % 'no-such-repo-group'
         expected = ret
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_get_repo_groups.py

@@ -1,40 +1,39 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 
 from rhodecode.model.repo_group import RepoGroupModel
 from rhodecode.api.tests.utils import build_data, api_call, assert_ok, jsonify
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestApiGetRepoGroups(object):
     def test_api_get_repo_groups(self):
         id_, params = build_data(self.apikey, 'get_repo_groups')
         response = api_call(self.app, params)
 
         result = []
         for repo in RepoGroupModel().get_all():
             result.append(repo.get_api_data())
         ret = jsonify(result)
 
         expected = ret
         assert_ok(id_, expected, given=response.body)

rhodecode/api/tests/test_get_repo_nodes.py

@@ -1,142 +1,141 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 
 from rhodecode.model.meta import Session
 from rhodecode.model.repo import RepoModel
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_error, assert_ok)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestGetRepoNodes(object):
     @pytest.mark.parametrize("name, ret_type", [
         ('all', 'all'),
         ('dirs', 'dirs'),
         ('files', 'files'),
     ])
     def test_api_get_repo_nodes(self, name, ret_type, backend):
         commit_id = 'tip'
         path = '/'
         id_, params = build_data(
             self.apikey, 'get_repo_nodes',
             repoid=backend.repo_name, revision=commit_id,
             root_path=path,
             ret_type=ret_type)
         response = api_call(self.app, params)
 
         # we don't the actual return types here since it's tested somewhere
         # else
         expected = response.json['result']
         assert_ok(id_, expected, given=response.body)
 
     def test_api_get_repo_nodes_bad_commits(self, backend):
         commit_id = 'i-dont-exist'
         path = '/'
         id_, params = build_data(
             self.apikey, 'get_repo_nodes',
             repoid=backend.repo_name, revision=commit_id,
             root_path=path, )
         response = api_call(self.app, params)
 
         expected = 'failed to get repo: `%s` nodes' % (backend.repo_name,)
         assert_error(id_, expected, given=response.body)
 
     def test_api_get_repo_nodes_bad_path(self, backend):
         commit_id = 'tip'
         path = '/idontexits'
         id_, params = build_data(
             self.apikey, 'get_repo_nodes',
             repoid=backend.repo_name, revision=commit_id,
             root_path=path, )
         response = api_call(self.app, params)
 
         expected = 'failed to get repo: `%s` nodes' % (backend.repo_name,)
         assert_error(id_, expected, given=response.body)
 
     def test_api_get_repo_nodes_max_file_bytes(self, backend):
         commit_id = 'tip'
         path = '/'
         max_file_bytes = 500
 
         id_, params = build_data(
             self.apikey, 'get_repo_nodes',
             repoid=backend.repo_name, revision=commit_id, details='full',
             root_path=path)
         response = api_call(self.app, params)
         assert any(file['content'] and len(file['content']) > max_file_bytes
                    for file in response.json['result'])
 
         id_, params = build_data(
             self.apikey, 'get_repo_nodes',
             repoid=backend.repo_name, revision=commit_id,
             root_path=path, details='full',
             max_file_bytes=max_file_bytes)
         response = api_call(self.app, params)
         assert all(
             file['content'] is None if file['size'] > max_file_bytes else True
             for file in response.json['result'])
 
     def test_api_get_repo_nodes_bad_ret_type(self, backend):
         commit_id = 'tip'
         path = '/'
         ret_type = 'error'
         id_, params = build_data(
             self.apikey, 'get_repo_nodes',
             repoid=backend.repo_name, revision=commit_id,
             root_path=path,
             ret_type=ret_type)
         response = api_call(self.app, params)
 
         expected = ('ret_type must be one of %s'
                     % (','.join(['all', 'dirs', 'files'])))
         assert_error(id_, expected, given=response.body)
 
     @pytest.mark.parametrize("name, ret_type, grant_perm", [
         ('all', 'all', 'repository.write'),
         ('dirs', 'dirs', 'repository.admin'),
         ('files', 'files', 'repository.read'),
     ])
     def test_api_get_repo_nodes_by_regular_user(
             self, name, ret_type, grant_perm, backend):
         RepoModel().grant_user_permission(repo=backend.repo_name,
                                           user=self.TEST_USER_LOGIN,
                                           perm=grant_perm)
         Session().commit()
 
         commit_id = 'tip'
         path = '/'
         id_, params = build_data(
             self.apikey_regular, 'get_repo_nodes',
             repoid=backend.repo_name, revision=commit_id,
             root_path=path,
             ret_type=ret_type)
         response = api_call(self.app, params)
 
         # we don't the actual return types here since it's tested somewhere
         # else
         expected = response.json['result']
         try:
             assert_ok(id_, expected, given=response.body)
         finally:
             RepoModel().revoke_user_permission(
                 backend.repo_name, self.TEST_USER_LOGIN)

rhodecode/api/tests/test_get_repo_refs.py

@@ -1,40 +1,39 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 
 from rhodecode.model.meta import Session
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.user import UserModel
 from rhodecode.tests import TEST_USER_ADMIN_LOGIN
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_ok, assert_error, expected_permissions)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestGetRepo(object):
     def test_api_get_repo_refs(self, backend, user_util):
         repo = backend.create_repo()
         id_, params = build_data(self.apikey, 'get_repo_refs',
                                  **{'repoid': repo.repo_name,})
         response = api_call(self.app, params)
         expected = repo.scm_instance().refs()
         assert_ok(id_, expected, given=response.body)

rhodecode/api/tests/test_get_repos.py

@@ -1,129 +1,128 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 
 from rhodecode.model.repo import RepoModel
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_ok, assert_error, jsonify)
 from rhodecode.model.db import User
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestGetRepos(object):
     def test_api_get_repos(self):
         id_, params = build_data(self.apikey, 'get_repos')
         response = api_call(self.app, params)
 
         result = []
         for repo in RepoModel().get_all():
             result.append(repo.get_api_data(include_secrets=True))
         ret = jsonify(result)
 
         expected = ret
         assert_ok(id_, expected, given=response.body)
 
     def test_api_get_repos_only_toplevel(self, user_util):
         repo_group = user_util.create_repo_group(auto_cleanup=True)
         user_util.create_repo(parent=repo_group)
 
         id_, params = build_data(self.apikey, 'get_repos', traverse=0)
         response = api_call(self.app, params)
 
         result = []
         for repo in RepoModel().get_repos_for_root(root=None):
             result.append(repo.get_api_data(include_secrets=True))
         expected = jsonify(result)
 
         assert_ok(id_, expected, given=response.body)
 
     def test_api_get_repos_with_wrong_root(self):
         id_, params = build_data(self.apikey, 'get_repos', root='abracadabra')
         response = api_call(self.app, params)
 
         expected = 'Root repository group `abracadabra` does not exist'
         assert_error(id_, expected, given=response.body)
 
     def test_api_get_repos_with_root(self, user_util):
         repo_group = user_util.create_repo_group(auto_cleanup=True)
         repo_group_name = repo_group.group_name
 
         user_util.create_repo(parent=repo_group)
         user_util.create_repo(parent=repo_group)
 
         # nested, should not show up
         user_util._test_name = '{}/'.format(repo_group_name)
         sub_repo_group = user_util.create_repo_group(auto_cleanup=True)
         user_util.create_repo(parent=sub_repo_group)
 
         id_, params = build_data(self.apikey, 'get_repos',
                                  root=repo_group_name, traverse=0)
         response = api_call(self.app, params)
 
         result = []
         for repo in RepoModel().get_repos_for_root(repo_group):
             result.append(repo.get_api_data(include_secrets=True))
 
         assert len(result) == 2
         expected = jsonify(result)
         assert_ok(id_, expected, given=response.body)
 
     def test_api_get_repos_with_root_and_traverse(self, user_util):
         repo_group = user_util.create_repo_group(auto_cleanup=True)
         repo_group_name = repo_group.group_name
 
         user_util.create_repo(parent=repo_group)
         user_util.create_repo(parent=repo_group)
 
         # nested, should not show up
         user_util._test_name = '{}/'.format(repo_group_name)
         sub_repo_group = user_util.create_repo_group(auto_cleanup=True)
         user_util.create_repo(parent=sub_repo_group)
 
         id_, params = build_data(self.apikey, 'get_repos',
                                  root=repo_group_name, traverse=1)
         response = api_call(self.app, params)
 
         result = []
         for repo in RepoModel().get_repos_for_root(
                 repo_group_name, traverse=True):
             result.append(repo.get_api_data(include_secrets=True))
 
         assert len(result) == 3
         expected = jsonify(result)
         assert_ok(id_, expected, given=response.body)
 
     def test_api_get_repos_non_admin(self):
         id_, params = build_data(self.apikey_regular, 'get_repos')
         response = api_call(self.app, params)
 
         user = User.get_by_username(self.TEST_USER_LOGIN)
         allowed_repos = user.AuthUser().permissions['repositories']
 
         result = []
         for repo in RepoModel().get_all():
             perm = allowed_repos[repo.repo_name]
             if perm in ['repository.read', 'repository.write', 'repository.admin']:
                 result.append(repo.get_api_data())
         ret = jsonify(result)
 
         expected = ret
         assert_ok(id_, expected, given=response.body)

rhodecode/api/tests/test_get_server_info.py

@@ -1,84 +1,83 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 
 from rhodecode.model.scm import ScmModel
 from rhodecode.api.tests.utils import build_data, api_call, assert_ok
 
 
 @pytest.fixture()
 def http_host_stub():
     """
     To ensure that we can get an IP address, this test shall run with a
     hostname set to "localhost".
     """
     return 'localhost:80'
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestGetServerInfo(object):
     def test_api_get_server_info(self):
         id_, params = build_data(self.apikey, 'get_server_info')
         response = api_call(self.app, params)
         resp = response.json
         expected = ScmModel().get_server_info()
         expected['memory'] = resp['result']['memory']
         expected['uptime'] = resp['result']['uptime']
         expected['load'] = resp['result']['load']
         expected['cpu'] = resp['result']['cpu']
         expected['storage'] = resp['result']['storage']
         expected['storage_temp'] = resp['result']['storage_temp']
         expected['storage_inodes'] = resp['result']['storage_inodes']
         expected['server'] = resp['result']['server']
 
         expected['index_storage'] = resp['result']['index_storage']
         expected['storage'] = resp['result']['storage']
 
         assert_ok(id_, expected, given=response.body)
 
     def test_api_get_server_info_ip(self):
         id_, params = build_data(self.apikey, 'get_server_info')
         response = api_call(self.app, params)
         resp = response.json
         expected = ScmModel().get_server_info({'SERVER_NAME': 'unknown'})
         expected['memory'] = resp['result']['memory']
         expected['uptime'] = resp['result']['uptime']
         expected['load'] = resp['result']['load']
         expected['cpu'] = resp['result']['cpu']
         expected['storage'] = resp['result']['storage']
         expected['storage_temp'] = resp['result']['storage_temp']
         expected['storage_inodes'] = resp['result']['storage_inodes']
         expected['server'] = resp['result']['server']
 
         expected['index_storage'] = resp['result']['index_storage']
         expected['storage'] = resp['result']['storage']
 
         assert_ok(id_, expected, given=response.body)
 
     def test_api_get_server_info_data_for_search_index_build(self):
         id_, params = build_data(self.apikey, 'get_server_info')
         response = api_call(self.app, params)
         resp = response.json
 
         # required by indexer
         assert resp['result']['index_storage']
         assert resp['result']['storage']

rhodecode/api/tests/test_get_user.py

@@ -1,86 +1,85 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 
 from rhodecode.lib.auth import AuthUser
 from rhodecode.model.user import UserModel
 from rhodecode.tests import TEST_USER_ADMIN_LOGIN
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_ok, assert_error)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestGetUser(object):
     def test_api_get_user(self):
         id_, params = build_data(
             self.apikey, 'get_user', userid=TEST_USER_ADMIN_LOGIN)
         response = api_call(self.app, params)
 
         usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
         ret = usr.get_api_data(include_secrets=True)
         permissions = AuthUser(usr.user_id).permissions
         ret['permissions'] = permissions
         ret['permissions_summary'] = permissions
 
         expected = ret
         assert_ok(id_, expected, given=response.body)
 
     def test_api_get_user_not_existing(self):
         id_, params = build_data(self.apikey, 'get_user', userid='trololo')
         response = api_call(self.app, params)
 
         expected = "user `%s` does not exist" % 'trololo'
         assert_error(id_, expected, given=response.body)
 
     def test_api_get_user_without_giving_userid(self):
         id_, params = build_data(self.apikey, 'get_user')
         response = api_call(self.app, params)
 
         usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
         ret = usr.get_api_data(include_secrets=True)
         permissions = AuthUser(usr.user_id).permissions
         ret['permissions'] = permissions
         ret['permissions_summary'] = permissions
 
         expected = ret
         assert_ok(id_, expected, given=response.body)
 
     def test_api_get_user_without_giving_userid_non_admin(self):
         id_, params = build_data(self.apikey_regular, 'get_user')
         response = api_call(self.app, params)
 
         usr = UserModel().get_by_username(self.TEST_USER_LOGIN)
         ret = usr.get_api_data(include_secrets=True)
         permissions = AuthUser(usr.user_id).permissions
         ret['permissions'] = permissions
         ret['permissions_summary'] = permissions
 
         expected = ret
         assert_ok(id_, expected, given=response.body)
 
     def test_api_get_user_with_giving_userid_non_admin(self):
         id_, params = build_data(
             self.apikey_regular, 'get_user',
             userid=self.TEST_USER_LOGIN)
         response = api_call(self.app, params)
 
         expected = 'userid is not the same as your user'
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_get_user_group.py

@@ -1,76 +1,75 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 
 from rhodecode.model.user import UserModel
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_ok, assert_error, expected_permissions)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestGetUserGroups(object):
     def test_api_get_user_group(self, user_util):
         user, group = user_util.create_user_with_group()
         id_, params = build_data(
             self.apikey, 'get_user_group', usergroupid=group.users_group_name)
         response = api_call(self.app, params)
 
         ret = group.get_api_data()
         ret['users'] = [user.get_api_data()]
 
         permissions = expected_permissions(group)
 
         ret['permissions'] = permissions
         ret['permissions_summary'] = response.json['result']['permissions_summary']
         expected = ret
         assert_ok(id_, expected, given=response.body)
 
     def test_api_get_user_group_regular_user(self, user_util):
         user, group = user_util.create_user_with_group()
         id_, params = build_data(
             self.apikey_regular, 'get_user_group',
             usergroupid=group.users_group_name)
         response = api_call(self.app, params)
 
         ret = group.get_api_data()
         ret['users'] = [user.get_api_data()]
 
         permissions = expected_permissions(group)
 
         ret['permissions'] = permissions
         ret['permissions_summary'] = response.json['result']['permissions_summary']
         expected = ret
         assert_ok(id_, expected, given=response.body)
 
     def test_api_get_user_group_regular_user_permission_denied(
             self, user_util):
         group = user_util.create_user_group()
         user = UserModel().get_by_username(self.TEST_USER_LOGIN)
         group_name = group.users_group_name
         user_util.grant_user_permission_to_user_group(
             group, user, 'usergroup.none')
 
         id_, params = build_data(
             self.apikey_regular, 'get_user_group', usergroupid=group_name)
         response = api_call(self.app, params)
 
         expected = 'user group `%s` does not exist' % (group_name,)
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_get_user_groups.py

@@ -1,71 +1,69 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 json
-
 import pytest
 
 from rhodecode.model.user import UserModel
 from rhodecode.api.tests.utils import build_data, api_call
+from rhodecode.lib.ext_json import json
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestGetUserGroups(object):
     @pytest.mark.parametrize("apikey_attr, expect_secrets", [
         ('apikey', True),
         ('apikey_regular', False),
     ])
     def test_api_get_user_groups(self, apikey_attr, expect_secrets, user_util):
         first_group = user_util.create_user_group()
         second_group = user_util.create_user_group()
         expected = [
             g.get_api_data(include_secrets=expect_secrets)
             for g in (first_group, second_group)]
 
         apikey = getattr(self, apikey_attr)
         id_, params = build_data(apikey, 'get_user_groups', )
         response = api_call(self.app, params)
         self._assert_ok(id_, expected, response)
 
     def test_api_get_user_groups_regular_user(self, user_util):
         first_group = user_util.create_user_group()
         second_group = user_util.create_user_group()
         expected = [g.get_api_data() for g in (first_group, second_group)]
 
         id_, params = build_data(self.apikey_regular, 'get_user_groups', )
         response = api_call(self.app, params)
         self._assert_ok(id_, expected, response)
 
     def test_api_get_user_groups_regular_user_no_permission(self, user_util):
         group = user_util.create_user_group()
         user = UserModel().get_by_username(self.TEST_USER_LOGIN)
         user_util.grant_user_permission_to_user_group(
             group, user, 'usergroup.none')
         id_, params = build_data(self.apikey_regular, 'get_user_groups', )
         response = api_call(self.app, params)
         expected = []
         self._assert_ok(id_, expected, response)
 
     def _assert_ok(self, id_, expected_list, response):
         result = json.loads(response.body)
         assert result['id'] == id_
         assert result['error'] is None
-        assert sorted(result['result']) == sorted(expected_list)
+        assert result['result'] == expected_list

rhodecode/api/tests/test_get_users.py

@@ -1,40 +1,39 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 
 from rhodecode.model.db import User
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_ok, jsonify)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestGetUsers(object):
     def test_api_get_users(self):
         id_, params = build_data(self.apikey, 'get_users', )
         response = api_call(self.app, params)
         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(include_secrets=True)
             ret_all.append(jsonify(ret))
         expected = ret_all
         assert_ok(id_, expected, given=response.body)

rhodecode/api/tests/test_grant_user_group_permission.py

@@ -1,90 +1,89 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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,)
         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,173 +1,172 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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,)
             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. 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_group_permission_to_user_group.py

@@ -1,97 +1,96 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 
 from rhodecode.model.user_group import UserGroupModel
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_ok, assert_error)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestGrantUserGroupPermissionFromUserGroup(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_group_permission_to_user_group(
             self, name, perm, user_util):
         group = user_util.create_user_group()
         target_group = user_util.create_user_group()
 
         id_, params = build_data(
             self.apikey,
             'grant_user_group_permission_to_user_group',
             usergroupid=target_group.users_group_name,
             sourceusergroupid=group.users_group_name,
             perm=perm)
         response = api_call(self.app, params)
 
         expected = {
             'msg': (
                 'Granted perm: `%s` for user group: `%s`'
                 ' in user group: `%s`' % (
                     perm, group.users_group_name,
                     target_group.users_group_name
                 )
             ),
             'success': True
         }
         try:
             assert_ok(id_, expected, given=response.body)
         finally:
             UserGroupModel().revoke_user_group_permission(
                 target_group.users_group_id, group.users_group_id)
 
     def test_api_grant_user_group_permission_to_user_group_same_failure(
             self, user_util):
         group = user_util.create_user_group()
 
         id_, params = build_data(
             self.apikey,
             'grant_user_group_permission_to_user_group',
             usergroupid=group.users_group_name,
             sourceusergroupid=group.users_group_name,
             perm='usergroup.none')
         response = api_call(self.app, params)
 
         expected = (
             'failed to edit permission for user group: `%s`'
             ' in user group: `%s`' % (
                 group.users_group_name, group.users_group_name)
         )
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_grant_user_permission.py

@@ -1,87 +1,86 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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,)
         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,156 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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. 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,155 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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. 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/tests/test_invalidate_cache.py

@@ -1,68 +1,67 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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.scm import ScmModel
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_ok, assert_error, crash)
 from rhodecode.model.repo import RepoModel
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestInvalidateCache(object):
 
     def _set_cache(self, repo_name):
         repo = RepoModel().get_by_repo_name(repo_name)
         repo.scm_instance(cache=True)
 
     def test_api_invalidate_cache(self, backend):
         self._set_cache(backend.repo_name)
 
         id_, params = build_data(
             self.apikey, 'invalidate_cache', repoid=backend.repo_name)
         response = api_call(self.app, params)
 
         expected = {
             'msg': "Cache for repository `%s` was invalidated" % (
                 backend.repo_name,),
             'repository': backend.repo_name,
         }
         assert_ok(id_, expected, given=response.body)
 
     @mock.patch.object(ScmModel, 'mark_for_invalidation', crash)
     def test_api_invalidate_cache_error(self, backend):
         id_, params = build_data(
             self.apikey, 'invalidate_cache', repoid=backend.repo_name)
         response = api_call(self.app, params)
 
         expected = 'Error occurred during cache invalidation action'
         assert_error(id_, expected, given=response.body)
 
     def test_api_invalidate_cache_regular_user_no_permission(self, backend):
         self._set_cache(backend.repo_name)
 
         id_, params = build_data(
             self.apikey_regular, 'invalidate_cache', repoid=backend.repo_name)
         response = api_call(self.app, params)
 
         expected = "repository `%s` does not exist" % (backend.repo_name,)
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_merge_pull_request.py

@@ -1,259 +1,258 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 
 from rhodecode.model.db import UserLog, PullRequest
 from rhodecode.model.meta import Session
 from rhodecode.tests import TEST_USER_ADMIN_LOGIN
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_error, assert_ok)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestMergePullRequest(object):
 
     @pytest.mark.backends("git", "hg")
     def test_api_merge_pull_request_merge_failed(self, pr_util, no_notifications):
         pull_request = pr_util.create_pull_request(mergeable=True)
         pull_request_id = pull_request.pull_request_id
         pull_request_repo = pull_request.target_repo.repo_name
 
         id_, params = build_data(
             self.apikey, 'merge_pull_request',
             repoid=pull_request_repo,
             pullrequestid=pull_request_id)
 
         response = api_call(self.app, params)
 
         # The above api call detaches the pull request DB object from the
         # session because of an unconditional transaction rollback in our
         # middleware. Therefore we need to add it back here if we want to use it.
         Session().add(pull_request)
 
         expected = 'merge not possible for following reasons: ' \
                    'Pull request reviewer approval is pending.'
         assert_error(id_, expected, given=response.body)
 
     @pytest.mark.backends("git", "hg")
     def test_api_merge_pull_request_merge_failed_disallowed_state(
             self, pr_util, no_notifications):
         pull_request = pr_util.create_pull_request(mergeable=True, approved=True)
         pull_request_id = pull_request.pull_request_id
         pull_request_repo = pull_request.target_repo.repo_name
 
         pr = PullRequest.get(pull_request_id)
         pr.pull_request_state = pull_request.STATE_UPDATING
         Session().add(pr)
         Session().commit()
 
         id_, params = build_data(
             self.apikey, 'merge_pull_request',
             repoid=pull_request_repo,
             pullrequestid=pull_request_id)
 
         response = api_call(self.app, params)
         expected = 'Operation forbidden because pull request is in state {}, '\
                    'only state {} is allowed.'.format(PullRequest.STATE_UPDATING,
                                                       PullRequest.STATE_CREATED)
         assert_error(id_, expected, given=response.body)
 
     @pytest.mark.backends("git", "hg")
     def test_api_merge_pull_request(self, pr_util, no_notifications):
         pull_request = pr_util.create_pull_request(mergeable=True, approved=True)
         author = pull_request.user_id
         repo = pull_request.target_repo.repo_id
         pull_request_id = pull_request.pull_request_id
         pull_request_repo = pull_request.target_repo.repo_name
 
         id_, params = build_data(
             self.apikey, 'comment_pull_request',
             repoid=pull_request_repo,
             pullrequestid=pull_request_id,
             status='approved')
 
         response = api_call(self.app, params)
         expected = {
             'comment_id': response.json.get('result', {}).get('comment_id'),
             'pull_request_id': pull_request_id,
             'status': {'given': 'approved', 'was_changed': True}
         }
         assert_ok(id_, expected, given=response.body)
 
         id_, params = build_data(
             self.apikey, 'merge_pull_request',
             repoid=pull_request_repo,
             pullrequestid=pull_request_id)
 
         response = api_call(self.app, params)
 
         pull_request = PullRequest.get(pull_request_id)
 
         expected = {
             'executed': True,
             'failure_reason': 0,
             'merge_status_message': 'This pull request can be automatically merged.',
             'possible': True,
             'merge_commit_id': pull_request.shadow_merge_ref.commit_id,
-            'merge_ref': pull_request.shadow_merge_ref._asdict()
+            'merge_ref': pull_request.shadow_merge_ref.asdict()
         }
 
         assert_ok(id_, expected, response.body)
 
         journal = UserLog.query()\
             .filter(UserLog.user_id == author)\
             .filter(UserLog.repository_id == repo) \
             .order_by(UserLog.user_log_id.asc()) \
             .all()
         assert journal[-2].action == 'repo.pull_request.merge'
         assert journal[-1].action == 'repo.pull_request.close'
 
         id_, params = build_data(
             self.apikey, 'merge_pull_request',
             repoid=pull_request_repo, pullrequestid=pull_request_id)
         response = api_call(self.app, params)
 
         expected = 'merge not possible for following reasons: This pull request is closed.'
         assert_error(id_, expected, given=response.body)
 
     @pytest.mark.backends("git", "hg")
     def test_api_merge_pull_request_as_another_user_no_perms_to_merge(
             self, pr_util, no_notifications, user_util):
         merge_user = user_util.create_user()
         merge_user_id = merge_user.user_id
         merge_user_username = merge_user.username
 
         pull_request = pr_util.create_pull_request(mergeable=True, approved=True)
 
         pull_request_id = pull_request.pull_request_id
         pull_request_repo = pull_request.target_repo.repo_name
 
         id_, params = build_data(
             self.apikey, 'comment_pull_request',
             repoid=pull_request_repo,
             pullrequestid=pull_request_id,
             status='approved')
 
         response = api_call(self.app, params)
         expected = {
             'comment_id': response.json.get('result', {}).get('comment_id'),
             'pull_request_id': pull_request_id,
             'status': {'given': 'approved', 'was_changed': True}
         }
         assert_ok(id_, expected, given=response.body)
         id_, params = build_data(
             self.apikey, 'merge_pull_request',
             repoid=pull_request_repo,
             pullrequestid=pull_request_id,
             userid=merge_user_id
         )
 
         response = api_call(self.app, params)
         expected = 'merge not possible for following reasons: User `{}` ' \
                    'not allowed to perform merge.'.format(merge_user_username)
         assert_error(id_, expected, response.body)
 
     @pytest.mark.backends("git", "hg")
     def test_api_merge_pull_request_as_another_user(self, pr_util, no_notifications, user_util):
         merge_user = user_util.create_user()
         merge_user_id = merge_user.user_id
         pull_request = pr_util.create_pull_request(mergeable=True, approved=True)
         user_util.grant_user_permission_to_repo(
             pull_request.target_repo, merge_user, 'repository.write')
         author = pull_request.user_id
         repo = pull_request.target_repo.repo_id
         pull_request_id = pull_request.pull_request_id
         pull_request_repo = pull_request.target_repo.repo_name
 
         id_, params = build_data(
             self.apikey, 'comment_pull_request',
             repoid=pull_request_repo,
             pullrequestid=pull_request_id,
             status='approved')
 
         response = api_call(self.app, params)
         expected = {
             'comment_id': response.json.get('result', {}).get('comment_id'),
             'pull_request_id': pull_request_id,
             'status': {'given': 'approved', 'was_changed': True}
         }
         assert_ok(id_, expected, given=response.body)
 
         id_, params = build_data(
             self.apikey, 'merge_pull_request',
             repoid=pull_request_repo,
             pullrequestid=pull_request_id,
             userid=merge_user_id
         )
 
         response = api_call(self.app, params)
 
         pull_request = PullRequest.get(pull_request_id)
 
         expected = {
             'executed': True,
             'failure_reason': 0,
             'merge_status_message': 'This pull request can be automatically merged.',
             'possible': True,
             'merge_commit_id': pull_request.shadow_merge_ref.commit_id,
-            'merge_ref': pull_request.shadow_merge_ref._asdict()
+            'merge_ref': pull_request.shadow_merge_ref.asdict()
         }
 
         assert_ok(id_, expected, response.body)
 
         journal = UserLog.query() \
             .filter(UserLog.user_id == merge_user_id) \
             .filter(UserLog.repository_id == repo) \
             .order_by(UserLog.user_log_id.asc()) \
             .all()
         assert journal[-2].action == 'repo.pull_request.merge'
         assert journal[-1].action == 'repo.pull_request.close'
 
         id_, params = build_data(
             self.apikey, 'merge_pull_request',
             repoid=pull_request_repo, pullrequestid=pull_request_id, userid=merge_user_id)
         response = api_call(self.app, params)
 
         expected = 'merge not possible for following reasons: This pull request is closed.'
         assert_error(id_, expected, given=response.body)
 
     @pytest.mark.backends("git", "hg")
     def test_api_merge_pull_request_repo_error(self, pr_util):
         pull_request = pr_util.create_pull_request()
         id_, params = build_data(
             self.apikey, 'merge_pull_request',
             repoid=666, pullrequestid=pull_request.pull_request_id)
         response = api_call(self.app, params)
 
         expected = 'repository `666` does not exist'
         assert_error(id_, expected, given=response.body)
 
     @pytest.mark.backends("git", "hg")
     def test_api_merge_pull_request_non_admin_with_userid_error(self, pr_util):
         pull_request = pr_util.create_pull_request(mergeable=True)
         id_, params = build_data(
             self.apikey_regular, 'merge_pull_request',
             repoid=pull_request.target_repo.repo_name,
             pullrequestid=pull_request.pull_request_id,
             userid=TEST_USER_ADMIN_LOGIN)
         response = api_call(self.app, params)
 
         expected = 'userid is not the same as your user'
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_pull.py

@@ -1,55 +1,54 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 os
 import mock
 import pytest
 
 from rhodecode.tests import TESTS_TMP_PATH
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_ok, assert_error)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestPull(object):
 
     @pytest.mark.backends("git", "hg")
     def test_api_pull(self, backend):
         r = backend.create_repo()
         repo_name = r.repo_name
         clone_uri = os.path.join(TESTS_TMP_PATH, backend.repo_name)
         r.clone_uri = clone_uri
 
         id_, params = build_data(self.apikey, 'pull', repoid=repo_name,)
         with mock.patch('rhodecode.model.scm.url_validator'):
             response = api_call(self.app, params)
             msg = 'Pulled from url `%s` on repo `%s`' % (
                 clone_uri, repo_name)
             expected = {'msg': msg,
                         'repository': repo_name}
             assert_ok(id_, expected, given=response.body)
 
     def test_api_pull_error(self, backend):
         id_, params = build_data(
             self.apikey, 'pull', repoid=backend.repo_name)
         response = api_call(self.app, params)
 
         expected = 'Unable to pull changes from `None`'
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_remove_field_from_repo.py

@@ -1,66 +1,65 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 
 from rhodecode.model.db import Repository, RepositoryField
 from rhodecode.api.tests.utils import build_data, api_call, assert_ok
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestRemoveFieldFromRepo(object):
     def test_api_remove_field_from_repo(self, backend):
         repo = backend.create_repo()
         repo_name = repo.repo_name
 
         id_, params = build_data(
             self.apikey, 'add_field_to_repo',
             repoid=repo_name,
             key='extra_field',
             label='extra_field_label',
             description='extra_field_desc')
         response = api_call(self.app, params)
         expected = {
             'msg': 'Added new repository field `extra_field`',
             'success': True,
         }
         assert_ok(id_, expected, given=response.body)
 
         repo = Repository.get_by_repo_name(repo_name)
         repo_field = RepositoryField.get_by_key_name('extra_field', repo)
         _data = repo_field.get_dict()
         assert _data['field_desc'] == 'extra_field_desc'
         assert _data['field_key'] == 'extra_field'
         assert _data['field_label'] == 'extra_field_label'
 
         id_, params = build_data(
             self.apikey, 'remove_field_from_repo',
             repoid=repo_name,
             key='extra_field')
         response = api_call(self.app, params)
         expected = {
             'msg': 'Deleted repository field `extra_field`',
             'success': True,
         }
         assert_ok(id_, expected, given=response.body)
         repo = Repository.get_by_repo_name(repo_name)
         repo_field = RepositoryField.get_by_key_name('extra_field', repo)
 
         assert repo_field is None

rhodecode/api/tests/test_remove_user_from_user_group.py

@@ -1,58 +1,57 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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_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 TestRemoveUserFromUserGroup(object):
     def test_api_remove_user_from_user_group(self, user_util):
         user, group = user_util.create_user_with_group()
         user_name = user.username
         group_name = group.users_group_name
         id_, params = build_data(
             self.apikey, 'remove_user_from_user_group',
             usergroupid=group_name,
             userid=user.username)
         response = api_call(self.app, params)
 
         expected = {
             'msg': 'removed member `%s` from user group `%s`' % (
                 user_name, group_name
             ),
             'success': True}
         assert_ok(id_, expected, given=response.body)
 
     @mock.patch.object(UserGroupModel, 'remove_user_from_group', crash)
     def test_api_remove_user_from_user_group_exception_occurred(
             self, user_util):
         user, group = user_util.create_user_with_group()
         id_, params = build_data(
             self.apikey, 'remove_user_from_user_group',
             usergroupid=group.users_group_name, userid=user.username)
         response = api_call(self.app, params)
         expected = 'failed to remove member from user group `%s`' % (
             group.users_group_name)
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_repo_locking.py

@@ -1,184 +1,183 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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.db import Repository
 from rhodecode.model.user import UserModel
 from rhodecode.lib.ext_json import json
 from rhodecode.lib.utils2 import time_to_datetime
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_ok, assert_error, crash)
 from rhodecode.tests import TEST_USER_ADMIN_LOGIN
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestLock(object):
     def test_api_lock_repo_lock_aquire(self, backend):
         id_, params = build_data(
             self.apikey, 'lock',
             userid=TEST_USER_ADMIN_LOGIN,
             repoid=backend.repo_name,
             locked=True)
         response = api_call(self.app, params)
         expected = {
             'repo': backend.repo_name, 'locked': True,
             'locked_since': response.json['result']['locked_since'],
             'locked_by': TEST_USER_ADMIN_LOGIN,
             'lock_state_changed': True,
             'lock_reason': Repository.LOCK_API,
             'msg': ('User `%s` set lock state for repo `%s` to `%s`'
                     % (TEST_USER_ADMIN_LOGIN, backend.repo_name, True))
         }
         assert_ok(id_, expected, given=response.body)
 
     def test_repo_lock_aquire_by_non_admin(self, backend):
         repo = backend.create_repo(cur_user=self.TEST_USER_LOGIN)
         repo_name = repo.repo_name
         id_, params = build_data(
             self.apikey_regular, 'lock',
             repoid=repo_name,
             locked=True)
         response = api_call(self.app, params)
         expected = {
             'repo': repo_name,
             'locked': True,
             'locked_since': response.json['result']['locked_since'],
             'locked_by': self.TEST_USER_LOGIN,
             'lock_state_changed': True,
             'lock_reason': Repository.LOCK_API,
             'msg': ('User `%s` set lock state for repo `%s` to `%s`'
                     % (self.TEST_USER_LOGIN, repo_name, True))
         }
         assert_ok(id_, expected, given=response.body)
 
     def test_api_lock_repo_lock_aquire_non_admin_with_userid(self, backend):
         repo = backend.create_repo(cur_user=self.TEST_USER_LOGIN)
         repo_name = repo.repo_name
         id_, params = build_data(
             self.apikey_regular, 'lock',
             userid=TEST_USER_ADMIN_LOGIN,
             repoid=repo_name,
             locked=True)
         response = api_call(self.app, params)
         expected = 'userid is not the same as your user'
         assert_error(id_, expected, given=response.body)
 
     def test_api_lock_repo_lock_aquire_non_admin_not_his_repo(self, backend):
         id_, params = build_data(
             self.apikey_regular, 'lock',
             repoid=backend.repo_name,
             locked=True)
         response = api_call(self.app, params)
         expected = 'repository `%s` does not exist' % (backend.repo_name, )
         assert_error(id_, expected, given=response.body)
 
     def test_api_lock_repo_lock_release(self, backend):
         id_, params = build_data(
             self.apikey, 'lock',
             userid=TEST_USER_ADMIN_LOGIN,
             repoid=backend.repo_name,
             locked=False)
         response = api_call(self.app, params)
         expected = {
             'repo': backend.repo_name,
             'locked': False,
             'locked_since': None,
             'locked_by': TEST_USER_ADMIN_LOGIN,
             'lock_state_changed': True,
             'lock_reason': Repository.LOCK_API,
             'msg': ('User `%s` set lock state for repo `%s` to `%s`'
                     % (TEST_USER_ADMIN_LOGIN, backend.repo_name, False))
         }
         assert_ok(id_, expected, given=response.body)
 
     def test_api_lock_repo_lock_aquire_optional_userid(self, backend):
         id_, params = build_data(
             self.apikey, 'lock',
             repoid=backend.repo_name,
             locked=True)
         response = api_call(self.app, params)
         time_ = response.json['result']['locked_since']
         expected = {
             'repo': backend.repo_name,
             'locked': True,
             'locked_since': time_,
             'locked_by': TEST_USER_ADMIN_LOGIN,
             'lock_state_changed': True,
             'lock_reason': Repository.LOCK_API,
             'msg': ('User `%s` set lock state for repo `%s` to `%s`'
                     % (TEST_USER_ADMIN_LOGIN, backend.repo_name, True))
         }
 
         assert_ok(id_, expected, given=response.body)
 
     def test_api_lock_repo_lock_optional_locked(self, backend):
         # TODO: Provide a fixture locked_repository or similar
         repo = Repository.get_by_repo_name(backend.repo_name)
         user = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
         Repository.lock(repo, user.user_id, lock_reason=Repository.LOCK_API)
 
         id_, params = build_data(self.apikey, 'lock', repoid=backend.repo_name)
         response = api_call(self.app, params)
         time_ = response.json['result']['locked_since']
         expected = {
             'repo': backend.repo_name,
             'locked': True,
             'locked_since': time_,
             'locked_by': TEST_USER_ADMIN_LOGIN,
             'lock_state_changed': False,
             'lock_reason': Repository.LOCK_API,
             'msg': ('Repo `%s` locked by `%s` on `%s`.'
                     % (backend.repo_name, TEST_USER_ADMIN_LOGIN,
                        json.dumps(time_to_datetime(time_))))
         }
         assert_ok(id_, expected, given=response.body)
 
     def test_api_lock_repo_lock_optional_not_locked(self, backend):
         repo = backend.create_repo(cur_user=self.TEST_USER_LOGIN)
         repo_name = repo.repo_name
         assert repo.locked == [None, None, None]
         id_, params = build_data(self.apikey, 'lock', repoid=repo.repo_id)
         response = api_call(self.app, params)
         expected = {
             'repo': repo_name,
             'locked': False,
             'locked_since': None,
             'locked_by': None,
             'lock_state_changed': False,
             'lock_reason': None,
             'msg': ('Repo `%s` not locked.' % (repo_name,))
         }
         assert_ok(id_, expected, given=response.body)
 
     @mock.patch.object(Repository, 'lock', crash)
     def test_api_lock_error(self, backend):
         id_, params = build_data(
             self.apikey, 'lock',
             userid=TEST_USER_ADMIN_LOGIN,
             repoid=backend.repo_name,
             locked=True)
         response = api_call(self.app, params)
 
         expected = 'Error occurred locking repository `%s`' % (
             backend.repo_name,)
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_rescan_repos.py

@@ -1,44 +1,43 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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.scm import ScmModel
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_ok, assert_error, crash)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestRescanRepos(object):
     def test_api_rescan_repos(self):
         id_, params = build_data(self.apikey, 'rescan_repos')
         response = api_call(self.app, params)
 
         expected = {'added': [], 'removed': []}
         assert_ok(id_, expected, given=response.body)
 
     @mock.patch.object(ScmModel, 'repo_scan', crash)
     def test_api_rescann_error(self):
         id_, params = build_data(self.apikey, 'rescan_repos', )
         response = api_call(self.app, params)
 
         expected = 'Error occurred during rescan repositories action'
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_revoke_user_group_permission.py

@@ -1,67 +1,66 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 TestRevokeUserGroupPermission(object):
     def test_api_revoke_user_group_permission(self, backend, user_util):
         repo = backend.create_repo()
         user_group = user_util.create_user_group()
         user_util.grant_user_group_permission_to_repo(
             repo, user_group, 'repository.read')
         id_, params = build_data(
             self.apikey,
             'revoke_user_group_permission',
             repoid=backend.repo_name,
             usergroupid=user_group.users_group_name)
         response = api_call(self.app, params)
 
         expected = {
             'msg': 'Revoked perm for user group: `%s` in repo: `%s`' % (
                 user_group.users_group_name, backend.repo_name
             ),
             'success': True
         }
         assert_ok(id_, expected, given=response.body)
 
     @mock.patch.object(RepoModel, 'revoke_user_group_permission', crash)
     def test_api_revoke_user_group_permission_exception_when_adding(
             self, backend, user_util):
         user_group = user_util.create_user_group()
         id_, params = build_data(
             self.apikey,
             'revoke_user_group_permission',
             repoid=backend.repo_name,
             usergroupid=user_group.users_group_name)
         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_revoke_user_group_permission_from_repo_group.py

@@ -1,129 +1,128 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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_group import RepoGroupModel
 from rhodecode.tests import TEST_USER_ADMIN_LOGIN
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_error, assert_ok, crash)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestRevokeUserGroupPermissionFromRepoGroup(object):
     @pytest.mark.parametrize("name, apply_to_children", [
         ('none', 'none'),
         ('all', 'all'),
         ('repos', 'repos'),
         ('groups', 'groups'),
     ])
     def test_api_revoke_user_group_permission_from_repo_group(
             self, name, apply_to_children, user_util):
         user_group = user_util.create_user_group()
         repo_group = user_util.create_repo_group()
         user_util.grant_user_group_permission_to_repo_group(
             repo_group, user_group, 'group.read')
 
         id_, params = build_data(
             self.apikey, 'revoke_user_group_permission_from_repo_group',
             repogroupid=repo_group.name,
             usergroupid=user_group.users_group_name,
             apply_to_children=apply_to_children,)
         response = api_call(self.app, params)
 
         expected = {
             'msg': (
                 'Revoked perm (recursive:%s) for user group: `%s`'
                 ' in repo group: `%s`' % (
                     apply_to_children, user_group.users_group_name,
                     repo_group.name
                 )
             ),
             'success': True
         }
         assert_ok(id_, expected, given=response.body)
 
     @pytest.mark.parametrize(
         "name, apply_to_children, grant_admin, access_ok", [
             ('none', 'none', False, False),
             ('all', 'all', False, False),
             ('repos', 'repos', False, False),
             ('groups', 'groups', False, False),
 
             # after granting admin rights
             ('none', 'none', False, False),
             ('all', 'all', False, False),
             ('repos', 'repos', False, False),
             ('groups', 'groups', False, False),
         ]
     )
     def test_api_revoke_user_group_permission_from_repo_group_by_regular_user(
             self, name, apply_to_children, grant_admin, access_ok, user_util):
         user_group = user_util.create_user_group()
         repo_group = user_util.create_repo_group()
         user_util.grant_user_group_permission_to_repo_group(
             repo_group, user_group, 'group.read')
 
         if grant_admin:
             user_util.grant_user_permission_to_repo_group(
                 repo_group.name, self.TEST_USER_LOGIN, 'group.admin')
 
         id_, params = build_data(
             self.apikey_regular,
             'revoke_user_group_permission_from_repo_group',
             repogroupid=repo_group.name,
             usergroupid=user_group.users_group_name,
             apply_to_children=apply_to_children,)
         response = api_call(self.app, params)
         if access_ok:
             expected = {
                 'msg': (
                     'Revoked perm (recursive:%s) for user group: `%s`'
                     ' in repo group: `%s`' % (
                         apply_to_children, TEST_USER_ADMIN_LOGIN,
                         repo_group.name
                     )
                 ),
                 'success': True
             }
             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)
 
     @mock.patch.object(RepoGroupModel, 'revoke_user_group_permission', crash)
     def test_api_revoke_user_group_permission_from_repo_group_exception_on_add(
             self, user_util):
         user_group = user_util.create_user_group()
         repo_group = user_util.create_repo_group()
         id_, params = build_data(
             self.apikey, 'revoke_user_group_permission_from_repo_group',
             repogroupid=repo_group.name,
             usergroupid=user_group.users_group_name)
         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_revoke_user_group_permission_from_user_group.py

@@ -1,58 +1,57 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 
 from rhodecode.model.user import UserModel
 from rhodecode.api.tests.utils import build_data, api_call, assert_ok
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestRevokeUserGroupPermissionFromUserGroup(object):
     @pytest.mark.parametrize("name", [
         ('none',),
         ('all',),
         ('repos',),
         ('groups',),
     ])
     def test_api_revoke_user_group_permission_from_user_group(
             self, name, user_util):
         user = UserModel().get_by_username(self.TEST_USER_LOGIN)
         group = user_util.create_user_group()
         source_group = user_util.create_user_group()
 
         user_util.grant_user_permission_to_user_group(
             group, user, 'usergroup.read')
         user_util.grant_user_group_permission_to_user_group(
             source_group, group, 'usergroup.read')
 
         id_, params = build_data(
             self.apikey, 'revoke_user_group_permission_from_user_group',
             usergroupid=group.users_group_name,
             sourceusergroupid=source_group.users_group_name)
         response = api_call(self.app, params)
 
         expected = {
             'msg': 'Revoked perm for user group: `%s` in user group: `%s`' % (
                 source_group.users_group_name, group.users_group_name
             ),
             'success': True
         }
         assert_ok(id_, expected, given=response.body)

rhodecode/api/tests/test_revoke_user_permission.py

@@ -1,66 +1,65 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 TestRevokeUserPermission(object):
     def test_api_revoke_user_permission(self, backend, user_util):
         repo = backend.create_repo()
         user = user_util.create_user()
         user_util.grant_user_permission_to_repo(
             repo, user, 'repository.read')
 
         id_, params = build_data(
             self.apikey,
             'revoke_user_permission',
             repoid=repo.repo_name,
             userid=user.username)
         response = api_call(self.app, params)
 
         expected = {
             'msg': 'Revoked perm for user: `%s` in repo: `%s`' % (
                 user.username, backend.repo_name
             ),
             'success': True
         }
         assert_ok(id_, expected, given=response.body)
 
     @mock.patch.object(RepoModel, 'revoke_user_permission', crash)
     def test_api_revoke_user_permission_exception_when_adding(
             self, backend, user_util):
         user = user_util.create_user()
         id_, params = build_data(
             self.apikey,
             'revoke_user_permission',
             repoid=backend.repo_name,
             userid=user.username)
         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_revoke_user_permission_from_repo_group.py

@@ -1,126 +1,125 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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_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 TestRevokeUserPermissionFromRepoGroup(object):
     @pytest.mark.parametrize("name, apply_to_children", [
         ('none', 'none'),
         ('all', 'all'),
         ('repos', 'repos'),
         ('groups', 'groups'),
     ])
     def test_api_revoke_user_permission_from_repo_group(
             self, name, apply_to_children, user_util):
         user = user_util.create_user()
         repo_group = user_util.create_repo_group()
         user_util.grant_user_permission_to_repo_group(
             repo_group, user, 'group.read')
 
         id_, params = build_data(
             self.apikey,
             'revoke_user_permission_from_repo_group',
             repogroupid=repo_group.name,
             userid=user.username,
             apply_to_children=apply_to_children,)
         response = api_call(self.app, params)
 
         expected = {
             'msg': (
                 'Revoked perm (recursive:%s) for user: `%s`'
                 ' in repo group: `%s`' % (
                     apply_to_children, user.username, repo_group.name
                 )
             ),
             'success': True
         }
         assert_ok(id_, expected, given=response.body)
 
     @pytest.mark.parametrize(
         "name, apply_to_children, grant_admin, access_ok", [
             ('none', 'none', False, False),
             ('all', 'all', False, False),
             ('repos', 'repos', False, False),
             ('groups', 'groups', False, False),
 
             # after granting admin rights
             ('none', 'none', False, False),
             ('all', 'all', False, False),
             ('repos', 'repos', False, False),
             ('groups', 'groups', False, False),
         ]
     )
     def test_api_revoke_user_permission_from_repo_group_by_regular_user(
             self, name, apply_to_children, grant_admin, access_ok, user_util):
         user = user_util.create_user()
         repo_group = user_util.create_repo_group()
         permission = 'group.admin' if grant_admin else 'group.read'
         user_util.grant_user_permission_to_repo_group(
             repo_group, user, permission)
 
         id_, params = build_data(
             self.apikey_regular,
             'revoke_user_permission_from_repo_group',
             repogroupid=repo_group.name,
             userid=user.username,
             apply_to_children=apply_to_children,)
         response = api_call(self.app, params)
         if access_ok:
             expected = {
                 'msg': (
                     'Revoked perm (recursive:%s) for user: `%s`'
                     ' in repo group: `%s`' % (
                         apply_to_children, user.username, repo_group.name
                     )
                 ),
                 'success': True
             }
             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)
 
     @mock.patch.object(RepoGroupModel, 'revoke_user_permission', crash)
     def test_api_revoke_user_permission_from_repo_group_exception_when_adding(
             self, user_util):
         user = user_util.create_user()
         repo_group = user_util.create_repo_group()
         id_, params = build_data(
             self.apikey,
             'revoke_user_permission_from_repo_group',
             repogroupid=repo_group.name,
             userid=user.username
         )
         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_revoke_user_permission_from_user_group.py

@@ -1,112 +1,111 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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_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 TestRevokeUserPermissionFromUserGroup(object):
     @pytest.mark.parametrize("name", [
         ('none',),
         ('all',),
         ('repos',),
         ('groups',),
     ])
     def test_api_revoke_user_permission_from_user_group(self, name, user_util):
         user = user_util.create_user()
         group = user_util.create_user_group()
         user_util.grant_user_permission_to_user_group(
             group, user, 'usergroup.admin')
 
         id_, params = build_data(
             self.apikey,
             'revoke_user_permission_from_user_group',
             usergroupid=group.users_group_name,
             userid=user.username)
         response = api_call(self.app, params)
 
         expected = {
             'msg': 'Revoked perm for user: `%s` in user group: `%s`' % (
                 user.username, group.users_group_name
             ),
             'success': True
         }
         assert_ok(id_, expected, given=response.body)
 
     @pytest.mark.parametrize("name, grant_admin, access_ok", [
         ('none', False, False),
         ('all', False, False),
         ('repos', False, False),
         ('groups', False, False),
 
         # after granting admin rights
         ('none', False, False),
         ('all', False, False),
         ('repos', False, False),
         ('groups', False, False),
     ])
     def test_api_revoke_user_permission_from_user_group_by_regular_user(
             self, name, grant_admin, access_ok, user_util):
         user = user_util.create_user()
         group = user_util.create_user_group()
         permission = 'usergroup.admin' if grant_admin else 'usergroup.read'
         user_util.grant_user_permission_to_user_group(group, user, permission)
 
         id_, params = build_data(
             self.apikey_regular,
             'revoke_user_permission_from_user_group',
             usergroupid=group.users_group_name,
             userid=user.username)
         response = api_call(self.app, params)
         if access_ok:
             expected = {
                 'msg': 'Revoked perm for user: `%s` in user group: `%s`' % (
                     user.username, group.users_group_name
                 ),
                 'success': True
             }
             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)
 
     @mock.patch.object(UserGroupModel, 'revoke_user_permission', crash)
     def test_api_revoke_user_permission_from_user_group_exception_when_adding(
             self, user_util):
         user = user_util.create_user()
         group = user_util.create_user_group()
         id_, params = build_data(
             self.apikey,
             'revoke_user_permission_from_user_group',
             usergroupid=group.users_group_name,
             userid=user.username)
         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/tests/test_store_exception.py

@@ -1,59 +1,58 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 
 from rhodecode.api.tests.utils import build_data, api_call, assert_ok, assert_error
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestStoreException(object):
 
     def test_store_exception_invalid_json(self):
         id_, params = build_data(self.apikey, 'store_exception',
                                  exc_data_json='XXX,{')
         response = api_call(self.app, params)
 
         expected = 'Failed to parse JSON data from exc_data_json field. ' \
                    'Please make sure it contains a valid JSON.'
         assert_error(id_, expected, given=response.body)
 
     def test_store_exception_missing_json_params_json(self):
         id_, params = build_data(self.apikey, 'store_exception',
                                  exc_data_json='{"foo":"bar"}')
         response = api_call(self.app, params)
 
         expected = "Missing exc_traceback, or exc_type_name in " \
                    "exc_data_json field. Missing: 'exc_traceback'"
         assert_error(id_, expected, given=response.body)
 
     def test_store_exception(self):
         id_, params = build_data(
             self.apikey, 'store_exception',
             exc_data_json='{"exc_traceback": "invalid", "exc_type_name":"ValueError"}')
         response = api_call(self.app, params)
         exc_id = response.json['result']['exc_id']
 
         expected = {
             'exc_id': exc_id,
             'exc_url': 'http://example.com/_admin/settings/exceptions/{}'.format(exc_id)
         }
         assert_ok(id_, expected, given=response.body)

rhodecode/api/tests/test_update_pull_request.py

@@ -1,215 +1,214 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 
 from rhodecode.lib.vcs.nodes import FileNode
 from rhodecode.model.db import User
 from rhodecode.model.pull_request import PullRequestModel
 from rhodecode.tests import TEST_USER_ADMIN_LOGIN
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_ok, assert_error)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestUpdatePullRequest(object):
 
     @pytest.mark.backends("git", "hg")
     def test_api_update_pull_request_title_or_description(
             self, pr_util, no_notifications):
         pull_request = pr_util.create_pull_request()
 
         id_, params = build_data(
             self.apikey, 'update_pull_request',
             repoid=pull_request.target_repo.repo_name,
             pullrequestid=pull_request.pull_request_id,
             title='New TITLE OF A PR',
             description='New DESC OF A PR',
         )
         response = api_call(self.app, params)
 
         expected = {
             "msg": "Updated pull request `{}`".format(
                 pull_request.pull_request_id),
             "pull_request": response.json['result']['pull_request'],
             "updated_commits": {"added": [], "common": [], "removed": []},
             "updated_reviewers": {"added": [], "removed": []},
             "updated_observers": {"added": [], "removed": []},
         }
 
         response_json = response.json['result']
         assert response_json == expected
         pr = response_json['pull_request']
         assert pr['title'] == 'New TITLE OF A PR'
         assert pr['description'] == 'New DESC OF A PR'
 
     @pytest.mark.backends("git", "hg")
     def test_api_try_update_closed_pull_request(
             self, pr_util, no_notifications):
         pull_request = pr_util.create_pull_request()
         PullRequestModel().close_pull_request(
             pull_request, TEST_USER_ADMIN_LOGIN)
 
         id_, params = build_data(
             self.apikey, 'update_pull_request',
             repoid=pull_request.target_repo.repo_name,
             pullrequestid=pull_request.pull_request_id)
         response = api_call(self.app, params)
 
         expected = 'pull request `{}` update failed, pull request ' \
                    'is closed'.format(pull_request.pull_request_id)
 
         assert_error(id_, expected, response.body)
 
     @pytest.mark.backends("git", "hg")
     def test_api_update_update_commits(self, pr_util, no_notifications):
         commits = [
             {'message': 'a'},
-            {'message': 'b', 'added': [FileNode('file_b', 'test_content\n')]},
-            {'message': 'c', 'added': [FileNode('file_c', 'test_content\n')]},
+            {'message': 'b', 'added': [FileNode(b'file_b', b'test_content\n')]},
+            {'message': 'c', 'added': [FileNode(b'file_c', b'test_content\n')]},
         ]
         pull_request = pr_util.create_pull_request(
             commits=commits, target_head='a', source_head='b', revisions=['b'])
         pr_util.update_source_repository(head='c')
         repo = pull_request.source_repo.scm_instance()
         commits = [x for x in repo.get_commits()]
 
         added_commit_id = commits[-1].raw_id  # c commit
         common_commit_id = commits[1].raw_id  # b commit is common ancestor
         total_commits = [added_commit_id, common_commit_id]
 
         id_, params = build_data(
             self.apikey, 'update_pull_request',
             repoid=pull_request.target_repo.repo_name,
             pullrequestid=pull_request.pull_request_id,
             update_commits=True
         )
         response = api_call(self.app, params)
 
         expected = {
             "msg": "Updated pull request `{}`".format(
                 pull_request.pull_request_id),
             "pull_request": response.json['result']['pull_request'],
             "updated_commits": {"added": [added_commit_id],
                                 "common": [common_commit_id],
                                 "total": total_commits,
                                 "removed": []},
             "updated_reviewers": {"added": [], "removed": []},
             "updated_observers": {"added": [], "removed": []},
         }
 
         assert_ok(id_, expected, response.body)
 
     @pytest.mark.backends("git", "hg")
     def test_api_update_change_reviewers(
             self, user_util, pr_util, no_notifications):
         a = user_util.create_user()
         b = user_util.create_user()
         c = user_util.create_user()
         new_reviewers = [
             {'username': b.username, 'reasons': ['updated via API'],
              'mandatory':False},
             {'username': c.username, 'reasons': ['updated via API'],
              'mandatory':False},
         ]
 
         added = [b.username, c.username]
         removed = [a.username]
 
         pull_request = pr_util.create_pull_request(
             reviewers=[(a.username, ['added via API'], False, 'reviewer', [])])
 
         id_, params = build_data(
             self.apikey, 'update_pull_request',
             repoid=pull_request.target_repo.repo_name,
             pullrequestid=pull_request.pull_request_id,
             reviewers=new_reviewers)
         response = api_call(self.app, params)
         expected = {
             "msg": "Updated pull request `{}`".format(
                 pull_request.pull_request_id),
             "pull_request": response.json['result']['pull_request'],
             "updated_commits": {"added": [], "common": [], "removed": []},
             "updated_reviewers": {"added": added, "removed": removed},
             "updated_observers": {"added": [], "removed": []},
         }
 
         assert_ok(id_, expected, response.body)
 
     @pytest.mark.backends("git", "hg")
     def test_api_update_bad_user_in_reviewers(self, pr_util):
         pull_request = pr_util.create_pull_request()
 
         id_, params = build_data(
             self.apikey, 'update_pull_request',
             repoid=pull_request.target_repo.repo_name,
             pullrequestid=pull_request.pull_request_id,
             reviewers=[{'username': 'bad_name'}])
         response = api_call(self.app, params)
 
         expected = 'user `bad_name` does not exist'
 
         assert_error(id_, expected, response.body)
 
     @pytest.mark.backends("git", "hg")
     def test_api_update_repo_error(self, pr_util):
         pull_request = pr_util.create_pull_request()
         id_, params = build_data(
             self.apikey, 'update_pull_request',
             repoid='fake',
             pullrequestid=pull_request.pull_request_id,
             reviewers=[{'username': 'bad_name'}])
         response = api_call(self.app, params)
 
         expected = 'repository `fake` does not exist'
 
         response_json = response.json['error']
         assert response_json == expected
 
     @pytest.mark.backends("git", "hg")
     def test_api_update_pull_request_error(self, pr_util):
         pull_request = pr_util.create_pull_request()
 
         id_, params = build_data(
             self.apikey, 'update_pull_request',
             repoid=pull_request.target_repo.repo_name,
             pullrequestid=999999,
             reviewers=[{'username': 'bad_name'}])
         response = api_call(self.app, params)
 
         expected = 'pull request `999999` does not exist'
         assert_error(id_, expected, response.body)
 
     @pytest.mark.backends("git", "hg")
     def test_api_update_pull_request_no_perms_to_update(
             self, user_util, pr_util):
         user = user_util.create_user()
         pull_request = pr_util.create_pull_request()
 
         id_, params = build_data(
             user.api_key, 'update_pull_request',
             repoid=pull_request.target_repo.repo_name,
             pullrequestid=pull_request.pull_request_id,)
         response = api_call(self.app, params)
 
         expected = ('pull request `%s` update failed, '
                     'no permission to update.') % pull_request.pull_request_id
 
         assert_error(id_, expected, response.body)

rhodecode/api/tests/test_update_repo.py

@@ -1,210 +1,209 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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.model.scm import ScmModel
 from rhodecode.tests import TEST_USER_ADMIN_LOGIN, TEST_USER_REGULAR_LOGIN
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_error, assert_ok, crash, jsonify)
 from rhodecode.tests.fixture import Fixture
 from rhodecode.tests.fixture_mods.fixture_utils import plain_http_host_only_stub
 
 fixture = Fixture()
 
 UPDATE_REPO_NAME = 'api_update_me'
 
 
 class SAME_AS_UPDATES(object):
     """ Constant used for tests below """
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestApiUpdateRepo(object):
 
     @pytest.mark.parametrize("updates, expected", [
         ({'owner': TEST_USER_REGULAR_LOGIN},
          SAME_AS_UPDATES),
 
         ({'description': 'new description'},
          SAME_AS_UPDATES),
 
         ({'clone_uri': 'http://foo.com/repo'},
          SAME_AS_UPDATES),
 
         ({'clone_uri': None},
          {'clone_uri': ''}),
 
         ({'clone_uri': ''},
          {'clone_uri': ''}),
 
         ({'clone_uri': 'http://example.com/repo_pull'},
          {'clone_uri': 'http://example.com/repo_pull'}),
 
         ({'push_uri': ''},
          {'push_uri': ''}),
 
         ({'push_uri': 'http://example.com/repo_push'},
          {'push_uri': 'http://example.com/repo_push'}),
 
         ({'landing_rev': None},  # auto-updated based on type of repo
          {'landing_rev': [None, None]}),
 
         ({'enable_statistics': True},
          SAME_AS_UPDATES),
 
         ({'enable_locking': True},
          SAME_AS_UPDATES),
 
         ({'enable_downloads': True},
          SAME_AS_UPDATES),
 
         ({'repo_name': 'new_repo_name'},
          {
             'repo_name': 'new_repo_name',
             'url': 'http://{}/new_repo_name'.format(plain_http_host_only_stub())
          }),
 
         ({'repo_name': 'test_group_for_update/{}'.format(UPDATE_REPO_NAME),
           '_group': 'test_group_for_update'},
          {
             'repo_name': 'test_group_for_update/{}'.format(UPDATE_REPO_NAME),
             'url': 'http://{}/test_group_for_update/{}'.format(
                 plain_http_host_only_stub(), UPDATE_REPO_NAME)
          }),
     ])
     def test_api_update_repo(self, updates, expected, backend):
         repo_name = UPDATE_REPO_NAME
         repo = fixture.create_repo(repo_name, repo_type=backend.alias)
         if updates.get('_group'):
             fixture.create_repo_group(updates['_group'])
 
         if 'landing_rev' in updates:
             default_landing_ref, _lbl = ScmModel.backend_landing_ref(backend.alias)
             _type, _name = default_landing_ref.split(':')
             updates['landing_rev'] = default_landing_ref
             expected['landing_rev'] = [_type, _name]
 
         expected_api_data = repo.get_api_data(include_secrets=True)
         if expected is SAME_AS_UPDATES:
             expected_api_data.update(updates)
         else:
             expected_api_data.update(expected)
 
         id_, params = build_data(
             self.apikey, 'update_repo', repoid=repo_name, **updates)
 
         with mock.patch('rhodecode.model.validation_schema.validators.url_validator'):
             response = api_call(self.app, params)
 
         if updates.get('repo_name'):
             repo_name = updates['repo_name']
 
         try:
             expected = {
                 'msg': 'updated repo ID:%s %s' % (repo.repo_id, repo_name),
                 'repository': jsonify(expected_api_data)
             }
             assert_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
             if updates.get('_group'):
                 fixture.destroy_repo_group(updates['_group'])
 
     def test_api_update_repo_fork_of_field(self, backend):
         master_repo = backend.create_repo()
         repo = backend.create_repo()
         updates = {
             'fork_of': master_repo.repo_name,
             'fork_of_id': master_repo.repo_id
         }
         expected_api_data = repo.get_api_data(include_secrets=True)
         expected_api_data.update(updates)
 
         id_, params = build_data(
             self.apikey, 'update_repo', repoid=repo.repo_name, **updates)
         response = api_call(self.app, params)
         expected = {
             'msg': 'updated repo ID:%s %s' % (repo.repo_id, repo.repo_name),
             'repository': jsonify(expected_api_data)
         }
         assert_ok(id_, expected, given=response.body)
         result = response.json['result']['repository']
         assert result['fork_of'] == master_repo.repo_name
         assert result['fork_of_id'] == master_repo.repo_id
 
     def test_api_update_repo_fork_of_not_found(self, backend):
         master_repo_name = 'fake-parent-repo'
         repo = backend.create_repo()
         updates = {
             'fork_of': master_repo_name
         }
         id_, params = build_data(
             self.apikey, 'update_repo', repoid=repo.repo_name, **updates)
         response = api_call(self.app, params)
         expected = {
             'repo_fork_of': 'Fork with id `{}` does not exists'.format(
                 master_repo_name)}
         assert_error(id_, expected, given=response.body)
 
     def test_api_update_repo_with_repo_group_not_existing(self):
         repo_name = 'admin_owned'
         fake_repo_group = 'test_group_for_update'
         fixture.create_repo(repo_name)
         updates = {'repo_name': '{}/{}'.format(fake_repo_group, repo_name)}
         id_, params = build_data(
             self.apikey, 'update_repo', repoid=repo_name, **updates)
         response = api_call(self.app, params)
         try:
             expected = {
                 'repo_group': 'Repository group `{}` does not exist'.format(fake_repo_group)
             }
             assert_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
 
     def test_api_update_repo_regular_user_not_allowed(self):
         repo_name = 'admin_owned'
         fixture.create_repo(repo_name)
         updates = {'active': False}
         id_, params = build_data(
             self.apikey_regular, 'update_repo', repoid=repo_name, **updates)
         response = api_call(self.app, params)
         try:
             expected = 'repository `%s` does not exist' % (repo_name,)
             assert_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
 
     @mock.patch.object(RepoModel, 'update', crash)
     def test_api_update_repo_exception_occurred(self, backend):
         repo_name = UPDATE_REPO_NAME
         fixture.create_repo(repo_name, repo_type=backend.alias)
         id_, params = build_data(
             self.apikey, 'update_repo', repoid=repo_name,
             owner=TEST_USER_ADMIN_LOGIN,)
         response = api_call(self.app, params)
         try:
             expected = 'failed to update repo `%s`' % (repo_name,)
             assert_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)

rhodecode/api/tests/test_update_repo_group.py

@@ -1,150 +1,149 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 os
 
 import pytest
 
 from rhodecode.model.repo_group import RepoGroupModel
 from rhodecode.model.user import UserModel
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_error, assert_ok)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestApiUpdateRepoGroup(object):
 
     def test_update_group_name(self, user_util):
         new_group_name = 'new-group'
         initial_name = self._update(user_util, group_name=new_group_name)
         assert RepoGroupModel()._get_repo_group(initial_name) is None
         new_group = RepoGroupModel()._get_repo_group(new_group_name)
         assert new_group is not None
         assert new_group.full_path == new_group_name
 
     def test_update_group_name_change_parent(self, user_util):
 
         parent_group = user_util.create_repo_group()
         parent_group_name = parent_group.name
 
         expected_group_name = '{}/{}'.format(parent_group_name, 'new-group')
         initial_name = self._update(user_util, group_name=expected_group_name)
 
         repo_group = RepoGroupModel()._get_repo_group(expected_group_name)
 
         assert repo_group is not None
         assert repo_group.group_name == expected_group_name
         assert repo_group.full_path == expected_group_name
         assert RepoGroupModel()._get_repo_group(initial_name) is None
 
         new_path = os.path.join(
             RepoGroupModel().repos_path, *repo_group.full_path_splitted)
         assert os.path.exists(new_path)
 
     def test_update_enable_locking(self, user_util):
         initial_name = self._update(user_util, enable_locking=True)
         repo_group = RepoGroupModel()._get_repo_group(initial_name)
         assert repo_group.enable_locking is True
 
     def test_update_description(self, user_util):
         description = 'New description'
         initial_name = self._update(user_util, description=description)
         repo_group = RepoGroupModel()._get_repo_group(initial_name)
         assert repo_group.group_description == description
 
     def test_update_owner(self, user_util):
         owner = self.TEST_USER_LOGIN
         initial_name = self._update(user_util, owner=owner)
         repo_group = RepoGroupModel()._get_repo_group(initial_name)
         assert repo_group.user.username == owner
 
     def test_update_group_name_conflict_with_existing(self, user_util):
         group_1 = user_util.create_repo_group()
         group_2 = user_util.create_repo_group()
         repo_group_name_1 = group_1.group_name
         repo_group_name_2 = group_2.group_name
 
         id_, params = build_data(
             self.apikey, 'update_repo_group', repogroupid=repo_group_name_1,
             group_name=repo_group_name_2)
         response = api_call(self.app, params)
         expected = {
             'unique_repo_group_name':
                 'Repository group with name `{}` already exists'.format(
                     repo_group_name_2)}
         assert_error(id_, expected, given=response.body)
 
     def test_api_update_repo_group_by_regular_user_no_permission(self, user_util):
         temp_user = user_util.create_user()
         temp_user_api_key = temp_user.api_key
         parent_group = user_util.create_repo_group()
         repo_group_name = parent_group.group_name
         id_, params = build_data(
             temp_user_api_key, 'update_repo_group', repogroupid=repo_group_name)
         response = api_call(self.app, params)
         expected = 'repository group `%s` does not exist' % (repo_group_name,)
         assert_error(id_, expected, given=response.body)
 
     def test_api_update_repo_group_regular_user_no_root_write_permissions(
             self, user_util):
         temp_user = user_util.create_user()
         temp_user_api_key = temp_user.api_key
         parent_group = user_util.create_repo_group(owner=temp_user.username)
         repo_group_name = parent_group.group_name
 
         id_, params = build_data(
             temp_user_api_key, 'update_repo_group', repogroupid=repo_group_name,
             group_name='at-root-level')
         response = api_call(self.app, params)
         expected = {
             'repo_group': 'You do not have the permission to store '
                           'repository groups in the root location.'}
         assert_error(id_, expected, given=response.body)
 
     def _update(self, user_util, **kwargs):
         repo_group = user_util.create_repo_group()
         initial_name = repo_group.name
         user = UserModel().get_by_username(self.TEST_USER_LOGIN)
         user_util.grant_user_permission_to_repo_group(
             repo_group, user, 'group.admin')
 
         id_, params = build_data(
             self.apikey, 'update_repo_group', repogroupid=initial_name,
             **kwargs)
         response = api_call(self.app, params)
 
         repo_group = RepoGroupModel.cls.get(repo_group.group_id)
 
         expected = {
             'msg': 'updated repository group ID:{} {}'.format(
                 repo_group.group_id, repo_group.group_name),
             'repo_group': {
                 'repositories': [],
                 'group_name': repo_group.group_name,
                 'group_description': repo_group.group_description,
                 'owner': repo_group.user.username,
                 'group_id': repo_group.group_id,
                 'parent_group': (
                     repo_group.parent_group.name
                     if repo_group.parent_group else None)
             }
         }
         assert_ok(id_, expected, given=response.body)
         return initial_name

rhodecode/api/tests/test_update_user.py

@@ -1,121 +1,120 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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.db import User
 from rhodecode.model.user import UserModel
 from rhodecode.tests import TEST_USER_ADMIN_LOGIN
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_ok, assert_error, crash, jsonify)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestUpdateUser(object):
     @pytest.mark.parametrize("name, expected", [
         ('firstname', 'new_username'),
         ('lastname', 'new_username'),
         ('email', 'new_username'),
         ('admin', True),
         ('admin', False),
         ('extern_type', 'ldap'),
         ('extern_type', None),
         ('extern_name', 'test'),
         ('extern_name', None),
         ('active', False),
         ('active', True),
         ('password', 'newpass'),
         ('description', 'CTO 4 Life')
     ])
     def test_api_update_user(self, name, expected, user_util):
         usr = user_util.create_user()
 
         kw = {name: expected, 'userid': usr.user_id}
         id_, params = build_data(self.apikey, 'update_user', **kw)
         response = api_call(self.app, params)
 
         ret = {
             'msg': 'updated user ID:%s %s' % (usr.user_id, usr.username),
             'user': jsonify(
                 UserModel()
                 .get_by_username(usr.username)
                 .get_api_data(include_secrets=True)
             )
         }
 
         expected = ret
         assert_ok(id_, expected, given=response.body)
 
     def test_api_update_user_no_changed_params(self):
         usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
         ret = jsonify(usr.get_api_data(include_secrets=True))
         id_, params = build_data(
             self.apikey, 'update_user', userid=TEST_USER_ADMIN_LOGIN)
 
         response = api_call(self.app, params)
         ret = {
             'msg': 'updated user ID:%s %s' % (
                 usr.user_id, TEST_USER_ADMIN_LOGIN),
             'user': ret
         }
         expected = ret
         expected['user']['last_activity'] = response.json['result']['user'][
             'last_activity']
         assert_ok(id_, expected, given=response.body)
 
     def test_api_update_user_by_user_id(self):
         usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
         ret = jsonify(usr.get_api_data(include_secrets=True))
         id_, params = build_data(
             self.apikey, 'update_user', userid=usr.user_id)
 
         response = api_call(self.app, params)
         ret = {
             'msg': 'updated user ID:%s %s' % (
                 usr.user_id, TEST_USER_ADMIN_LOGIN),
             'user': ret
         }
         expected = ret
         expected['user']['last_activity'] = response.json['result']['user'][
             'last_activity']
         assert_ok(id_, expected, given=response.body)
 
     def test_api_update_user_default_user(self):
         usr = User.get_default_user()
         id_, params = build_data(
             self.apikey, 'update_user', userid=usr.user_id)
 
         response = api_call(self.app, params)
         expected = 'editing default user is forbidden'
         assert_error(id_, expected, given=response.body)
 
     @mock.patch.object(UserModel, 'update_user', crash)
     def test_api_update_user_when_exception_happens(self):
         usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
         ret = jsonify(usr.get_api_data(include_secrets=True))
         id_, params = build_data(
             self.apikey, 'update_user', userid=usr.user_id)
 
         response = api_call(self.app, params)
         ret = 'failed to update user `%s`' % (usr.user_id,)
 
         expected = ret
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_update_user_group.py

@@ -1,125 +1,124 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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.tests import TEST_USER_ADMIN_EMAIL
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_error, assert_ok, crash, jsonify)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestUpdateUserGroup(object):
     @pytest.mark.parametrize("changing_attr, updates", [
         ('group_name', {'group_name': 'new_group_name'}),
         ('group_name', {'group_name': 'test_group_for_update'}),
         # ('owner', {'owner': TEST_USER_REGULAR_LOGIN}),
         ('owner_email', {'owner_email': TEST_USER_ADMIN_EMAIL}),
         ('active', {'active': False}),
         ('active', {'active': True}),
         ('sync', {'sync': False}),
         ('sync', {'sync': True})
     ])
     def test_api_update_user_group(self, changing_attr, updates, user_util):
         user_group = user_util.create_user_group()
         group_name = user_group.users_group_name
         expected_api_data = user_group.get_api_data()
         expected_api_data.update(updates)
 
         id_, params = build_data(
             self.apikey, 'update_user_group', usergroupid=group_name,
             **updates)
         response = api_call(self.app, params)
 
         # special case for sync
         if changing_attr == 'sync' and updates['sync'] is False:
             expected_api_data['sync'] = None
         elif changing_attr == 'sync' and updates['sync'] is True:
             expected_api_data['sync'] = 'manual_api'
 
         expected = {
             'msg': 'updated user group ID:%s %s' % (
                 user_group.users_group_id, user_group.users_group_name),
             'user_group': jsonify(expected_api_data)
         }
         assert_ok(id_, expected, given=response.body)
 
     @pytest.mark.parametrize("changing_attr, updates", [
         # TODO: mikhail: decide if we need to test against the commented params
         # ('group_name', {'group_name': 'new_group_name'}),
         # ('group_name', {'group_name': 'test_group_for_update'}),
         # ('owner', {'owner': TEST_USER_REGULAR_LOGIN}),
         ('owner_email', {'owner_email': TEST_USER_ADMIN_EMAIL}),
         ('active', {'active': False}),
         ('active', {'active': True}),
         ('sync', {'sync': False}),
         ('sync', {'sync': True})
     ])
     def test_api_update_user_group_regular_user(
             self, changing_attr, updates, user_util):
         user_group = user_util.create_user_group()
         group_name = user_group.users_group_name
         expected_api_data = user_group.get_api_data()
         expected_api_data.update(updates)
 
         # grant permission to this user
         user = UserModel().get_by_username(self.TEST_USER_LOGIN)
 
         user_util.grant_user_permission_to_user_group(
             user_group, user, 'usergroup.admin')
         id_, params = build_data(
             self.apikey_regular, 'update_user_group',
             usergroupid=group_name, **updates)
         response = api_call(self.app, params)
         # special case for sync
         if changing_attr == 'sync' and updates['sync'] is False:
             expected_api_data['sync'] = None
         elif changing_attr == 'sync' and updates['sync'] is True:
             expected_api_data['sync'] = 'manual_api'
 
         expected = {
             'msg': 'updated user group ID:%s %s' % (
                 user_group.users_group_id, user_group.users_group_name),
             'user_group': jsonify(expected_api_data)
         }
         assert_ok(id_, expected, given=response.body)
 
     def test_api_update_user_group_regular_user_no_permission(self, user_util):
         user_group = user_util.create_user_group()
         group_name = user_group.users_group_name
         id_, params = build_data(
             self.apikey_regular, 'update_user_group', usergroupid=group_name)
         response = api_call(self.app, params)
 
         expected = 'user group `%s` does not exist' % (group_name)
         assert_error(id_, expected, given=response.body)
 
     @mock.patch.object(UserGroupModel, 'update', crash)
     def test_api_update_user_group_exception_occurred(self, user_util):
         user_group = user_util.create_user_group()
         group_name = user_group.users_group_name
         id_, params = build_data(
             self.apikey, 'update_user_group', usergroupid=group_name)
         response = api_call(self.app, params)
         expected = 'failed to update user group `%s`' % (group_name,)
         assert_error(id_, expected, given=response.body)

rhodecode/api/tests/test_utils.py

@@ -1,295 +1,297 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 from mock import Mock, patch
 
 from rhodecode.api import utils
 from rhodecode.api import JSONRPCError
 from rhodecode.lib.vcs.exceptions import RepositoryError
 
 
 class TestGetCommitOrError(object):
-    def setup(self):
+
+    def setup_method(self):
         self.commit_hash = 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa10'
 
     @pytest.mark.parametrize("ref", ['ref', '12345', 'a:b:c:d', 'branch:name'])
     def test_ref_cannot_be_parsed(self, ref):
         repo = Mock()
         with pytest.raises(JSONRPCError) as excinfo:
             utils.get_commit_or_error(ref, repo)
         expected_message = (
             'Ref `{ref}` given in a wrong format. Please check the API'
             ' documentation for more details'.format(ref=ref)
         )
         assert excinfo.value.message == expected_message
 
     def test_success_with_hash_specified(self):
         repo = Mock()
         ref_type = 'branch'
         ref = '{}:master:{}'.format(ref_type, self.commit_hash)
 
         with patch('rhodecode.api.utils.get_commit_from_ref_name') as get_commit:
             result = utils.get_commit_or_error(ref, repo)
             get_commit.assert_called_once_with(
                 repo, self.commit_hash)
             assert result == get_commit()
 
     def test_raises_an_error_when_commit_not_found(self):
         repo = Mock()
         ref = 'branch:master:{}'.format(self.commit_hash)
 
         with patch('rhodecode.api.utils.get_commit_from_ref_name') as get_commit:
             get_commit.side_effect = RepositoryError('Commit not found')
             with pytest.raises(JSONRPCError) as excinfo:
                 utils.get_commit_or_error(ref, repo)
             expected_message = 'Ref `{}` does not exist'.format(ref)
             assert excinfo.value.message == expected_message
 
 
 class TestResolveRefOrError(object):
-    def setup(self):
+
+    def setup_method(self):
         self.commit_hash = 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa10'
 
     def test_success_with_no_hash_specified(self):
         repo = Mock()
         ref_type = 'branch'
         ref_name = 'master'
         ref = '{}:{}'.format(ref_type, ref_name)
 
         with patch('rhodecode.api.utils._get_ref_hash') \
                 as _get_ref_hash:
             _get_ref_hash.return_value = self.commit_hash
             result = utils.resolve_ref_or_error(ref, repo)
             _get_ref_hash.assert_called_once_with(repo, ref_type, ref_name)
             assert result == '{}:{}'.format(ref, self.commit_hash)
 
     def test_non_supported_refs(self):
         repo = Mock()
         ref = 'bookmark:ref'
         with pytest.raises(JSONRPCError) as excinfo:
             utils.resolve_ref_or_error(ref, repo)
         expected_message = (
             'The specified value:bookmark:`ref` does not exist, or is not allowed.')
         assert excinfo.value.message == expected_message
 
     def test_branch_is_not_found(self):
         repo = Mock()
         ref = 'branch:non-existing-one'
         with patch('rhodecode.api.utils._get_ref_hash')\
                 as _get_ref_hash:
             _get_ref_hash.side_effect = KeyError()
             with pytest.raises(JSONRPCError) as excinfo:
                 utils.resolve_ref_or_error(ref, repo)
         expected_message = (
             'The specified value:branch:`non-existing-one` does not exist, or is not allowed.')
         assert excinfo.value.message == expected_message
 
     def test_bookmark_is_not_found(self):
         repo = Mock()
         ref = 'bookmark:non-existing-one'
         with patch('rhodecode.api.utils._get_ref_hash')\
                 as _get_ref_hash:
             _get_ref_hash.side_effect = KeyError()
             with pytest.raises(JSONRPCError) as excinfo:
                 utils.resolve_ref_or_error(ref, repo)
         expected_message = (
             'The specified value:bookmark:`non-existing-one` does not exist, or is not allowed.')
         assert excinfo.value.message == expected_message
 
     @pytest.mark.parametrize("ref", ['ref', '12345', 'a:b:c:d'])
     def test_ref_cannot_be_parsed(self, ref):
         repo = Mock()
         with pytest.raises(JSONRPCError) as excinfo:
             utils.resolve_ref_or_error(ref, repo)
         expected_message = (
             'Ref `{ref}` given in a wrong format. Please check the API'
             ' documentation for more details'.format(ref=ref)
         )
         assert excinfo.value.message == expected_message
 
 
 class TestGetRefHash(object):
-    def setup(self):
+
+    def setup_method(self):
         self.commit_hash = 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa10'
         self.bookmark_name = 'test-bookmark'
 
     @pytest.mark.parametrize("alias, branch_name", [
         ("git", "master"),
         ("hg", "default")
     ])
     def test_returns_hash_by_branch_name(self, alias, branch_name):
         with patch('rhodecode.model.db.Repository') as repo:
             repo.scm_instance().alias = alias
             repo.scm_instance().branches = {branch_name: self.commit_hash}
             result_hash = utils._get_ref_hash(repo, 'branch', branch_name)
             assert result_hash == self.commit_hash
 
     @pytest.mark.parametrize("alias, branch_name", [
         ("git", "master"),
         ("hg", "default")
     ])
     def test_raises_error_when_branch_is_not_found(self, alias, branch_name):
         with patch('rhodecode.model.db.Repository') as repo:
             repo.scm_instance().alias = alias
             repo.scm_instance().branches = {}
             with pytest.raises(KeyError):
                 utils._get_ref_hash(repo, 'branch', branch_name)
 
     def test_returns_hash_when_bookmark_is_specified_for_hg(self):
         with patch('rhodecode.model.db.Repository') as repo:
             repo.scm_instance().alias = 'hg'
             repo.scm_instance().bookmarks = {
                 self.bookmark_name: self.commit_hash}
             result_hash = utils._get_ref_hash(
                 repo, 'bookmark', self.bookmark_name)
             assert result_hash == self.commit_hash
 
     def test_raises_error_when_bookmark_is_not_found_in_hg_repo(self):
         with patch('rhodecode.model.db.Repository') as repo:
             repo.scm_instance().alias = 'hg'
             repo.scm_instance().bookmarks = {}
             with pytest.raises(KeyError):
                 utils._get_ref_hash(repo, 'bookmark', self.bookmark_name)
 
     def test_raises_error_when_bookmark_is_specified_for_git(self):
         with patch('rhodecode.model.db.Repository') as repo:
             repo.scm_instance().alias = 'git'
             repo.scm_instance().bookmarks = {
                 self.bookmark_name: self.commit_hash}
             with pytest.raises(ValueError):
                 utils._get_ref_hash(repo, 'bookmark', self.bookmark_name)
 
 
 class TestUserByNameOrError(object):
     def test_user_found_by_id(self):
         fake_user = Mock(id=123)
 
         patcher = patch('rhodecode.model.user.UserModel.get_user')
         with patcher as get_user:
             get_user.return_value = fake_user
 
             patcher = patch('rhodecode.model.user.UserModel.get_by_username')
             with patcher as get_by_username:
                 result = utils.get_user_or_error(123)
                 assert result == fake_user
 
     def test_user_not_found_by_id_as_str(self):
         fake_user = Mock(id=123)
 
         patcher = patch('rhodecode.model.user.UserModel.get_user')
         with patcher as get_user:
             get_user.return_value = fake_user
             patcher = patch('rhodecode.model.user.UserModel.get_by_username')
             with patcher as get_by_username:
                 get_by_username.return_value = None
 
                 with pytest.raises(JSONRPCError):
                     utils.get_user_or_error('123')
 
     def test_user_found_by_name(self):
         fake_user = Mock(id=123)
 
         patcher = patch('rhodecode.model.user.UserModel.get_user')
         with patcher as get_user:
             get_user.return_value = None
 
             patcher = patch('rhodecode.model.user.UserModel.get_by_username')
             with patcher as get_by_username:
                 get_by_username.return_value = fake_user
 
                 result = utils.get_user_or_error('test')
                 assert result == fake_user
 
     def test_user_not_found_by_id(self):
         patcher = patch('rhodecode.model.user.UserModel.get_user')
         with patcher as get_user:
             get_user.return_value = None
             patcher = patch('rhodecode.model.user.UserModel.get_by_username')
             with patcher as get_by_username:
                 get_by_username.return_value = None
 
                 with pytest.raises(JSONRPCError) as excinfo:
                     utils.get_user_or_error(123)
 
                 expected_message = 'user `123` does not exist'
                 assert excinfo.value.message == expected_message
 
     def test_user_not_found_by_name(self):
         patcher = patch('rhodecode.model.user.UserModel.get_by_username')
         with patcher as get_by_username:
             get_by_username.return_value = None
             with pytest.raises(JSONRPCError) as excinfo:
                 utils.get_user_or_error('test')
 
         expected_message = 'user `test` does not exist'
         assert excinfo.value.message == expected_message
 
 
 class TestGetCommitDict(object):
     @pytest.mark.parametrize('filename, expected', [
         (b'sp\xc3\xa4cial', u'sp\xe4cial'),
         (b'sp\xa4cial', u'sp\ufffdcial'),
     ])
     def test_decodes_filenames_to_unicode(self, filename, expected):
         result = utils._get_commit_dict(filename=filename, op='A')
         assert result['filename'] == expected
 
 
 class TestRepoAccess(object):
     def setup_method(self, method):
 
         self.admin_perm_patch = patch(
             'rhodecode.api.utils.HasPermissionAnyApi')
         self.repo_perm_patch = patch(
             'rhodecode.api.utils.HasRepoPermissionAnyApi')
 
     def test_has_superadmin_permission_checks_for_admin(self):
         admin_mock = Mock()
         with self.admin_perm_patch as amock:
             amock.return_value = admin_mock
             assert utils.has_superadmin_permission('fake_user')
             amock.assert_called_once_with('hg.admin')
 
         admin_mock.assert_called_once_with(user='fake_user')
 
     def test_has_repo_permissions_checks_for_repo_access(self):
         repo_mock = Mock()
         fake_repo = Mock()
         with self.repo_perm_patch as rmock:
             rmock.return_value = repo_mock
             assert utils.validate_repo_permissions(
                 'fake_user', 'fake_repo_id', fake_repo,
                 ['perm1', 'perm2'])
             rmock.assert_called_once_with(*['perm1', 'perm2'])
 
         repo_mock.assert_called_once_with(
             user='fake_user', repo_name=fake_repo.repo_name)
 
     def test_has_repo_permissions_raises_not_found(self):
         repo_mock = Mock(return_value=False)
         fake_repo = Mock()
         with self.repo_perm_patch as rmock:
             rmock.return_value = repo_mock
             with pytest.raises(JSONRPCError) as excinfo:
                 utils.validate_repo_permissions(
                     'fake_user', 'fake_repo_id', fake_repo, 'perms')
                 assert 'fake_repo_id' in excinfo

rhodecode/api/tests/utils.py

@@ -1,124 +1,123 b''
-# -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 random
 import pytest
 
 from rhodecode.api.utils import get_origin
 from rhodecode.lib.ext_json import json
 
 
 def jsonify(obj):
     return json.loads(json.dumps(obj))
 
 
 API_URL = '/_admin/api'
 
 
 def assert_call_ok(id_, given):
     expected = jsonify({
         'id': id_,
         'error': None,
         'result': None
     })
     given = json.loads(given)
 
     assert expected['id'] == given['id']
     assert expected['error'] == given['error']
     return given['result']
 
 
 def assert_ok(id_, expected, given):
     given = json.loads(given)
     if given.get('error'):
         err = given['error']
-        pytest.fail(u"Unexpected ERROR in success response: {}".format(err))
+        pytest.fail(f"Unexpected ERROR in expected success response: `{err}`")
 
     expected = jsonify({
         'id': id_,
         'error': None,
         'result': expected
     })
 
     assert expected == given
 
 
 def assert_error(id_, expected, given):
     expected = jsonify({
         'id': id_,
         'error': expected,
         'result': None
     })
     given = json.loads(given)
     assert expected == given
 
 
 def build_data(apikey, method, **kw):
     """
     Builds API data with given random ID
     """
     random_id = random.randrange(1, 9999)
     return random_id, json.dumps({
         "id": random_id,
         "api_key": apikey,
         "method": method,
         "args": kw
     })
 
 
 def api_call(app, params, status=None):
     response = app.post(
         API_URL, content_type='application/json', params=params, status=status,
         headers=[('Content-Type', 'application/json')])
     return response
 
 
 def crash(*args, **kwargs):
     raise Exception('Total Crash !')
 
 
 def expected_permissions(object_with_permissions):
     """
     Returns the expected permissions structure for the given object.
 
     The object is expected to be a `Repository`, `RepositoryGroup`,
     or `UserGroup`. They all implement the same permission handling
     API.
     """
     permissions = []
     for _user in object_with_permissions.permissions():
         user_data = {
             'name': _user.username,
             'permission': _user.permission,
             'origin': get_origin(_user),
             'type': "user",
         }
         permissions.append(user_data)
 
     for _user_group in object_with_permissions.permission_user_groups():
         user_group_data = {
             'name': _user_group.users_group_name,
             'permission': _user_group.permission,
             'origin': get_origin(_user_group),
             'type': "user_group",
         }
         permissions.append(user_group_data)
     return permissions

rhodecode/api/views/search_api.py

@@ -1,152 +1,152 b''
-# -*- coding: utf-8 -*-
+
 
 # Copyright (C) 2011-2020 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 logging
 
 from rhodecode.api import jsonrpc_method
 from rhodecode.api.exc import JSONRPCValidationError, JSONRPCForbidden
 from rhodecode.api.utils import Optional, has_superadmin_permission
 from rhodecode.lib.index import searcher_from_config
 from rhodecode.lib.user_log_filter import user_log_filter
 from rhodecode.model import validation_schema
 from rhodecode.model.db import joinedload, UserLog
 from rhodecode.model.validation_schema.schemas import search_schema
 
 log = logging.getLogger(__name__)
 
 
 @jsonrpc_method()
 def search(request, apiuser, search_query, search_type, page_limit=Optional(10),
            page=Optional(1), search_sort=Optional('desc:date'),
            repo_name=Optional(None), repo_group_name=Optional(None)):
     """
     Fetch Full Text Search results using API.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param search_query: Search query.
     :type search_query: str
     :param search_type: Search type. The following are valid options:
         * commit
         * content
         * path
     :type search_type: str
     :param page_limit: Page item limit, from 1 to 500. Default 10 items.
     :type page_limit: Optional(int)
     :param page: Page number. Default first page.
     :type page: Optional(int)
     :param search_sort: Search sort order.Must start with asc: or desc: Default desc:date.
         The following are valid options:
         * asc|desc:message.raw
         * asc|desc:date
         * asc|desc:author.email.raw
         * asc|desc:message.raw
         * newfirst (old legacy equal to desc:date)
         * oldfirst (old legacy equal to asc:date)
 
     :type search_sort: Optional(str)
     :param repo_name: Filter by one repo. Default is all.
     :type repo_name: Optional(str)
     :param repo_group_name: Filter by one repo group. Default is all.
     :type repo_group_name: Optional(str)
     """
 
     data = {'execution_time': ''}
     repo_name = Optional.extract(repo_name)
     repo_group_name = Optional.extract(repo_group_name)
 
     schema = search_schema.SearchParamsSchema()
 
     try:
         search_params = schema.deserialize(
             dict(search_query=search_query,
                  search_type=search_type,
                  search_sort=Optional.extract(search_sort),
                  page_limit=Optional.extract(page_limit),
                  requested_page=Optional.extract(page))
         )
     except validation_schema.Invalid as err:
         raise JSONRPCValidationError(colander_exc=err)
 
     search_query = search_params.get('search_query')
     search_type = search_params.get('search_type')
     search_sort = search_params.get('search_sort')
 
     if search_params.get('search_query'):
         page_limit = search_params['page_limit']
         requested_page = search_params['requested_page']
 
         searcher = searcher_from_config(request.registry.settings)
 
         try:
             search_result = searcher.search(
                 search_query, search_type, apiuser, repo_name, repo_group_name,
                 requested_page=requested_page, page_limit=page_limit, sort=search_sort)
 
             data.update(dict(
                 results=list(search_result['results']), page=requested_page,
                 item_count=search_result['count'],
                 items_per_page=page_limit))
         finally:
             searcher.cleanup()
 
         if not search_result['error']:
             data['execution_time'] = '%s results (%.4f seconds)' % (
                 search_result['count'],
                 search_result['runtime'])
         else:
             node = schema['search_query']
             raise JSONRPCValidationError(
                 colander_exc=validation_schema.Invalid(node, search_result['error']))
 
     return data
 
 
 @jsonrpc_method()
 def get_audit_logs(request, apiuser, query):
     """
     return full audit logs based on the query.
 
     Please see `example query in admin > settings > audit logs` for examples
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param query: filter query, example: action:repo.artifact.add date:[20200401 TO 20200601]"
     :type query: str
     """
 
     if not has_superadmin_permission(apiuser):
         raise JSONRPCForbidden()
 
     filter_term = query
     ret = []
 
     # show all user actions
     user_log = UserLog.query() \
         .options(joinedload(UserLog.user)) \
         .options(joinedload(UserLog.repository)) \
         .order_by(UserLog.action_date.desc())
 
     audit_log = user_log_filter(user_log, filter_term)
 
     for entry in audit_log:
         ret.append(entry)
     return ret