From a7ec7210086b219130dbb80da1b257cd3a1fe360 Mon Sep 17 00:00:00 2001 From: Pierre-Yves Chibon Date: Dec 19 2019 11:58:53 +0000 Subject: [PATCH 1/3] Move the tests to their own folder This way, all the tests can be ran at once, instead of having to call each file separately. Signed-off-by: Pierre-Yves Chibon --- diff --git a/bugzilla-override-tests.py b/bugzilla-override-tests.py deleted file mode 100644 index 138ebcb..0000000 --- a/bugzilla-override-tests.py +++ /dev/null @@ -1,198 +0,0 @@ -from __future__ import print_function - -import os -import json - -# These are the tests from the pagure/ git repo. -# Run with: -# PYTHONPATH=.:/path/to/pagure/checkout nosetests dist_git_auth_tests.py -import tests - -from pagure_distgit import plugin - - -def setUp(): - tests.setUp() - - -def tearDown(): - tests.tearDown() - - -class PagureFlaskApiProjectBZOverrideTests(tests.Modeltests): - def setUp(self): - """ Set up the environnment, ran before every tests. """ - super(PagureFlaskApiProjectBZOverrideTests, self).setUp() - self.session.flush() - tests.create_projects(self.session) - tests.create_projects_git(os.path.join(self.path, "repos"), bare=True) - tests.create_tokens(self.session) - tests.create_tokens_acl(self.session) - self._app.register_blueprint(plugin.DISTGIT_NS) - - def test_override_endpoint(self): - """Test bugzilla overrides. """ - - output = self.app.get("/api/0/somenamespace/test3") - self.assertEqual(output.status_code, 200) - output = self.app.get("/_dg/bzoverrides/somenamespace/test3") - self.assertEqual(output.status_code, 200) - data = json.loads(output.get_data(as_text=True)) - expected_result = {"epel_assignee": "pingou", "fedora_assignee": "pingou"} - data = json.loads(output.get_data(as_text=True)) - self.assertDictEqual(data, expected_result) - - headers = {"Authorization": "token foo_token"} - datainput = { - "epel_assignee": "foo", - "fedora_assignee": "pingou", - } - output = self.app.post( - "/_dg/bzoverrides/somenamespace/test3", data=datainput, headers=headers - ) - # invalid token - self.assertEqual(output.status_code, 401) - - headers = {"Authorization": "token BBBZZZOOO"} - output = self.app.post( - "/_dg/bzoverrides/somenamespace/test3", data=datainput, headers=headers - ) - self.assertEqual(output.status_code, 401) - - output = self.app.get("/_dg/bzoverrides/somenamespace/test3") - self.assertEqual(output.status_code, 200) - data = json.loads(output.get_data(as_text=True)) - self.assertDictEqual(data, expected_result) - - # Change assignees with a valid token - headers = {"Authorization": "token aaabbbcccddd"} - output = self.app.post( - "/_dg/bzoverrides/somenamespace/test3", data=datainput, headers=headers - ) - self.assertEqual(output.status_code, 200) - expected_result = {"epel_assignee": "foo", "fedora_assignee": "pingou"} - output = self.app.get("/_dg/bzoverrides/somenamespace/test3") - data = json.loads(output.get_data(as_text=True)) - self.assertDictEqual(data, expected_result) - - # set both assignees to foo - datainput = {"epel_assignee": "foo", "fedora_assignee": "foo"} - output = self.app.post( - "/_dg/bzoverrides/somenamespace/test3", data=datainput, headers=headers - ) - self.assertEqual(output.status_code, 200) - output = self.app.get("/_dg/bzoverrides/somenamespace/test3") - data = json.loads(output.get_data(as_text=True)) - self.assertDictEqual(data, datainput) - - # Change both assignees back to default - datainput = {"epel_assignee": "", "fedora_assignee": None} - expected_result = {"epel_assignee": "pingou", "fedora_assignee": "pingou"} - output = self.app.post( - "/_dg/bzoverrides/somenamespace/test3", data=datainput, headers=headers - ) - self.assertEqual(output.status_code, 200) - output = self.app.get("/_dg/bzoverrides/somenamespace/test3") - data = json.loads(output.get_data(as_text=True)) - self.assertDictEqual(data, expected_result) - - # Change only one assignee, then change assignee back to default - expected_result = {"epel_assignee": "pingou", "fedora_assignee": "pingou"} - datainput = {"epel_assignee": "foo", "fedora_assignee": "pingou"} - output = self.app.post( - "/_dg/bzoverrides/somenamespace/test3", data=datainput, headers=headers - ) - self.assertEqual(output.status_code, 200) - datainput = {"epel_assignee": "", "fedora_assignee": "pingou"} - output = self.app.post( - "/_dg/bzoverrides/somenamespace/test3", data=datainput, headers=headers - ) - self.assertEqual(output.status_code, 200) - output = self.app.get("/_dg/bzoverrides/somenamespace/test3") - data = json.loads(output.get_data(as_text=True)) - self.assertDictEqual(data, expected_result) - - user = tests.FakeUser(username="pingou") - with tests.user_set(self.app.application, user): - # change one assignee - expected_result = {"epel_assignee": "foo", "fedora_assignee": "pingou"} - datainput = {"epel_assignee": "foo", "fedora_assignee": "pingou"} - output = self.app.post( - "/_dg/bzoverrides/somenamespace/test3", data=datainput - ) - self.assertEqual(output.status_code, 200) - data = json.loads(output.get_data(as_text=True)) - self.assertDictEqual(data, expected_result) - - # change one assignee - expected_result = {"epel_assignee": "foo", "fedora_assignee": "foo"} - datainput = {"epel_assignee": "foo","fedora_assignee": "foo"} - output = self.app.post( - "/_dg/bzoverrides/somenamespace/test3", data=datainput - ) - self.assertEqual(output.status_code, 200) - data = json.loads(output.get_data(as_text=True)) - self.assertDictEqual(data, expected_result) - - # reset assignees - expected_result = {"epel_assignee": "pingou", "fedora_assignee": "pingou"} - datainput = {"epel_assignee": None, "fedora_assignee": ""} - output = self.app.post( - "/_dg/bzoverrides/somenamespace/test3", data=datainput - ) - self.assertEqual(output.status_code, 200) - data = json.loads(output.get_data(as_text=True)) - self.assertDictEqual(data, expected_result) - - # change both assignees to foo, then change only one back - # As the other is considered empty, both should be back to pingou - expected_result = {"epel_assignee": "foo", "fedora_assignee": "foo"} - datainput = {"epel_assignee": "foo","fedora_assignee": "foo"} - output = self.app.post( - "/_dg/bzoverrides/somenamespace/test3", data=datainput - ) - self.assertEqual(output.status_code, 200) - data = json.loads(output.get_data(as_text=True)) - self.assertDictEqual(data, expected_result) - expected_result = {"epel_assignee": "pingou", "fedora_assignee": "pingou"} - datainput = {"epel_assignee": "pingou"} - output = self.app.post( - "/_dg/bzoverrides/somenamespace/test3", data=datainput - ) - self.assertEqual(output.status_code, 200) - data = json.loads(output.get_data(as_text=True)) - self.assertDictEqual(data, expected_result) - - # Invalid datainput resets to default - datainput = {"epel_assignee": "foo","fedora_assignee": "foo"} - output = self.app.post( - "/_dg/bzoverrides/somenamespace/test3", data=datainput - ) - self.assertEqual(output.status_code, 200) - datainput = {} # This is invalid input - output = self.app.post( - "/_dg/bzoverrides/somenamespace/test3", data=datainput - ) - self.assertEqual(output.status_code, 200) - data = json.loads(output.get_data(as_text=True)) - self.assertDictEqual(data, expected_result) - - datainput = {"epel_assignee": "foo","fedora_assignee": "foo"} - output = self.app.post( - "/_dg/bzoverrides/somenamespace/test3", data=datainput - ) - self.assertEqual(output.status_code, 200) - datainput = None # Invalid input - output = self.app.post( - "/_dg/bzoverrides/somenamespace/test3", data=datainput - ) - data = json.loads(output.get_data(as_text=True)) - self.assertEqual(output.status_code, 200) - - user = tests.FakeUser(username="foo") - with tests.user_set(self.app.application, user): - expected_result = {"epel_assignee": "pingou", "fedora_assignee": "pingou"} - output = self.app.post( - "/_dg/bzoverrides/somenamespace/test3", data=expected_result - ) - self.assertEqual(output.status_code, 401) diff --git a/dist_git_auth_tests.py b/dist_git_auth_tests.py deleted file mode 100644 index e52b99a..0000000 --- a/dist_git_auth_tests.py +++ /dev/null @@ -1,778 +0,0 @@ -from __future__ import print_function - -import tempfile -import os - -from mock import Mock, patch - -# These are the tests from the pagure/ git repo. -# Run with: -# PYTHONPATH=.:/path/to/pagure/checkout nosetests dist_git_auth_tests.py -import pagure.config -import pagure.exceptions -import pagure.lib.model -try: - from pagure.lib import _get_project as get_project -except ImportError: - # From pagure 5.2, code has been moved to pagure.lib.query - from pagure.lib.query import _get_project as get_project - -import tests - -import dist_git_auth - - -def setUp(): - tests.setUp() - - -def tearDown(): - tests.tearDown() - - -def patch_pdc(values): - """ Decorator to patch the PDC calls to return values for this test. - - Args: - values (dict): A dictionary where the keys are project fullnames - (i.e. namespace/name), and the values are dicts with key branch - names, and values their supported status. - Note that as namespace, the PDC "type" is used, which lacks the "s" - for rpms and modules. - e.g.: {'rpm/test': {'f28': True, 'f27': False}} - """ - def pdc_get_paged(_, global_component, type, name, fields): - """ Function that emulates the pdc.get_paged call - - Args as provided by dist_git_auth's calls to it. - - Args: - _ (anything): PDCClient internal - global_component (string): package name - type (string): The PDC "type": "rpm", "module", ... - name (string): The branch name - fields (list): Always ["active"] - """ - fullname = '%s/%s' % (type, global_component) - if fullname not in values: - return [] - val = values[fullname] - if name not in val: - return [] - val = val[name] - if val in (True, False): - return [{"active": val}] - # This case is used to emulate "weird" results - return val - - def decorator(func): - def test_wrapper(*args, **kwargs): - with patch.object(dist_git_auth.PDCClient, '__getitem__'): - with patch.object(dist_git_auth.PDCClient, 'get_paged', - side_effect=pdc_get_paged): - return func(*args, **kwargs) - return test_wrapper - return decorator - - -class DistGitAuthTests(tests.Modeltests): - """ Test DistGitAuth ACLs with Fedora config. """ - - maxDiff = None - - def setUp(self): - """ Set up the environment in which to run the tests. """ - super(DistGitAuthTests, self).setUp() - - pagure.config.config['ACL_DEBUG'] = True - pagure.config.config['EXTERNAL_COMMITTER'] = {"relenggroup": {}} - pagure.config.config.update(self.dga_config) - - self.dga = dist_git_auth.DistGitAuth() - self.dga.info = Mock(side_effect=print) - # We default to saying it's not a forced push - dist_git_auth.is_forced_push = Mock(return_value=False) - - # Create an RCM user/group - rcmuser = pagure.lib.model.User( - user='releng', - fullname='Release Engineering', - token='aaabbbcd', - default_email='rcm@local.local', - ) - self.session.add(rcmuser) - self.session.flush() - rcmgroup = pagure.lib.model.PagureGroup( - group_name='relenggroup', - group_type='user', - display_name='Releng group', - user_id=rcmuser.id, - ) - self.session.add(rcmgroup) - self.session.flush() - rcmuser.group_objs.append(rcmgroup) - self.session.commit() - - def tearDown(self): - """ Tear down the environment in which the tests ran. """ - self.dga = None - super(DistGitAuthTests, self).tearDown() - - def create_namespaced_project(self, namespace, name, is_fork=False, settings={}): - item = pagure.lib.model.Project( - user_id=1, # pingou - name=name, - is_fork=is_fork, - parent_id=3 if is_fork else None, - description='namespaced test project', - hook_token='aaabbbeee', - namespace=namespace, - settings=settings - ) - item.close_status = [ - 'Invalid', 'Insufficient data', 'Fixed', 'Duplicate'] - self.session.add(item) - self.session.commit() - return get_project( - self.session, name=name, namespace=namespace) - - def expect_info_msg(self, expect_msg): - found = False - for call in self.dga.info.call_args_list: - args = call[0] - msg = args[0] - if msg == expect_msg: - found = True - if not found: - raise AssertionError( - "Info message '%s' expected but not found" % expect_msg) - - -class DistGitAuthTestsGeneric(DistGitAuthTests): - dga_config = {} - - def test_unused_repotype(self): - self.assertFalse( - self.dga.check_acl( - self.session, - project=None, - username=None, - refname=None, - pull_request=None, - repodir=None, - repotype='tickets', - revfrom=None, - revto=None, - is_internal=False, - ) - ) - - self.expect_info_msg("Repotype tickets not in use") - - def test_branch_deletion(self): - dist_git_auth.is_forced_push.return_value = True - - project = self.create_namespaced_project('rpms', 'test') - - self.assertFalse( - self.dga.check_acl( - self.session, - project=project, - username=None, - refname=None, - pull_request=None, - repodir=None, - repotype='main', - revfrom=None, - revto='0000000000000000000000000000000000000000', - is_internal=False, - ) - ) - - self.expect_info_msg("Branch deletion is not allowed") - - def test_forced_push(self): - dist_git_auth.is_forced_push.return_value = True - - project = self.create_namespaced_project('rpms', 'test') - - self.assertFalse( - self.dga.check_acl( - self.session, - project=project, - username=None, - refname=None, - pull_request=None, - repodir=None, - repotype='main', - revfrom=None, - revto=None, - is_internal=False, - ) - ) - - self.expect_info_msg("Forced pushes are not allowed") - - def test_internal(self): - project = self.create_namespaced_project('rpms', 'test') - - self.assertTrue( - self.dga.check_acl( - self.session, - project=project, - username=None, - refname=None, - pull_request=None, - repodir=None, - repotype='main', - revfrom=None, - revto=None, - is_internal=True, - ) - ) - - self.expect_info_msg("Internal push allowed") - - def test_deploykey(self): - project = self.create_namespaced_project('rpms', 'test') - - self.assertFalse( - self.dga.check_acl( - self.session, - project=project, - username="deploykey_foobar", - refname=None, - pull_request=None, - repodir=None, - repotype='main', - revfrom=None, - revto=None, - is_internal=False, - ) - ) - - self.expect_info_msg("Deploy keys are disabled") - - def test_invalid_user(self): - project = self.create_namespaced_project('rpms', 'test') - - self.assertRaises( - pagure.exceptions.PagureException, - self.dga.check_acl, - self.session, - project=project, - username="nosuchuser", - refname=None, - pull_request=None, - repodir=None, - repotype='main', - revfrom=None, - revto=None, - is_internal=False, - ) - - def test_unprotected_committer(self): - project = self.create_namespaced_project('unprotected', 'test') - - self.assertTrue( - self.dga.check_acl( - self.session, - project=project, - username="pingou", - refname="refs/heads/mywip", - pull_request=None, - repodir=None, - repotype='main', - revfrom=None, - revto=None, - is_internal=False, - ) - ) - - self.expect_info_msg("Committer push") - - def test_unprotected_non_committer(self): - project = self.create_namespaced_project('unprotected', 'test') - - self.assertFalse( - self.dga.check_acl( - self.session, - project=project, - username="foo", - refname="refs/heads/mywip", - pull_request=None, - repodir=None, - repotype='main', - revfrom=None, - revto=None, - is_internal=False, - ) - ) - - self.expect_info_msg("Fall-through deny") - - def test_unprotected_pr_required_pr(self): - project = self.create_namespaced_project('unprotected', 'test') - self.dga.global_pr_only = True - - self.assertTrue( - self.dga.check_acl( - self.session, - project=project, - username="pingou", - refname="refs/heads/mywip", - pull_request=True, - repodir=None, - repotype='main', - revfrom=None, - revto=None, - is_internal=False, - ) - ) - - self.expect_info_msg("Committer push") - - def test_unprotected_pr_required_no_pr(self): - project = self.create_namespaced_project('unprotected', 'test') - self.dga.global_pr_only = True - - self.assertFalse( - self.dga.check_acl( - self.session, - project=project, - username="pingou", - refname="refs/heads/mywip", - pull_request=None, - repodir=None, - repotype='main', - revfrom=None, - revto=None, - is_internal=False, - ) - ) - - self.expect_info_msg("A pull request is required for this branch") - - def test_unprotected_pr_required_no_pr_cvsadmin(self): - project = self.create_namespaced_project('unprotected', 'test') - self.dga.global_pr_only = True - - self.assertTrue( - self.dga.check_acl( - self.session, - project=project, - username="releng", - refname="refs/heads/mywip", - pull_request=None, - repodir=None, - repotype='main', - revfrom=None, - revto=None, - is_internal=False, - ) - ) - - self.expect_info_msg("Committer push") - - def test_unprotected_pr_required_requests(self): - project = self.create_namespaced_project('unprotected', 'test') - self.dga.global_pr_only = True - - self.assertTrue( - self.dga.check_acl( - self.session, - project=project, - username="pingou", - refname="refs/heads/mywip", - pull_request=None, - repodir=None, - repotype='requests', - revfrom=None, - revto=None, - is_internal=False, - ) - ) - - self.expect_info_msg("Committer push") - - def test_unprotected_pr_required_repo_pr_only_no_pr(self): - settings = {"pull_request_access_only": True} - project = self.create_namespaced_project('unprotected', 'test', settings=settings) - - self.assertFalse( - self.dga.check_acl( - self.session, - project=project, - username="pingou", - refname="refs/heads/mywip", - pull_request=None, - repodir=None, - repotype='main', - revfrom=None, - revto=None, - is_internal=False, - ) - ) - - self.expect_info_msg("A pull request is required for this branch") - - def test_unprotected_pr_required_repo_pr_only(self): - settings = {"pull_request_access_only": True} - project = self.create_namespaced_project( - 'unprotected', - 'test', - settings=settings - ) - self.assertTrue( - self.dga.check_acl( - self.session, - project=project, - username="pingou", - refname="refs/heads/mywip", - pull_request="pull_request", - repodir=None, - repotype='main', - revfrom=None, - revto=None, - is_internal=False, - ) - ) - - self.expect_info_msg("Committer push") - - -class DistGitAuthTestsFedora(DistGitAuthTests): - dga_config = { - 'PR_ONLY': False, - 'ACL_BLOCK_UNSPECIFIED': False, - 'BLACKLIST_RES': ['refs/heads/c[0-9]+.*'], - 'UNSPECIFIED_BLACKLIST_RES': [ - 'refs/heads/f[0-9]+', - ], - 'RCM_GROUP': 'relenggroup', - 'RCM_BRANCHES': ['refs/heads/f[0-9]+'], - 'ACL_PROTECTED_NAMESPACES': ['rpms', 'modules', 'container'], - 'PDC_URL': 'invalid://', - } - - def test_protected_blacklisted_ref(self): - project = self.create_namespaced_project('rpms', 'test') - - self.assertFalse( - self.dga.check_acl( - self.session, - project=project, - username="pingou", - refname="refs/heads/c7", - pull_request=None, - repodir=None, - repotype='main', - revfrom=None, - revto=None, - is_internal=False, - ) - ) - - self.expect_info_msg("Ref refs/heads/c7 is blocked") - - def test_protected_rcm(self): - project = self.create_namespaced_project('rpms', 'test') - - self.assertTrue( - self.dga.check_acl( - self.session, - project=project, - username="releng", - refname="refs/heads/f27", - pull_request=None, - repodir=None, - repotype='main', - revfrom=None, - revto=None, - is_internal=False, - ) - ) - - self.expect_info_msg("RCM push") - - @patch_pdc({"rpm/test": {"f26": False, "f27": True}}) - def test_protected_unsupported_branch(self): - project = self.create_namespaced_project('rpms', 'test') - - self.assertFalse( - self.dga.check_acl( - self.session, - project=project, - username="pingou", - refname="refs/heads/f26", - pull_request=None, - repodir=None, - repotype='main', - revfrom=None, - revto=None, - is_internal=False, - ) - ) - - self.expect_info_msg("Branch refs/heads/f26 is unsupported") - - @patch_pdc({"rpm/test": {"f26": False, "f27": True}}) - def test_protected_supported_branch_committer(self): - project = self.create_namespaced_project('rpms', 'test') - - self.assertTrue( - self.dga.check_acl( - self.session, - project=project, - username="pingou", - refname="refs/heads/f27", - pull_request=None, - repodir=None, - repotype='main', - revfrom=None, - revto=None, - is_internal=False, - ) - ) - - self.expect_info_msg("Branch refs/heads/f27 is supported") - - @patch_pdc({"rpm/test": {"f26": False, "f27": True}}) - def test_protected_supported_branch_non_committer(self): - project = self.create_namespaced_project('rpms', 'test') - - self.assertFalse( - self.dga.check_acl( - self.session, - project=project, - username="foo", - refname="refs/heads/f27", - pull_request=None, - repodir=None, - repotype='main', - revfrom=None, - revto=None, - is_internal=False, - ) - ) - - self.expect_info_msg("Branch refs/heads/f27 is supported") - - @patch_pdc({"rpm/test": {"f26": False, "f27": True}}) - def test_protected_unspecified_branch_blacklisted(self): - project = self.create_namespaced_project('rpms', 'test') - - self.assertFalse( - self.dga.check_acl( - self.session, - project=project, - username="pingou", - refname="refs/heads/f28", - pull_request=None, - repodir=None, - repotype='main', - revfrom=None, - revto=None, - is_internal=False, - ) - ) - - self.expect_info_msg("Unspecified ref refs/heads/f28 is blocked") - - @patch_pdc({"rpm/test": {"f26": False, "f27": True}}) - def test_protected_unspecified_branch_normal_committer(self): - project = self.create_namespaced_project('rpms', 'test') - - self.assertTrue( - self.dga.check_acl( - self.session, - project=project, - username="pingou", - refname="refs/heads/mywip", - pull_request=None, - repodir=None, - repotype='main', - revfrom=None, - revto=None, - is_internal=False, - ) - ) - - self.expect_info_msg("Unspecified branch push") - - @patch_pdc({"rpm/test": {"f26": False, "f27": True}}) - def test_protected_unspecified_branch_normal_non_committer(self): - project = self.create_namespaced_project('rpms', 'test') - - self.assertFalse( - self.dga.check_acl( - self.session, - project=project, - username="foo", - refname="refs/heads/mywip", - pull_request=None, - repodir=None, - repotype='main', - revfrom=None, - revto=None, - is_internal=False, - ) - ) - - self.expect_info_msg("Unspecified branch push") - - -class DistGitAuthTestsCentOS(DistGitAuthTests): - dga_config = { - 'PR_ONLY': False, - 'ACL_BLOCK_UNSPECIFIED': True, - 'BLACKLIST_RES': ['refs/heads/f[0-9]+.*'], - 'RCM_GROUP': 'relenggroup', - 'RCM_BRANCHES': ['refs/heads/c[0-9]+'], - 'SUPPORTED_SIGS': ['sig-core'], - 'SIG_PREFIXES': ['refs/heads/c7'], - 'ACL_PROTECTED_NAMESPACES': ['rpms'], - } - - def setUp(self): - super(DistGitAuthTestsCentOS, self).setUp() - - # Create an RCM user/group - arrfab = pagure.lib.model.User( - user='arrfab', - fullname='Fabian Arriton', - token='aaabbbcd', - default_email='arrfab@local.local', - ) - self.session.add(arrfab) - self.session.flush() - sigcore = pagure.lib.model.PagureGroup( - group_name='sig-core', - group_type='user', - display_name='Core SIG group', - user_id=arrfab.id, - ) - self.session.add(sigcore) - self.session.flush() - arrfab.group_objs.append(sigcore) - self.session.commit() - - def test_protected_blacklisted_ref(self): - project = self.create_namespaced_project('rpms', 'test') - - self.assertFalse( - self.dga.check_acl( - self.session, - project=project, - username="pingou", - refname="refs/heads/f27", - pull_request=None, - repodir=None, - repotype='main', - revfrom=None, - revto=None, - is_internal=False, - ) - ) - - self.expect_info_msg("Ref refs/heads/f27 is blocked") - - def test_protected_rcm(self): - project = self.create_namespaced_project('rpms', 'test') - - self.assertTrue( - self.dga.check_acl( - self.session, - project=project, - username="releng", - refname="refs/heads/c7", - pull_request=None, - repodir=None, - repotype='main', - revfrom=None, - revto=None, - is_internal=False, - ) - ) - - self.expect_info_msg("RCM push") - - def test_protected_sig_sig_member(self): - project = self.create_namespaced_project('rpms', 'test') - - self.assertTrue( - self.dga.check_acl( - self.session, - project=project, - username="arrfab", - refname="refs/heads/c7-sig-core-test", - pull_request=None, - repodir=None, - repotype='main', - revfrom=None, - revto=None, - is_internal=False, - ) - ) - - self.expect_info_msg("SIG push") - - def test_protected_sig_sig_member_precise(self): - project = self.create_namespaced_project('rpms', 'test') - - self.assertTrue( - self.dga.check_acl( - self.session, - project=project, - username="arrfab", - refname="refs/heads/c7-sig-core", - pull_request=None, - repodir=None, - repotype='main', - revfrom=None, - revto=None, - is_internal=False, - ) - ) - - self.expect_info_msg("SIG push") - - def test_protected_sig_no_sig_member(self): - project = self.create_namespaced_project('rpms', 'test') - - self.assertFalse( - self.dga.check_acl( - self.session, - project=project, - username="pingou", - refname="refs/heads/c7-sig-core-test", - pull_request=None, - repodir=None, - repotype='main', - revfrom=None, - revto=None, - is_internal=False, - ) - ) - - self.expect_info_msg("Access to namespace rpms is restricted") - - def test_protected_sig_sig_member_no_sig_branch(self): - project = self.create_namespaced_project('rpms', 'test') - - self.assertFalse( - self.dga.check_acl( - self.session, - project=project, - username="arrfab", - refname="refs/heads/c7", - pull_request=None, - repodir=None, - repotype='main', - revfrom=None, - revto=None, - is_internal=False, - ) - ) - - self.expect_info_msg("Access to namespace rpms is restricted") diff --git a/pagure_distgit_tests/__init__.py b/pagure_distgit_tests/__init__.py new file mode 100644 index 0000000..ec609ec --- /dev/null +++ b/pagure_distgit_tests/__init__.py @@ -0,0 +1,11 @@ +from __future__ import print_function + +import tests + + +def setUp(): + tests.setUp() + + +def tearDown(): + tests.tearDown() diff --git a/pagure_distgit_tests/bugzilla_overrides_tests.py b/pagure_distgit_tests/bugzilla_overrides_tests.py new file mode 100644 index 0000000..19b14cb --- /dev/null +++ b/pagure_distgit_tests/bugzilla_overrides_tests.py @@ -0,0 +1,188 @@ +from __future__ import print_function + +import os +import json + +import tests + +from pagure_distgit import plugin + + +class PagureFlaskApiProjectBZOverrideTests(tests.Modeltests): + + def setUp(self): + """ Set up the environnment, ran before every tests. """ + super(PagureFlaskApiProjectBZOverrideTests, self).setUp() + self.session.flush() + tests.create_projects(self.session) + tests.create_projects_git(os.path.join(self.path, "repos"), bare=True) + tests.create_tokens(self.session) + tests.create_tokens_acl(self.session) + self._app.register_blueprint(plugin.DISTGIT_NS) + + def test_override_endpoint(self): + """Test bugzilla overrides. """ + + output = self.app.get("/api/0/somenamespace/test3") + self.assertEqual(output.status_code, 200) + output = self.app.get("/_dg/bzoverrides/somenamespace/test3") + self.assertEqual(output.status_code, 200) + data = json.loads(output.get_data(as_text=True)) + expected_result = {"epel_assignee": "pingou", "fedora_assignee": "pingou"} + data = json.loads(output.get_data(as_text=True)) + self.assertDictEqual(data, expected_result) + + headers = {"Authorization": "token foo_token"} + datainput = { + "epel_assignee": "foo", + "fedora_assignee": "pingou", + } + output = self.app.post( + "/_dg/bzoverrides/somenamespace/test3", data=datainput, headers=headers + ) + # invalid token + self.assertEqual(output.status_code, 401) + + headers = {"Authorization": "token BBBZZZOOO"} + output = self.app.post( + "/_dg/bzoverrides/somenamespace/test3", data=datainput, headers=headers + ) + self.assertEqual(output.status_code, 401) + + output = self.app.get("/_dg/bzoverrides/somenamespace/test3") + self.assertEqual(output.status_code, 200) + data = json.loads(output.get_data(as_text=True)) + self.assertDictEqual(data, expected_result) + + # Change assignees with a valid token + headers = {"Authorization": "token aaabbbcccddd"} + output = self.app.post( + "/_dg/bzoverrides/somenamespace/test3", data=datainput, headers=headers + ) + self.assertEqual(output.status_code, 200) + expected_result = {"epel_assignee": "foo", "fedora_assignee": "pingou"} + output = self.app.get("/_dg/bzoverrides/somenamespace/test3") + data = json.loads(output.get_data(as_text=True)) + self.assertDictEqual(data, expected_result) + + # set both assignees to foo + datainput = {"epel_assignee": "foo", "fedora_assignee": "foo"} + output = self.app.post( + "/_dg/bzoverrides/somenamespace/test3", data=datainput, headers=headers + ) + self.assertEqual(output.status_code, 200) + output = self.app.get("/_dg/bzoverrides/somenamespace/test3") + data = json.loads(output.get_data(as_text=True)) + self.assertDictEqual(data, datainput) + + # Change both assignees back to default + datainput = {"epel_assignee": "", "fedora_assignee": None} + expected_result = {"epel_assignee": "pingou", "fedora_assignee": "pingou"} + output = self.app.post( + "/_dg/bzoverrides/somenamespace/test3", data=datainput, headers=headers + ) + self.assertEqual(output.status_code, 200) + output = self.app.get("/_dg/bzoverrides/somenamespace/test3") + data = json.loads(output.get_data(as_text=True)) + self.assertDictEqual(data, expected_result) + + # Change only one assignee, then change assignee back to default + expected_result = {"epel_assignee": "pingou", "fedora_assignee": "pingou"} + datainput = {"epel_assignee": "foo", "fedora_assignee": "pingou"} + output = self.app.post( + "/_dg/bzoverrides/somenamespace/test3", data=datainput, headers=headers + ) + self.assertEqual(output.status_code, 200) + datainput = {"epel_assignee": "", "fedora_assignee": "pingou"} + output = self.app.post( + "/_dg/bzoverrides/somenamespace/test3", data=datainput, headers=headers + ) + self.assertEqual(output.status_code, 200) + output = self.app.get("/_dg/bzoverrides/somenamespace/test3") + data = json.loads(output.get_data(as_text=True)) + self.assertDictEqual(data, expected_result) + + user = tests.FakeUser(username="pingou") + with tests.user_set(self.app.application, user): + # change one assignee + expected_result = {"epel_assignee": "foo", "fedora_assignee": "pingou"} + datainput = {"epel_assignee": "foo", "fedora_assignee": "pingou"} + output = self.app.post( + "/_dg/bzoverrides/somenamespace/test3", data=datainput + ) + self.assertEqual(output.status_code, 200) + data = json.loads(output.get_data(as_text=True)) + self.assertDictEqual(data, expected_result) + + # change one assignee + expected_result = {"epel_assignee": "foo", "fedora_assignee": "foo"} + datainput = {"epel_assignee": "foo","fedora_assignee": "foo"} + output = self.app.post( + "/_dg/bzoverrides/somenamespace/test3", data=datainput + ) + self.assertEqual(output.status_code, 200) + data = json.loads(output.get_data(as_text=True)) + self.assertDictEqual(data, expected_result) + + # reset assignees + expected_result = {"epel_assignee": "pingou", "fedora_assignee": "pingou"} + datainput = {"epel_assignee": None, "fedora_assignee": ""} + output = self.app.post( + "/_dg/bzoverrides/somenamespace/test3", data=datainput + ) + self.assertEqual(output.status_code, 200) + data = json.loads(output.get_data(as_text=True)) + self.assertDictEqual(data, expected_result) + + # change both assignees to foo, then change only one back + # As the other is considered empty, both should be back to pingou + expected_result = {"epel_assignee": "foo", "fedora_assignee": "foo"} + datainput = {"epel_assignee": "foo","fedora_assignee": "foo"} + output = self.app.post( + "/_dg/bzoverrides/somenamespace/test3", data=datainput + ) + self.assertEqual(output.status_code, 200) + data = json.loads(output.get_data(as_text=True)) + self.assertDictEqual(data, expected_result) + expected_result = {"epel_assignee": "pingou", "fedora_assignee": "pingou"} + datainput = {"epel_assignee": "pingou"} + output = self.app.post( + "/_dg/bzoverrides/somenamespace/test3", data=datainput + ) + self.assertEqual(output.status_code, 200) + data = json.loads(output.get_data(as_text=True)) + self.assertDictEqual(data, expected_result) + + # Invalid datainput resets to default + datainput = {"epel_assignee": "foo","fedora_assignee": "foo"} + output = self.app.post( + "/_dg/bzoverrides/somenamespace/test3", data=datainput + ) + self.assertEqual(output.status_code, 200) + datainput = {} # This is invalid input + output = self.app.post( + "/_dg/bzoverrides/somenamespace/test3", data=datainput + ) + self.assertEqual(output.status_code, 200) + data = json.loads(output.get_data(as_text=True)) + self.assertDictEqual(data, expected_result) + + datainput = {"epel_assignee": "foo","fedora_assignee": "foo"} + output = self.app.post( + "/_dg/bzoverrides/somenamespace/test3", data=datainput + ) + self.assertEqual(output.status_code, 200) + datainput = None # Invalid input + output = self.app.post( + "/_dg/bzoverrides/somenamespace/test3", data=datainput + ) + data = json.loads(output.get_data(as_text=True)) + self.assertEqual(output.status_code, 200) + + user = tests.FakeUser(username="foo") + with tests.user_set(self.app.application, user): + expected_result = {"epel_assignee": "pingou", "fedora_assignee": "pingou"} + output = self.app.post( + "/_dg/bzoverrides/somenamespace/test3", data=expected_result + ) + self.assertEqual(output.status_code, 401) diff --git a/pagure_distgit_tests/dist_git_auth_tests.py b/pagure_distgit_tests/dist_git_auth_tests.py new file mode 100644 index 0000000..36a2ce4 --- /dev/null +++ b/pagure_distgit_tests/dist_git_auth_tests.py @@ -0,0 +1,768 @@ +from __future__ import print_function + +import tempfile +import os + +from mock import Mock, patch + +import pagure.config +import pagure.exceptions +import pagure.lib.model +try: + from pagure.lib import _get_project as get_project +except ImportError: + # From pagure 5.2, code has been moved to pagure.lib.query + from pagure.lib.query import _get_project as get_project + +import tests + +import dist_git_auth + + + +def patch_pdc(values): + """ Decorator to patch the PDC calls to return values for this test. + + Args: + values (dict): A dictionary where the keys are project fullnames + (i.e. namespace/name), and the values are dicts with key branch + names, and values their supported status. + Note that as namespace, the PDC "type" is used, which lacks the "s" + for rpms and modules. + e.g.: {'rpm/test': {'f28': True, 'f27': False}} + """ + def pdc_get_paged(_, global_component, type, name, fields): + """ Function that emulates the pdc.get_paged call + + Args as provided by dist_git_auth's calls to it. + + Args: + _ (anything): PDCClient internal + global_component (string): package name + type (string): The PDC "type": "rpm", "module", ... + name (string): The branch name + fields (list): Always ["active"] + """ + fullname = '%s/%s' % (type, global_component) + if fullname not in values: + return [] + val = values[fullname] + if name not in val: + return [] + val = val[name] + if val in (True, False): + return [{"active": val}] + # This case is used to emulate "weird" results + return val + + def decorator(func): + def test_wrapper(*args, **kwargs): + with patch.object(dist_git_auth.PDCClient, '__getitem__'): + with patch.object(dist_git_auth.PDCClient, 'get_paged', + side_effect=pdc_get_paged): + return func(*args, **kwargs) + return test_wrapper + return decorator + + +class DistGitAuthTests(tests.Modeltests): + """ Test DistGitAuth ACLs with Fedora config. """ + + maxDiff = None + + def setUp(self): + """ Set up the environment in which to run the tests. """ + super(DistGitAuthTests, self).setUp() + + pagure.config.config['ACL_DEBUG'] = True + pagure.config.config['EXTERNAL_COMMITTER'] = {"relenggroup": {}} + pagure.config.config.update(self.dga_config) + + self.dga = dist_git_auth.DistGitAuth() + self.dga.info = Mock(side_effect=print) + # We default to saying it's not a forced push + dist_git_auth.is_forced_push = Mock(return_value=False) + + # Create an RCM user/group + rcmuser = pagure.lib.model.User( + user='releng', + fullname='Release Engineering', + token='aaabbbcd', + default_email='rcm@local.local', + ) + self.session.add(rcmuser) + self.session.flush() + rcmgroup = pagure.lib.model.PagureGroup( + group_name='relenggroup', + group_type='user', + display_name='Releng group', + user_id=rcmuser.id, + ) + self.session.add(rcmgroup) + self.session.flush() + rcmuser.group_objs.append(rcmgroup) + self.session.commit() + + def tearDown(self): + """ Tear down the environment in which the tests ran. """ + self.dga = None + super(DistGitAuthTests, self).tearDown() + + def create_namespaced_project(self, namespace, name, is_fork=False, settings={}): + item = pagure.lib.model.Project( + user_id=1, # pingou + name=name, + is_fork=is_fork, + parent_id=3 if is_fork else None, + description='namespaced test project', + hook_token='aaabbbeee', + namespace=namespace, + settings=settings + ) + item.close_status = [ + 'Invalid', 'Insufficient data', 'Fixed', 'Duplicate'] + self.session.add(item) + self.session.commit() + return get_project( + self.session, name=name, namespace=namespace) + + def expect_info_msg(self, expect_msg): + found = False + for call in self.dga.info.call_args_list: + args = call[0] + msg = args[0] + if msg == expect_msg: + found = True + if not found: + raise AssertionError( + "Info message '%s' expected but not found" % expect_msg) + + +class DistGitAuthTestsGeneric(DistGitAuthTests): + dga_config = {} + + def test_unused_repotype(self): + self.assertFalse( + self.dga.check_acl( + self.session, + project=None, + username=None, + refname=None, + pull_request=None, + repodir=None, + repotype='tickets', + revfrom=None, + revto=None, + is_internal=False, + ) + ) + + self.expect_info_msg("Repotype tickets not in use") + + def test_branch_deletion(self): + dist_git_auth.is_forced_push.return_value = True + + project = self.create_namespaced_project('rpms', 'test') + + self.assertFalse( + self.dga.check_acl( + self.session, + project=project, + username=None, + refname=None, + pull_request=None, + repodir=None, + repotype='main', + revfrom=None, + revto='0000000000000000000000000000000000000000', + is_internal=False, + ) + ) + + self.expect_info_msg("Branch deletion is not allowed") + + def test_forced_push(self): + dist_git_auth.is_forced_push.return_value = True + + project = self.create_namespaced_project('rpms', 'test') + + self.assertFalse( + self.dga.check_acl( + self.session, + project=project, + username=None, + refname=None, + pull_request=None, + repodir=None, + repotype='main', + revfrom=None, + revto=None, + is_internal=False, + ) + ) + + self.expect_info_msg("Forced pushes are not allowed") + + def test_internal(self): + project = self.create_namespaced_project('rpms', 'test') + + self.assertTrue( + self.dga.check_acl( + self.session, + project=project, + username=None, + refname=None, + pull_request=None, + repodir=None, + repotype='main', + revfrom=None, + revto=None, + is_internal=True, + ) + ) + + self.expect_info_msg("Internal push allowed") + + def test_deploykey(self): + project = self.create_namespaced_project('rpms', 'test') + + self.assertFalse( + self.dga.check_acl( + self.session, + project=project, + username="deploykey_foobar", + refname=None, + pull_request=None, + repodir=None, + repotype='main', + revfrom=None, + revto=None, + is_internal=False, + ) + ) + + self.expect_info_msg("Deploy keys are disabled") + + def test_invalid_user(self): + project = self.create_namespaced_project('rpms', 'test') + + self.assertRaises( + pagure.exceptions.PagureException, + self.dga.check_acl, + self.session, + project=project, + username="nosuchuser", + refname=None, + pull_request=None, + repodir=None, + repotype='main', + revfrom=None, + revto=None, + is_internal=False, + ) + + def test_unprotected_committer(self): + project = self.create_namespaced_project('unprotected', 'test') + + self.assertTrue( + self.dga.check_acl( + self.session, + project=project, + username="pingou", + refname="refs/heads/mywip", + pull_request=None, + repodir=None, + repotype='main', + revfrom=None, + revto=None, + is_internal=False, + ) + ) + + self.expect_info_msg("Committer push") + + def test_unprotected_non_committer(self): + project = self.create_namespaced_project('unprotected', 'test') + + self.assertFalse( + self.dga.check_acl( + self.session, + project=project, + username="foo", + refname="refs/heads/mywip", + pull_request=None, + repodir=None, + repotype='main', + revfrom=None, + revto=None, + is_internal=False, + ) + ) + + self.expect_info_msg("Fall-through deny") + + def test_unprotected_pr_required_pr(self): + project = self.create_namespaced_project('unprotected', 'test') + self.dga.global_pr_only = True + + self.assertTrue( + self.dga.check_acl( + self.session, + project=project, + username="pingou", + refname="refs/heads/mywip", + pull_request=True, + repodir=None, + repotype='main', + revfrom=None, + revto=None, + is_internal=False, + ) + ) + + self.expect_info_msg("Committer push") + + def test_unprotected_pr_required_no_pr(self): + project = self.create_namespaced_project('unprotected', 'test') + self.dga.global_pr_only = True + + self.assertFalse( + self.dga.check_acl( + self.session, + project=project, + username="pingou", + refname="refs/heads/mywip", + pull_request=None, + repodir=None, + repotype='main', + revfrom=None, + revto=None, + is_internal=False, + ) + ) + + self.expect_info_msg("A pull request is required for this branch") + + def test_unprotected_pr_required_no_pr_cvsadmin(self): + project = self.create_namespaced_project('unprotected', 'test') + self.dga.global_pr_only = True + + self.assertTrue( + self.dga.check_acl( + self.session, + project=project, + username="releng", + refname="refs/heads/mywip", + pull_request=None, + repodir=None, + repotype='main', + revfrom=None, + revto=None, + is_internal=False, + ) + ) + + self.expect_info_msg("Committer push") + + def test_unprotected_pr_required_requests(self): + project = self.create_namespaced_project('unprotected', 'test') + self.dga.global_pr_only = True + + self.assertTrue( + self.dga.check_acl( + self.session, + project=project, + username="pingou", + refname="refs/heads/mywip", + pull_request=None, + repodir=None, + repotype='requests', + revfrom=None, + revto=None, + is_internal=False, + ) + ) + + self.expect_info_msg("Committer push") + + def test_unprotected_pr_required_repo_pr_only_no_pr(self): + settings = {"pull_request_access_only": True} + project = self.create_namespaced_project('unprotected', 'test', settings=settings) + + self.assertFalse( + self.dga.check_acl( + self.session, + project=project, + username="pingou", + refname="refs/heads/mywip", + pull_request=None, + repodir=None, + repotype='main', + revfrom=None, + revto=None, + is_internal=False, + ) + ) + + self.expect_info_msg("A pull request is required for this branch") + + def test_unprotected_pr_required_repo_pr_only(self): + settings = {"pull_request_access_only": True} + project = self.create_namespaced_project( + 'unprotected', + 'test', + settings=settings + ) + self.assertTrue( + self.dga.check_acl( + self.session, + project=project, + username="pingou", + refname="refs/heads/mywip", + pull_request="pull_request", + repodir=None, + repotype='main', + revfrom=None, + revto=None, + is_internal=False, + ) + ) + + self.expect_info_msg("Committer push") + + +class DistGitAuthTestsFedora(DistGitAuthTests): + dga_config = { + 'PR_ONLY': False, + 'ACL_BLOCK_UNSPECIFIED': False, + 'BLACKLIST_RES': ['refs/heads/c[0-9]+.*'], + 'UNSPECIFIED_BLACKLIST_RES': [ + 'refs/heads/f[0-9]+', + ], + 'RCM_GROUP': 'relenggroup', + 'RCM_BRANCHES': ['refs/heads/f[0-9]+'], + 'ACL_PROTECTED_NAMESPACES': ['rpms', 'modules', 'container'], + 'PDC_URL': 'invalid://', + } + + def test_protected_blacklisted_ref(self): + project = self.create_namespaced_project('rpms', 'test') + + self.assertFalse( + self.dga.check_acl( + self.session, + project=project, + username="pingou", + refname="refs/heads/c7", + pull_request=None, + repodir=None, + repotype='main', + revfrom=None, + revto=None, + is_internal=False, + ) + ) + + self.expect_info_msg("Ref refs/heads/c7 is blocked") + + def test_protected_rcm(self): + project = self.create_namespaced_project('rpms', 'test') + + self.assertTrue( + self.dga.check_acl( + self.session, + project=project, + username="releng", + refname="refs/heads/f27", + pull_request=None, + repodir=None, + repotype='main', + revfrom=None, + revto=None, + is_internal=False, + ) + ) + + self.expect_info_msg("RCM push") + + @patch_pdc({"rpm/test": {"f26": False, "f27": True}}) + def test_protected_unsupported_branch(self): + project = self.create_namespaced_project('rpms', 'test') + + self.assertFalse( + self.dga.check_acl( + self.session, + project=project, + username="pingou", + refname="refs/heads/f26", + pull_request=None, + repodir=None, + repotype='main', + revfrom=None, + revto=None, + is_internal=False, + ) + ) + + self.expect_info_msg("Branch refs/heads/f26 is unsupported") + + @patch_pdc({"rpm/test": {"f26": False, "f27": True}}) + def test_protected_supported_branch_committer(self): + project = self.create_namespaced_project('rpms', 'test') + + self.assertTrue( + self.dga.check_acl( + self.session, + project=project, + username="pingou", + refname="refs/heads/f27", + pull_request=None, + repodir=None, + repotype='main', + revfrom=None, + revto=None, + is_internal=False, + ) + ) + + self.expect_info_msg("Branch refs/heads/f27 is supported") + + @patch_pdc({"rpm/test": {"f26": False, "f27": True}}) + def test_protected_supported_branch_non_committer(self): + project = self.create_namespaced_project('rpms', 'test') + + self.assertFalse( + self.dga.check_acl( + self.session, + project=project, + username="foo", + refname="refs/heads/f27", + pull_request=None, + repodir=None, + repotype='main', + revfrom=None, + revto=None, + is_internal=False, + ) + ) + + self.expect_info_msg("Branch refs/heads/f27 is supported") + + @patch_pdc({"rpm/test": {"f26": False, "f27": True}}) + def test_protected_unspecified_branch_blacklisted(self): + project = self.create_namespaced_project('rpms', 'test') + + self.assertFalse( + self.dga.check_acl( + self.session, + project=project, + username="pingou", + refname="refs/heads/f28", + pull_request=None, + repodir=None, + repotype='main', + revfrom=None, + revto=None, + is_internal=False, + ) + ) + + self.expect_info_msg("Unspecified ref refs/heads/f28 is blocked") + + @patch_pdc({"rpm/test": {"f26": False, "f27": True}}) + def test_protected_unspecified_branch_normal_committer(self): + project = self.create_namespaced_project('rpms', 'test') + + self.assertTrue( + self.dga.check_acl( + self.session, + project=project, + username="pingou", + refname="refs/heads/mywip", + pull_request=None, + repodir=None, + repotype='main', + revfrom=None, + revto=None, + is_internal=False, + ) + ) + + self.expect_info_msg("Unspecified branch push") + + @patch_pdc({"rpm/test": {"f26": False, "f27": True}}) + def test_protected_unspecified_branch_normal_non_committer(self): + project = self.create_namespaced_project('rpms', 'test') + + self.assertFalse( + self.dga.check_acl( + self.session, + project=project, + username="foo", + refname="refs/heads/mywip", + pull_request=None, + repodir=None, + repotype='main', + revfrom=None, + revto=None, + is_internal=False, + ) + ) + + self.expect_info_msg("Unspecified branch push") + + +class DistGitAuthTestsCentOS(DistGitAuthTests): + dga_config = { + 'PR_ONLY': False, + 'ACL_BLOCK_UNSPECIFIED': True, + 'BLACKLIST_RES': ['refs/heads/f[0-9]+.*'], + 'RCM_GROUP': 'relenggroup', + 'RCM_BRANCHES': ['refs/heads/c[0-9]+'], + 'SUPPORTED_SIGS': ['sig-core'], + 'SIG_PREFIXES': ['refs/heads/c7'], + 'ACL_PROTECTED_NAMESPACES': ['rpms'], + } + + def setUp(self): + super(DistGitAuthTestsCentOS, self).setUp() + + # Create an RCM user/group + arrfab = pagure.lib.model.User( + user='arrfab', + fullname='Fabian Arriton', + token='aaabbbcd', + default_email='arrfab@local.local', + ) + self.session.add(arrfab) + self.session.flush() + sigcore = pagure.lib.model.PagureGroup( + group_name='sig-core', + group_type='user', + display_name='Core SIG group', + user_id=arrfab.id, + ) + self.session.add(sigcore) + self.session.flush() + arrfab.group_objs.append(sigcore) + self.session.commit() + + def test_protected_blacklisted_ref(self): + project = self.create_namespaced_project('rpms', 'test') + + self.assertFalse( + self.dga.check_acl( + self.session, + project=project, + username="pingou", + refname="refs/heads/f27", + pull_request=None, + repodir=None, + repotype='main', + revfrom=None, + revto=None, + is_internal=False, + ) + ) + + self.expect_info_msg("Ref refs/heads/f27 is blocked") + + def test_protected_rcm(self): + project = self.create_namespaced_project('rpms', 'test') + + self.assertTrue( + self.dga.check_acl( + self.session, + project=project, + username="releng", + refname="refs/heads/c7", + pull_request=None, + repodir=None, + repotype='main', + revfrom=None, + revto=None, + is_internal=False, + ) + ) + + self.expect_info_msg("RCM push") + + def test_protected_sig_sig_member(self): + project = self.create_namespaced_project('rpms', 'test') + + self.assertTrue( + self.dga.check_acl( + self.session, + project=project, + username="arrfab", + refname="refs/heads/c7-sig-core-test", + pull_request=None, + repodir=None, + repotype='main', + revfrom=None, + revto=None, + is_internal=False, + ) + ) + + self.expect_info_msg("SIG push") + + def test_protected_sig_sig_member_precise(self): + project = self.create_namespaced_project('rpms', 'test') + + self.assertTrue( + self.dga.check_acl( + self.session, + project=project, + username="arrfab", + refname="refs/heads/c7-sig-core", + pull_request=None, + repodir=None, + repotype='main', + revfrom=None, + revto=None, + is_internal=False, + ) + ) + + self.expect_info_msg("SIG push") + + def test_protected_sig_no_sig_member(self): + project = self.create_namespaced_project('rpms', 'test') + + self.assertFalse( + self.dga.check_acl( + self.session, + project=project, + username="pingou", + refname="refs/heads/c7-sig-core-test", + pull_request=None, + repodir=None, + repotype='main', + revfrom=None, + revto=None, + is_internal=False, + ) + ) + + self.expect_info_msg("Access to namespace rpms is restricted") + + def test_protected_sig_sig_member_no_sig_branch(self): + project = self.create_namespaced_project('rpms', 'test') + + self.assertFalse( + self.dga.check_acl( + self.session, + project=project, + username="arrfab", + refname="refs/heads/c7", + pull_request=None, + repodir=None, + repotype='main', + revfrom=None, + revto=None, + is_internal=False, + ) + ) + + self.expect_info_msg("Access to namespace rpms is restricted") From 51d0155d24e75dd623733af2a78a469fd9e6c61b Mon Sep 17 00:00:00 2001 From: Pierre-Yves Chibon Date: Dec 19 2019 11:59:51 +0000 Subject: [PATCH 2/3] Rework the structure of the tests for the bugzilla override endpoints Signed-off-by: Pierre-Yves Chibon --- diff --git a/pagure_distgit_tests/bugzilla_overrides_tests.py b/pagure_distgit_tests/bugzilla_overrides_tests.py index 19b14cb..6f27149 100644 --- a/pagure_distgit_tests/bugzilla_overrides_tests.py +++ b/pagure_distgit_tests/bugzilla_overrides_tests.py @@ -9,6 +9,7 @@ from pagure_distgit import plugin class PagureFlaskApiProjectBZOverrideTests(tests.Modeltests): + """ Tests the bugzilla override endpoints added in pagure-dist-git. """ def setUp(self): """ Set up the environnment, ran before every tests. """ @@ -20,169 +21,240 @@ class PagureFlaskApiProjectBZOverrideTests(tests.Modeltests): tests.create_tokens_acl(self.session) self._app.register_blueprint(plugin.DISTGIT_NS) - def test_override_endpoint(self): - """Test bugzilla overrides. """ + def test_default_values(self): + """Test the default values returned by the bz overrides endpoint. """ - output = self.app.get("/api/0/somenamespace/test3") - self.assertEqual(output.status_code, 200) + expected_result = { + "epel_assignee": "pingou", + "fedora_assignee": "pingou", + } output = self.app.get("/_dg/bzoverrides/somenamespace/test3") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) - expected_result = {"epel_assignee": "pingou", "fedora_assignee": "pingou"} - data = json.loads(output.get_data(as_text=True)) self.assertDictEqual(data, expected_result) + def test_token_missing_ACL(self): + """Test the bz endpoint with an API token missing the `modify_project` ACL. + """ headers = {"Authorization": "token foo_token"} - datainput = { - "epel_assignee": "foo", - "fedora_assignee": "pingou", - } output = self.app.post( - "/_dg/bzoverrides/somenamespace/test3", data=datainput, headers=headers + "/_dg/bzoverrides/somenamespace/test3", headers=headers ) # invalid token self.assertEqual(output.status_code, 401) + def test_invalid_token(self): + """Test the bz endpoint with an invalid API token. """ headers = {"Authorization": "token BBBZZZOOO"} + datainput = {} output = self.app.post( - "/_dg/bzoverrides/somenamespace/test3", data=datainput, headers=headers + "/_dg/bzoverrides/somenamespace/test3", + data=datainput, + headers=headers, ) self.assertEqual(output.status_code, 401) - output = self.app.get("/_dg/bzoverrides/somenamespace/test3") + def test_change_both_assignee(self): + """Test the bz endpoint when changing both assignee at once. """ + headers = {"Authorization": "token aaabbbcccddd"} + datainput = {"epel_assignee": "foo", "fedora_assignee": "foo"} + output = self.app.post( + "/_dg/bzoverrides/somenamespace/test3", + data=datainput, + headers=headers, + ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) - self.assertDictEqual(data, expected_result) + self.assertDictEqual(data, datainput) - # Change assignees with a valid token + def test_change_fedora_assignee(self): + """Test the bz endpoint when changing the Fedora assignee while keeping + the EPEL one. + """ headers = {"Authorization": "token aaabbbcccddd"} + datainput = {"epel_assignee": "foo", "fedora_assignee": "pingou"} output = self.app.post( - "/_dg/bzoverrides/somenamespace/test3", data=datainput, headers=headers + "/_dg/bzoverrides/somenamespace/test3", + data=datainput, + headers=headers, + ) + self.assertEqual(output.status_code, 200) + data = json.loads(output.get_data(as_text=True)) + self.assertDictEqual(data, datainput) + + def test_change_epel_assignee(self): + """Test the bz endpoint when changing the EPEL assignee while keeping + the Fedora one. + """ + headers = {"Authorization": "token aaabbbcccddd"} + datainput = {"epel_assignee": "", "fedora_assignee": None} + expected_result = { + "epel_assignee": "pingou", + "fedora_assignee": "pingou", + } + output = self.app.post( + "/_dg/bzoverrides/somenamespace/test3", + data=datainput, + headers=headers, ) self.assertEqual(output.status_code, 200) - expected_result = {"epel_assignee": "foo", "fedora_assignee": "pingou"} - output = self.app.get("/_dg/bzoverrides/somenamespace/test3") data = json.loads(output.get_data(as_text=True)) self.assertDictEqual(data, expected_result) - # set both assignees to foo - datainput = {"epel_assignee": "foo", "fedora_assignee": "foo"} + def test_reset_fedora_assignees(self): + """Test the bz endpoint when resetting the Fedora assignee. + """ + headers = {"Authorization": "token aaabbbcccddd"} + datainput = {"fedora_assignee": "foo"} output = self.app.post( - "/_dg/bzoverrides/somenamespace/test3", data=datainput, headers=headers + "/_dg/bzoverrides/somenamespace/test3", + data=datainput, + headers=headers, ) self.assertEqual(output.status_code, 200) - output = self.app.get("/_dg/bzoverrides/somenamespace/test3") data = json.loads(output.get_data(as_text=True)) - self.assertDictEqual(data, datainput) + self.assertDictEqual( + data, {"epel_assignee": "pingou", "fedora_assignee": "foo"} + ) - # Change both assignees back to default - datainput = {"epel_assignee": "", "fedora_assignee": None} - expected_result = {"epel_assignee": "pingou", "fedora_assignee": "pingou"} + datainput = {"fedora_assignee": ""} output = self.app.post( - "/_dg/bzoverrides/somenamespace/test3", data=datainput, headers=headers + "/_dg/bzoverrides/somenamespace/test3", + data=datainput, + headers=headers, ) self.assertEqual(output.status_code, 200) - output = self.app.get("/_dg/bzoverrides/somenamespace/test3") data = json.loads(output.get_data(as_text=True)) - self.assertDictEqual(data, expected_result) + self.assertDictEqual( + data, {"epel_assignee": "pingou", "fedora_assignee": "pingou"} + ) - # Change only one assignee, then change assignee back to default - expected_result = {"epel_assignee": "pingou", "fedora_assignee": "pingou"} + def test_reset_epel_assignees(self): + """Test the bz endpoint when resetting the EPEL assignee. + """ + headers = {"Authorization": "token aaabbbcccddd"} datainput = {"epel_assignee": "foo", "fedora_assignee": "pingou"} output = self.app.post( - "/_dg/bzoverrides/somenamespace/test3", data=datainput, headers=headers + "/_dg/bzoverrides/somenamespace/test3", + data=datainput, + headers=headers, ) self.assertEqual(output.status_code, 200) + data = json.loads(output.get_data(as_text=True)) + self.assertDictEqual(data, datainput) + + datainput = {"fedora_assignee": "pingou"} + output = self.app.post( + "/_dg/bzoverrides/somenamespace/test3", + data=datainput, + headers=headers, + ) + self.assertEqual(output.status_code, 200) + data = json.loads(output.get_data(as_text=True)) + self.assertDictEqual( + data, {"epel_assignee": "pingou", "fedora_assignee": "pingou"} + ) + + def test_reset_both_assignees(self): + """Test the bz endpoint when resetting the both assignees. + """ + headers = {"Authorization": "token aaabbbcccddd"} datainput = {"epel_assignee": "", "fedora_assignee": "pingou"} output = self.app.post( - "/_dg/bzoverrides/somenamespace/test3", data=datainput, headers=headers + "/_dg/bzoverrides/somenamespace/test3", + data=datainput, + headers=headers, ) self.assertEqual(output.status_code, 200) - output = self.app.get("/_dg/bzoverrides/somenamespace/test3") data = json.loads(output.get_data(as_text=True)) - self.assertDictEqual(data, expected_result) + self.assertDictEqual( + data, {"epel_assignee": "pingou", "fedora_assignee": "pingou"} + ) - user = tests.FakeUser(username="pingou") + def test_changing_assignees_logged_in_invalid_user(self): + """Test the bz endpoint when changing the both assignees when logged in + the UI but with an user that is not allowed. + """ + user = tests.FakeUser(username="foo") with tests.user_set(self.app.application, user): - # change one assignee - expected_result = {"epel_assignee": "foo", "fedora_assignee": "pingou"} - datainput = {"epel_assignee": "foo", "fedora_assignee": "pingou"} + expected_result = { + "epel_assignee": "pingou", + "fedora_assignee": "pingou", + } output = self.app.post( - "/_dg/bzoverrides/somenamespace/test3", data=datainput + "/_dg/bzoverrides/somenamespace/test3", data=expected_result ) - self.assertEqual(output.status_code, 200) - data = json.loads(output.get_data(as_text=True)) - self.assertDictEqual(data, expected_result) + self.assertEqual(output.status_code, 401) + def test_change_epel_assignee_logged_in(self): + """Test the bz endpoint when changing the EPEL assignee when logged in + the UI. + """ + user = tests.FakeUser(username="pingou") + with tests.user_set(self.app.application, user): # change one assignee - expected_result = {"epel_assignee": "foo", "fedora_assignee": "foo"} - datainput = {"epel_assignee": "foo","fedora_assignee": "foo"} + expected_result = {"epel_assignee": "foo"} + datainput = {"epel_assignee": "foo"} output = self.app.post( "/_dg/bzoverrides/somenamespace/test3", data=datainput ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) - self.assertDictEqual(data, expected_result) + self.assertDictEqual( + data, {"epel_assignee": "foo", "fedora_assignee": "pingou"} + ) - # reset assignees - expected_result = {"epel_assignee": "pingou", "fedora_assignee": "pingou"} - datainput = {"epel_assignee": None, "fedora_assignee": ""} + def test_change_fedora_assignee_logged_in(self): + """Test the bz endpoint when changing the Fedora assignee when logged in + the UI. + """ + user = tests.FakeUser(username="pingou") + with tests.user_set(self.app.application, user): + # change one assignee + datainput = {"fedora_assignee": "foo"} output = self.app.post( "/_dg/bzoverrides/somenamespace/test3", data=datainput ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) - self.assertDictEqual(data, expected_result) + self.assertDictEqual( + data, {"epel_assignee": "pingou", "fedora_assignee": "foo"} + ) - # change both assignees to foo, then change only one back - # As the other is considered empty, both should be back to pingou - expected_result = {"epel_assignee": "foo", "fedora_assignee": "foo"} - datainput = {"epel_assignee": "foo","fedora_assignee": "foo"} + def test_change_both_assignees_logged_in(self): + """Test the bz endpoint when changing the both assignees when logged in + the UI. + """ + user = tests.FakeUser(username="pingou") + with tests.user_set(self.app.application, user): + datainput = {"epel_assignee": "foo", "fedora_assignee": "foo"} output = self.app.post( "/_dg/bzoverrides/somenamespace/test3", data=datainput ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) - self.assertDictEqual(data, expected_result) - expected_result = {"epel_assignee": "pingou", "fedora_assignee": "pingou"} - datainput = {"epel_assignee": "pingou"} - output = self.app.post( - "/_dg/bzoverrides/somenamespace/test3", data=datainput + self.assertDictEqual( + data, {"epel_assignee": "foo", "fedora_assignee": "foo"} ) - self.assertEqual(output.status_code, 200) - data = json.loads(output.get_data(as_text=True)) - self.assertDictEqual(data, expected_result) - # Invalid datainput resets to default - datainput = {"epel_assignee": "foo","fedora_assignee": "foo"} - output = self.app.post( - "/_dg/bzoverrides/somenamespace/test3", data=datainput - ) - self.assertEqual(output.status_code, 200) - datainput = {} # This is invalid input + def test_resetting_assignees_logged_in(self): + """Test the bz endpoint when resetting assignees when logged in the UI. + """ + user = tests.FakeUser(username="pingou") + with tests.user_set(self.app.application, user): + datainput = {"epel_assignee": "foo"} output = self.app.post( "/_dg/bzoverrides/somenamespace/test3", data=datainput ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) - self.assertDictEqual(data, expected_result) - - datainput = {"epel_assignee": "foo","fedora_assignee": "foo"} - output = self.app.post( - "/_dg/bzoverrides/somenamespace/test3", data=datainput + self.assertDictEqual( + data, {"epel_assignee": "foo", "fedora_assignee": "pingou"} ) + + output = self.app.post("/_dg/bzoverrides/somenamespace/test3") self.assertEqual(output.status_code, 200) - datainput = None # Invalid input - output = self.app.post( - "/_dg/bzoverrides/somenamespace/test3", data=datainput - ) data = json.loads(output.get_data(as_text=True)) - self.assertEqual(output.status_code, 200) - - user = tests.FakeUser(username="foo") - with tests.user_set(self.app.application, user): - expected_result = {"epel_assignee": "pingou", "fedora_assignee": "pingou"} - output = self.app.post( - "/_dg/bzoverrides/somenamespace/test3", data=expected_result + self.assertDictEqual( + data, {"epel_assignee": "pingou", "fedora_assignee": "pingou"} ) - self.assertEqual(output.status_code, 401) From 15aa95b40ff70840da8a6f2f9e1e251e0067a3b8 Mon Sep 17 00:00:00 2001 From: Pierre-Yves Chibon Date: Dec 19 2019 12:23:04 +0000 Subject: [PATCH 3/3] Adjust the instruction on how to run tests in the README Signed-off-by: Pierre-Yves Chibon --- diff --git a/README.rst b/README.rst index bd31ee3..4c65ede 100644 --- a/README.rst +++ b/README.rst @@ -75,9 +75,7 @@ Tests The tests here require the *test suite* of pagure itself to work. You have to modify your PYTHONPATH to find them. Run with:: - $ PYTHONPATH=.:/path/to/pagure/checkout nosetests dist_git_auth_tests.py -or - $ PYTHONPATH=.:/path/to/pagure/checkout nosetests bugzilla-override-tests.py + $ PYTHONPATH=.:/path/to/pagure/checkout nosetests pagure_distgit_tests/ You can use our requirements-testing.txt to install testing dependencies with pip: $ pip install -r /path/to/pagure/checkout/requirements.txt