""" Unit tests for `toddlers.plugins.scm_request_processor` """ import json import logging import re from unittest.mock import call, MagicMock, Mock, patch import xmlrpc.client import arrow from pagure_messages.issue_schema import IssueCommentAddedV1, IssueNewV1 import pytest from toddlers.exceptions import ValidationError import toddlers.plugins.scm_request_processor as scm_request_processor class TestAcceptsTopic: """ Test class for `toddlers.plugins.scm_request_processor.SCMRequestProcessor.accepts_topic` method. """ toddler_cls = scm_request_processor.SCMRequestProcessor def test_accetps_topic_invalid(self, toddler): """ Assert that invalid topic is not accepted. """ assert toddler.accepts_topic("foo.bar") is False @pytest.mark.parametrize( "topic", [ "io.pagure.*.pagure.issue.new", "io.pagure.*.pagure.issue.edit", "io.pagure.*.pagure.issue.comment.added", "io.pagure.stg.pagure.issue.new", "io.pagure.stg.pagure.issue.edit", "io.pagure.stg.pagure.issue.comment.added", "io.pagure.prod.pagure.issue.new", "io.pagure.prod.pagure.issue.edit", "io.pagure.prod.pagure.issue.comment.added", ], ) def test_accetps_topic_valid(self, topic, toddler): """ Assert that valid topics are accepted. """ assert toddler.accepts_topic(topic) class TestProcess: """ Test class for `toddlers.plugins.scm_request_processor.SCMRequestProcessor.process` method. """ toddler_cls = scm_request_processor.SCMRequestProcessor def test_process_invalid_project(self, caplog, toddler): """ Assert that messages from other projects than fedora_scm_requests will be skipped. """ caplog.set_level(logging.INFO) msg = IssueNewV1() msg.body = {"project": {"fullname": "foo/bar"}} with patch( "toddlers.plugins.scm_request_processor.SCMRequestProcessor.process_ticket" ) as mock_process_ticket: toddler.process({}, msg) mock_process_ticket.assert_not_called() assert ( caplog.records[-1].message == "The message doesn't belong to project releng/fedora-scm-requests. Skipping message." ) def test_process_issue_not_open(self, caplog, toddler): """ Assert that messages with closed issues will be skipped. """ caplog.set_level(logging.INFO) msg = IssueNewV1() msg.body = { "project": {"fullname": scm_request_processor.PROJECT_NAMESPACE}, "issue": {"id": 100, "status": "Closed"}, } with patch( "toddlers.plugins.scm_request_processor.SCMRequestProcessor.process_ticket" ) as mock_process_ticket: toddler.process({}, msg) mock_process_ticket.assert_not_called() assert ( caplog.records[-1].message == "The issue 100 is not open. Skipping message." ) def test_process_change_done_by_toddler(self, caplog, toddler): """ Assert that toddler will ignore messages that were emitted by it. """ caplog.set_level(logging.INFO) pagure_user = "pagure_user" config = {"pagure_user": pagure_user} msg = IssueCommentAddedV1() msg.body = { "project": {"fullname": scm_request_processor.PROJECT_NAMESPACE}, "issue": {"status": "Open"}, "agent": pagure_user, } with patch( "toddlers.plugins.scm_request_processor.SCMRequestProcessor.process_ticket" ) as mock_process_ticket: toddler.process(config, msg) mock_process_ticket.assert_not_called() assert caplog.records[ -1 ].message == "Last change on the ticket was done by {0}. Ignoring the message.".format( pagure_user ) @patch("toddlers.utils.pdc.set_pdc") @patch("toddlers.utils.pagure.set_pagure") @patch("toddlers.utils.fedora_account.set_fasjson") @patch("toddlers.utils.bugzilla_system.set_bz") def test_process_exception( self, mock_bugzilla, mock_fasjson, mock_pagure, mock_pdc, toddler ): """ Assert that message toddler will be initialized correctly, if message passes initial processing. """ msg = IssueNewV1() issue = {"id": 100, "status": "Open"} msg.body = { "project": {"fullname": scm_request_processor.PROJECT_NAMESPACE}, "issue": issue, "agent": "agent", } config = { "branch_slas": {}, "monitoring_choices": [], "pagure_namespace_to_component": {}, "pagure_namespace_to_product": {}, "temp_dir": "", "dist_git_url": "https://src.fedoraproject.org", "dist_git_token": "Private API Key", } mock_pagure_io = Mock() mock_pagure.return_value = mock_pagure_io with patch( "toddlers.plugins.scm_request_processor.SCMRequestProcessor.process_ticket" ) as mock_process_ticket: mock_process_ticket.side_effect = Exception("Exception") toddler.process(config, msg) mock_process_ticket.assert_called_with(issue) mock_pdc.assert_called_with(config) mock_pagure.assert_has_calls( [ call(config), call( { "pagure_url": "https://src.fedoraproject.org", "pagure_api_key": "Private API Key", } ), ] ) mock_fasjson.assert_called_with(config) mock_bugzilla.assert_called_with(config) mock_pagure_io.add_comment_to_issue.assert_called_once() @patch("toddlers.utils.pdc.set_pdc") @patch("toddlers.utils.pagure.set_pagure") @patch("toddlers.utils.fedora_account.set_fasjson") @patch("toddlers.utils.bugzilla_system.set_bz") def test_process(self, mock_bugzilla, mock_fasjson, mock_pagure, mock_pdc, toddler): """ Assert that message toddler will be initialized correctly, if message passes initial processing. """ msg = IssueNewV1() issue = {"id": 100, "status": "Open"} msg.body = { "project": {"fullname": scm_request_processor.PROJECT_NAMESPACE}, "issue": issue, "agent": "agent", } config = { "branch_slas": {}, "monitoring_choices": [], "pagure_namespace_to_component": {}, "pagure_namespace_to_product": {}, "temp_dir": "", "dist_git_url": "https://src.fedoraproject.org", "dist_git_token": "Private API Key", } with patch( "toddlers.plugins.scm_request_processor.SCMRequestProcessor.process_ticket" ) as mock_process_ticket: toddler.process(config, msg) mock_process_ticket.assert_called_with(issue) mock_pdc.assert_called_with(config) mock_pagure.assert_has_calls( [ call(config), call( { "pagure_url": "https://src.fedoraproject.org", "pagure_api_key": "Private API Key", } ), ] ) mock_fasjson.assert_called_with(config) mock_bugzilla.assert_called_with(config) class TestProcessTicket: """ Test class for `toddlers.plugins.scm_request_processor.SCMRequestProcessor.process_ticket` method. """ def setup(self): """ Initialize toddler. """ self.toddler = scm_request_processor.SCMRequestProcessor() self.toddler.pagure_io = Mock() self.toddler.dist_git = Mock() def test_process_ticket_invalid_json(self): """ Assert that invalid json in issue will end the processing. """ issue = { "id": 100, "content": "invalid JSON", "full_url": "https://blacklibrary.wh40k", } self.toddler.process_ticket(issue) self.toddler.pagure_io.close_issue.assert_called_with( 100, namespace=scm_request_processor.PROJECT_NAMESPACE, message="Invalid JSON provided", reason="Invalid", ) def test_process_ticket_invalid_slas(self): """ Assert that invalid SLAs in issue will end the processing. """ content = { "sls": {}, "branch": "branch", } issue = { "id": 100, "content": json.dumps(content), "full_url": "https://blacklibrary.wh40k", } with patch( "toddlers.plugins.scm_request_processor.SCMRequestProcessor.verify_slas" ) as mock_verify_slas: mock_verify_slas.side_effect = ValidationError("error") self.toddler.process_ticket(issue) mock_verify_slas.assert_called_with("branch", {}) self.toddler.pagure_io.close_issue.assert_called_with( 100, namespace=scm_request_processor.PROJECT_NAMESPACE, message="error", reason="Invalid", ) def test_process_ticket_missing_action(self): """ Assert that missing action in issue will end the processing. """ content = { "sls": {}, "branch": "branch", } issue = { "id": 100, "content": json.dumps(content), "full_url": "https://blacklibrary.wh40k", } self.toddler.process_ticket(issue) self.toddler.pagure_io.close_issue.assert_called_with( 100, namespace=scm_request_processor.PROJECT_NAMESPACE, message="Invalid or missing action field", reason="Invalid", ) def test_process_ticket_missing_sla(self): """ Assert that missing SLA for branch will end the processing. """ content = {"branch": "branch", "action": "new_repo"} issue = { "id": 100, "content": json.dumps(content), "full_url": "https://blacklibrary.wh40k", } self.toddler.process_ticket(issue) self.toddler.pagure_io.close_issue.assert_called_with( 100, namespace=scm_request_processor.PROJECT_NAMESPACE, message="Couldn't find standard SLA for branch 'branch'", reason="Invalid", ) def test_process_ticket_invalid_branch_name(self): """ Assert that invalid name for branch in specific namespace will end the processing. """ content = {"branch": "branch/", "action": "new_repo", "namespace": "flatpaks"} issue = { "id": 100, "content": json.dumps(content), "full_url": "https://blacklibrary.wh40k", } self.toddler.branch_slas = {"branch/": "SLA"} self.toddler.process_ticket(issue) self.toddler.pagure_io.close_issue.assert_called_with( 100, namespace=scm_request_processor.PROJECT_NAMESPACE, message=( "Only characters, numbers, periods, dashes, underscores, " "and pluses are allowed in flatpak branch names" ), reason="Invalid", ) def test_process_ticket_invalid_monitoring_setting(self): """ Assert that invalid monitoring setting for repo will end the processing. """ content = { "branch": "branch", "action": "new_repo", "namespace": "flatpaks", "monitor": "monitor", } issue = { "id": 100, "content": json.dumps(content), "full_url": "https://blacklibrary.wh40k", } self.toddler.branch_slas = {"branch": "SLA"} self.toddler.process_ticket(issue) self.toddler.pagure_io.close_issue.assert_called_with( 100, namespace=scm_request_processor.PROJECT_NAMESPACE, message='The monitor choice of "monitor" is invalid', reason="Invalid", ) def test_process_ticket_action_new_repo(self): """ Assert that action new_repo is correctly processed. """ content = { "branch": "branch", "action": "new_repo", } issue = { "id": 100, "content": json.dumps(content), "full_url": "https://blacklibrary.wh40k", } self.toddler.branch_slas = {"branch": "SLA"} with patch( "toddlers.plugins.scm_request_processor.SCMRequestProcessor.create_new_repo" ) as mock_new_repo: self.toddler.process_ticket(issue) content["sls"] = "SLA" mock_new_repo.assert_called_with( issue, content, initial_commit=True, ) def test_process_ticket_action_new_branch(self): """ Assert that action new_branch is correctly processed. """ content = { "branch": "branch", "action": "new_branch", } issue = { "id": 100, "content": json.dumps(content), "full_url": "https://blacklibrary.wh40k", } self.toddler.branch_slas = {"branch": "SLA"} with patch( "toddlers.plugins.scm_request_processor.SCMRequestProcessor.create_new_branch" ) as mock_new_branch: self.toddler.process_ticket(issue) content["sls"] = "SLA" mock_new_branch.assert_called_with( issue, content, ) class TestVerifySLAs: """ Test class for `toddlers.plugins.scm_request_processor.SCMRequestProcessor.verify_slas` method. """ def setup(self): """ Initialize toddler. """ self.toddler = scm_request_processor.SCMRequestProcessor() self.toddler.pagure_io = Mock() self.toddler.dist_git = Mock() def test_verify_slas_not_dict(self): """ Assert that SLA verification will fail if SLA isn't dict. """ sla = [] with pytest.raises(ValueError, match="The object provided is not a dict"): self.toddler.verify_slas("", sla) def test_verify_slas_no_branch(self): """ Assert that the validation will not fail if no branch is provided. This will fail later in create method, but the verification passes. """ sla = {} branch = None self.toddler.verify_slas(branch, sla) def test_verify_slas_branch_sla_correct(self): """ Assert that the validation will not fail if provided branch SLAs are correct. """ branch = "branch" sla = {"rawhide": "2022-06-01"} self.toddler.branch_slas = {branch: {"rawhide": "2022-06-01"}} self.toddler.verify_slas(branch, sla) def test_verify_slas_branch_sla_incorrect(self): """ Assert that the validation will fail if provided branch SLAs are incorrect. """ branch = "branch" sla = {"rawhide": "2022-01-01"} self.toddler.branch_slas = {branch: {"rawhide": "2022-06-01"}} error = 'The SLAs for the branch "{0}" are incorrect'.format(branch) with pytest.raises(ValidationError, match=error): self.toddler.verify_slas(branch, sla) def test_verify_slas_eol_not_string(self): """ Assert that the validation will fail if provided SLA EOL is not string. """ sla = {"rawhide": 100} error = 'The SL\'s EOL is not a string. It was type "{0}"'.format( type(100).__name__ ) with pytest.raises(ValidationError, match=error): self.toddler.verify_slas("", sla) def test_verify_slas_eol_invalid_format(self): """ Assert that the validation will fail if provided SLA EOL date is in invalid format. """ sla = {"rawhide": "01-01-2022"} error = 'The EOL date "{0}" is in an invalid format'.format("01-01-2022") with pytest.raises(ValidationError, match=error): self.toddler.verify_slas("", sla) def test_verify_slas_eol_expired(self): """ Assert that the validation will fail if provided SLA EOL date is already expired. """ sla = {"rawhide": "2022-01-01"} error = 'The SL "{0}" is already expired'.format("2022-01-01") with pytest.raises(ValidationError, match=error): self.toddler.verify_slas("", sla) def test_verify_slas_eol_date_invalid(self): """ Assert that the validation will fail if provided SLA EOL date is invalid. """ sla = {"rawhide": "2050-01-01"} error = 'The SL "{0}" must expire on June 1st or December 1st'.format( "2050-01-01" ) with pytest.raises(ValidationError, match=error): self.toddler.verify_slas("", sla) @patch("toddlers.utils.pdc.get_sla") def test_verify_slas_not_in_pdc(self, mock_pdc): """ Assert that the validation will fail if SLA is not in PDC. """ sla = {"rawhide": "2050-06-01"} error = 'The SL "{0}" is not in PDC'.format("rawhide") mock_pdc.return_value = None with pytest.raises(ValidationError, match=error): self.toddler.verify_slas("", sla) mock_pdc.assert_called_with("rawhide") @patch("toddlers.utils.pdc.get_sla") def test_verify_slas_in_pdc(self, mock_pdc): """ Assert that the validation will pass if SLA is in PDC. """ sla = {"rawhide": "2050-06-01"} mock_pdc.return_value = sla self.toddler.verify_slas("", sla) mock_pdc.assert_called_with("rawhide") class TestCreateNewRepo: """ Test class for `toddlers.plugins.scm_request_processor.SCMRequestProcessor.create_new_repo` method. """ def setup(self): """ Initialize toddler. """ self.toddler = scm_request_processor.SCMRequestProcessor() self.toddler.pagure_io = Mock() self.toddler.dist_git = Mock() def test_create_new_repo_missing_required_key(self): """ Assert that ticket will be closed if required key is missing in request. """ issue = { "id": 100, } self.toddler.create_new_repo(issue, {}) self.toddler.pagure_io.close_issue.assert_called_with( 100, namespace=scm_request_processor.PROJECT_NAMESPACE, message="Invalid body, missing required field: repo", reason="Invalid", ) def test_create_new_repo_invalid_repo_name(self): """ Assert that ticket will be closed if provided repository name is invalid. """ issue = {"id": 100, "user": {"name": "zlopez"}} json = { "repo": "+a", "branch": "rawhide", "namespace": "namespace", "bug_id": "123", "action": "new_repo", "sls": {"rawhide": "2050-06-01"}, "monitor": "monitor", } self.toddler.create_new_repo(issue, json) error = ( "The repository name is invalid. It must be at least two " "characters long with only letters, numbers, hyphens, " "underscores, plus signs, and/or periods. Please note that " "the project cannot start with a period or a plus sign." ) self.toddler.pagure_io.close_issue.assert_called_with( 100, namespace=scm_request_processor.PROJECT_NAMESPACE, message=error, reason="Invalid", ) def test_create_new_repo_exception_validated_by_invalid_user(self): """ Assert that processing will be interrupted if the ticket is validated by wrong user. """ # Preparation user = "zlopez" invalid_user = "Tzeentch" issue = { "id": 100, "user": {"name": user}, "comments": [ { "comment": "valid", "user": {"name": invalid_user}, "notification": False, } ], } repo = "repo" branch = "main" namespace = "tests" bug_id = "" action = "new_repo" sls = {branch: "2050-06-01"} monitor = "monitor" exception = False json = { "repo": repo, "branch": branch, "namespace": namespace, "bug_id": bug_id, "action": action, "sls": sls, "monitor": monitor, "exception": exception, } self.toddler.pagure_io.get_project_contributors.return_value = { "users": {"admin": [user], "commit": [], "ticket": []} } self.toddler.validation_comment = "valid" self.toddler.create_new_repo(issue, json) # asserts self.toddler.pagure_io.get_project_contributors.assert_called_with( scm_request_processor.PROJECT_NAMESPACE.split("/")[0], scm_request_processor.PROJECT_NAMESPACE.split("/")[1], ) def test_create_new_repo_exception_no_validation_comment(self): """ Assert that processing will be interrupted if the ticket validation comment is not found. """ # Preparation user = "zlopez" issue = { "id": 100, "user": {"name": user}, "comments": [ {"comment": "comment", "user": {"name": user}, "notification": False} ], } repo = "repo" branch = "main" namespace = "tests" bug_id = "" action = "new_repo" sls = {branch: "2050-06-01"} monitor = "monitor" exception = False json = { "repo": repo, "branch": branch, "namespace": namespace, "bug_id": bug_id, "action": action, "sls": sls, "monitor": monitor, "exception": exception, } self.toddler.pagure_io.get_project_contributors.return_value = { "users": {"admin": [user], "commit": [], "ticket": []} } self.toddler.validation_comment = "valid" self.toddler.create_new_repo(issue, json) # asserts self.toddler.pagure_io.get_project_contributors.assert_called_with( scm_request_processor.PROJECT_NAMESPACE.split("/")[0], scm_request_processor.PROJECT_NAMESPACE.split("/")[1], ) def test_create_new_repo_exception_notify_maintainers(self): """ Assert that comment will be added when the ticket needs manual validation. """ # Preparation user = "zlopez" issue = {"id": 100, "user": {"name": user}, "comments": []} repo = "repo" branch = "main" namespace = "tests" bug_id = "" action = "new_repo" sls = {branch: "2050-06-01"} monitor = "monitor" exception = False json = { "repo": repo, "branch": branch, "namespace": namespace, "bug_id": bug_id, "action": action, "sls": sls, "monitor": monitor, "exception": exception, } self.toddler.pagure_io.get_project_contributors.return_value = { "users": {"admin": [user, "Tzeentch"], "commit": [], "ticket": []} } self.toddler.ping_comment = "Look at this comment {maintainers}" self.toddler.create_new_repo(issue, json) # asserts self.toddler.pagure_io.get_project_contributors.assert_called_with( scm_request_processor.PROJECT_NAMESPACE.split("/")[0], scm_request_processor.PROJECT_NAMESPACE.split("/")[1], ) message = "Look at this comment @Tzeentch @{}".format(user) self.toddler.pagure_io.add_comment_to_issue.assert_called_with( 100, namespace=scm_request_processor.PROJECT_NAMESPACE, comment=message ) def test_create_new_repo_exception_not_valid_notification_comment_present(self): """ Assert that comment will not be added if the toddler already commented on the ticket. """ # Preparation user = "zlopez" issue = { "id": 100, "user": {"name": user}, "comments": [ {"comment": "comment", "user": {"name": user}, "notification": False} ], } repo = "repo" branch = "main" namespace = "tests" bug_id = "" action = "new_repo" sls = {branch: "2050-06-01"} monitor = "monitor" exception = False json = { "repo": repo, "branch": branch, "namespace": namespace, "bug_id": bug_id, "action": action, "sls": sls, "monitor": monitor, "exception": exception, } self.toddler.pagure_io.get_project_contributors.return_value = { "users": {"admin": ["Tzeentch"], "commit": [], "ticket": []} } self.toddler.pagure_user = user self.toddler.create_new_repo(issue, json) # asserts self.toddler.pagure_io.get_project_contributors.assert_called_with( scm_request_processor.PROJECT_NAMESPACE.split("/")[0], scm_request_processor.PROJECT_NAMESPACE.split("/")[1], ) self.toddler.pagure_io.add_comment_to_issue.assert_not_called() def test_create_new_repo_missing_bug_id(self): """ Assert that ticket will be closed if Bugzilla bug id is not provided. """ issue = {"id": 100, "user": {"name": "zlopez"}} json = { "repo": "repo", "branch": "rawhide", "namespace": "namespace", "bug_id": "", "action": "new_repo", "sls": {"rawhide": "2050-06-01"}, "monitor": "monitor", } self.toddler.create_new_repo(issue, json) self.toddler.pagure_io.close_issue.assert_called_with( 100, namespace=scm_request_processor.PROJECT_NAMESPACE, message="An invalid Bugzilla bug was provided", reason="Invalid", ) @patch( "toddlers.plugins.scm_request_processor.SCMRequestProcessor.validate_review_bug" ) def test_create_new_repo_invalid_review_bug(self, mock_validate_review_bug): """ Assert that ticket will be closed if Bugzilla bug is not valid. """ issue = {"id": 100, "user": {"name": "zlopez"}} json = { "repo": "repo", "branch": "rawhide", "namespace": "namespace", "bug_id": "123", "action": "new_repo", "sls": {"rawhide": "2050-06-01"}, "monitor": "monitor", } mock_validate_review_bug.side_effect = ValidationError("error") self.toddler.create_new_repo(issue, json) mock_validate_review_bug.assert_called_with( "123", "repo", "rawhide", namespace="namespace", pagure_user="zlopez" ) self.toddler.pagure_io.close_issue.assert_called_with( 100, namespace=scm_request_processor.PROJECT_NAMESPACE, message="error", reason="Invalid", ) @patch( "toddlers.plugins.scm_request_processor.SCMRequestProcessor.valid_epel_package" ) @patch( "toddlers.plugins.scm_request_processor.SCMRequestProcessor.validate_review_bug" ) def test_create_new_repo_invalid_epel( self, mock_validate_review_bug, mock_valid_epel_package ): """ Assert that ticket will be closed if repo is invalid EPEL repo. """ issue = {"id": 100, "user": {"name": "zlopez"}} repo = "repo" branch = "epel8" namespace = "rpms" bug_id = "123" action = "new_repo" sls = {"rawhide": "2050-06-01"} monitor = "monitor" exception = False json = { "repo": repo, "branch": branch, "namespace": namespace, "bug_id": bug_id, "action": action, "sls": sls, "monitor": monitor, "exception": exception, } mock_valid_epel_package.return_value = False self.toddler.create_new_repo(issue, json) mock_valid_epel_package.assert_called_with(repo, branch) mock_validate_review_bug.assert_called_with( bug_id, repo, branch, namespace=namespace, pagure_user="zlopez" ) self.toddler.pagure_io.close_issue.assert_called_with( 100, namespace=scm_request_processor.PROJECT_NAMESPACE, message=scm_request_processor.INVALID_EPEL_ERROR, reason="Invalid", ) @patch( "toddlers.plugins.scm_request_processor.SCMRequestProcessor.validate_review_bug" ) def test_create_new_repo_requester_not_in_dist_git(self, mock_validate_review_bug): """ Assert that ticket will be commented on when requester doesn't have a valid dist git account. """ issue = {"id": 100, "user": {"name": "zlopez"}} repo = "repo" branch = "rawhide" namespace = "rpms" bug_id = "123" action = "new_repo" sls = {"rawhide": "2050-06-01"} monitor = "monitor" exception = False json = { "repo": repo, "branch": branch, "namespace": namespace, "bug_id": bug_id, "action": action, "sls": sls, "monitor": monitor, "exception": exception, } self.toddler.dist_git.user_exists.return_value = False self.toddler.dist_git._pagure_url = "https://src.fedoraproject.org" self.toddler.create_new_repo(issue, json) mock_validate_review_bug.assert_called_with( bug_id, repo, branch, namespace=namespace, pagure_user="zlopez" ) self.toddler.dist_git.user_exists.assert_called_with("zlopez") message = "@zlopez needs to login to {0} to sync accounts before we can proceed.".format( self.toddler.dist_git._pagure_url ) self.toddler.pagure_io.add_comment_to_issue.assert_called_with( 100, namespace=scm_request_processor.PROJECT_NAMESPACE, comment=message, ) @patch( "toddlers.plugins.scm_request_processor.SCMRequestProcessor.validate_review_bug" ) def test_create_new_repo_project_exists(self, mock_validate_review_bug): """ Assert that ticket will be closed when repo already exists in dist git. """ issue = {"id": 100, "user": {"name": "zlopez"}} repo = "repo" branch = "rawhide" namespace = "rpms" bug_id = "123" action = "new_repo" sls = {"rawhide": "2050-06-01"} monitor = "monitor" exception = False json = { "repo": repo, "branch": branch, "namespace": namespace, "bug_id": bug_id, "action": action, "sls": sls, "monitor": monitor, "exception": exception, } self.toddler.dist_git.get_project.return_value = "project" self.toddler.create_new_repo(issue, json) mock_validate_review_bug.assert_called_with( bug_id, repo, branch, namespace=namespace, pagure_user="zlopez" ) self.toddler.dist_git.get_project.assert_called_with(namespace, repo) self.toddler.pagure_io.close_issue.assert_called_with( 100, namespace=scm_request_processor.PROJECT_NAMESPACE, message="The Pagure project already exists", reason="Invalid", ) @patch( "toddlers.plugins.scm_request_processor.SCMRequestProcessor.validate_review_bug" ) def test_create_new_repo_master_branch(self, mock_validate_review_bug): """ Assert that ticket will be closed when branch is set to master branch. Master branch is no longer allowed. """ issue = {"id": 100, "user": {"name": "zlopez"}} repo = "repo" branch = "master" namespace = "rpms" bug_id = "123" action = "new_repo" sls = {"rawhide": "2050-06-01"} monitor = "monitor" exception = False json = { "repo": repo, "branch": branch, "namespace": namespace, "bug_id": bug_id, "action": action, "sls": sls, "monitor": monitor, "exception": exception, } self.toddler.dist_git.get_project.return_value = None self.toddler.create_new_repo(issue, json) mock_validate_review_bug.assert_called_with( bug_id, repo, branch, namespace=namespace, pagure_user="zlopez" ) self.toddler.dist_git.get_project.assert_called_with(namespace, repo) self.toddler.pagure_io.close_issue.assert_called_with( 100, namespace=scm_request_processor.PROJECT_NAMESPACE, message="Branch `master` cannot be created, please request the right branch.", reason="Invalid", ) @patch( "toddlers.plugins.scm_request_processor.SCMRequestProcessor.validate_review_bug" ) def test_create_new_repo_unsupported_namespace(self, mock_validate_review_bug): """ Assert that ticket will be closed when requested namespace is not recognized. """ issue = {"id": 100, "user": {"name": "zlopez"}} repo = "repo" branch = "rawhide" namespace = "invalid" bug_id = "123" action = "new_repo" sls = {"rawhide": "2050-06-01"} monitor = "monitor" exception = False json = { "repo": repo, "branch": branch, "namespace": namespace, "bug_id": bug_id, "action": action, "sls": sls, "monitor": monitor, "exception": exception, } self.toddler.dist_git.get_project.return_value = None self.toddler.create_new_repo(issue, json) mock_validate_review_bug.assert_called_with( bug_id, repo, branch, namespace=namespace, pagure_user="zlopez" ) self.toddler.dist_git.get_project.assert_called_with(namespace, repo) error = ( "The requested namespace '{0}' is not recognized. " "Currently supported namespaces are: " "rpms, container, flatpaks, modules, tests.".format(namespace) ) self.toddler.pagure_io.close_issue.assert_called_with( 100, namespace=scm_request_processor.PROJECT_NAMESPACE, message=error, reason="Invalid", ) @pytest.mark.parametrize( "namespace, branch", [ ("rpms", "rawhide"), ("container", "rawhide"), ("flatpaks", "stable"), ("modules", "rawhide"), ], ) @patch("toddlers.plugins.scm_request_processor.pdc") def test_create_new_repo_namespaces(self, mock_pdc, namespace, branch): """ Assert that ticket will be processed when everything is in order and namespace is correct. """ # Preparation user = "zlopez" issue = { "id": 100, "user": {"name": user}, "comments": [ {"comment": "valid", "user": {"name": user}, "notification": False} ], } repo = "repo" bug_id = "" action = "new_repo" sls = {branch: "2050-06-01"} monitor = "monitor" exception = True json = { "repo": repo, "branch": branch, "namespace": namespace, "bug_id": bug_id, "action": action, "sls": sls, "monitor": monitor, "exception": exception, } dist_git_url = "https://src.fp.o" self.toddler.dist_git.get_project.return_value = None self.toddler.dist_git._pagure_url = dist_git_url self.toddler.pagure_io.get_project_contributors.return_value = { "users": {"admin": [user], "commit": [], "ticket": []} } self.toddler.validation_comment = "valid" # Method to test self.toddler.create_new_repo(issue, json) # asserts self.toddler.dist_git.get_project.assert_called_with(namespace, repo) self.toddler.dist_git.new_project.assert_called_with( namespace, repo, "", "", branch, initial_commit=True, alias=True ) self.toddler.dist_git.set_monitoring_status.assert_called_with( namespace, repo, monitor ) self.toddler.dist_git.change_project_main_admin.assert_called_with( namespace, repo, "zlopez" ) mock_pdc.new_global_component.assert_called_with( repo, "{0}/{1}/{2}".format(dist_git_url, namespace, repo) ) mock_pdc.new_branch.assert_called_with(repo, branch, namespace.rstrip("s")) mock_pdc.new_sla_to_branch.assert_called_with( branch, sls[branch], repo, branch, namespace.rstrip("s") ) message = "The Pagure repository was created at {0}/{1}/{2}".format( dist_git_url, namespace, repo ) self.toddler.pagure_io.close_issue.assert_called_with( 100, namespace=scm_request_processor.PROJECT_NAMESPACE, message=message, reason="Processed", ) def test_create_new_repo_tests_namespace(self): """ Assert that ticket will be processed when everything is in order and namespace is set to tests. """ # Preparation user = "zlopez" issue = { "id": 100, "user": {"name": user}, "comments": [ {"comment": "valid", "user": {"name": user}, "notification": False} ], } repo = "repo" branch = "main" namespace = "tests" bug_id = "" action = "new_repo" sls = {branch: "2050-06-01"} monitor = "monitor" exception = False json = { "repo": repo, "branch": branch, "namespace": namespace, "bug_id": bug_id, "action": action, "sls": sls, "monitor": monitor, "exception": exception, } dist_git_url = "https://src.fp.o" self.toddler.dist_git.get_project.return_value = None self.toddler.dist_git._pagure_url = dist_git_url self.toddler.pagure_io.get_project_contributors.return_value = { "users": {"admin": [user], "commit": [], "ticket": []} } self.toddler.validation_comment = "valid" self.toddler.create_new_repo(issue, json) # asserts self.toddler.pagure_io.get_project_contributors.assert_called_with( scm_request_processor.PROJECT_NAMESPACE.split("/")[0], scm_request_processor.PROJECT_NAMESPACE.split("/")[1], ) self.toddler.dist_git.get_project.assert_called_with(namespace, repo) self.toddler.dist_git.new_project.assert_called_with( namespace, repo, "", "", branch, initial_commit=True, alias=True ) self.toddler.dist_git.set_monitoring_status.assert_called_with( namespace, repo, monitor ) self.toddler.dist_git.change_project_main_admin.assert_called_with( namespace, repo, "zlopez" ) message = "The Pagure repository was created at {0}/{1}/{2}".format( dist_git_url, namespace, repo ) self.toddler.pagure_io.close_issue.assert_called_with( 100, namespace=scm_request_processor.PROJECT_NAMESPACE, message=message, reason="Processed", ) @patch("toddlers.plugins.scm_request_processor.bugzilla_system") @patch("toddlers.plugins.scm_request_processor.pdc") @patch( "toddlers.plugins.scm_request_processor.SCMRequestProcessor.validate_review_bug" ) def test_create_new_repo_non_default_branch( self, mock_validate_review_bug, mock_pdc, mock_bz ): """ Assert that ticket will be processed when everything is in order and requested branch is not default. """ issue = {"id": 100, "user": {"name": "zlopez"}} repo = "repo" branch = "f35" namespace = "rpms" bug_id = "123" action = "new_repo" sls = {branch: "2050-06-01"} monitor = "monitor" exception = False json = { "repo": repo, "branch": branch, "namespace": namespace, "bug_id": bug_id, "action": action, "sls": sls, "monitor": monitor, "exception": exception, } self.toddler.branch_slas = {"rawhide": {"rawhide": "2050-06-01"}} dist_git_url = "https://src.fp.o" self.toddler.dist_git.get_project.return_value = None self.toddler.dist_git._pagure_url = dist_git_url self.toddler.create_new_repo(issue, json) # asserts mock_validate_review_bug.assert_called_with( bug_id, repo, branch, namespace=namespace, pagure_user="zlopez" ) self.toddler.dist_git.get_project.assert_called_with(namespace, repo) self.toddler.dist_git.new_project.assert_called_with( namespace, repo, "", "", "rawhide", initial_commit=True, alias=True ) self.toddler.dist_git.set_monitoring_status.assert_called_with( namespace, repo, monitor ) self.toddler.dist_git.change_project_main_admin.assert_called_with( namespace, repo, "zlopez" ) mock_pdc.new_global_component.assert_called_with( repo, "{0}/{1}/{2}".format(dist_git_url, namespace, repo) ) mock_pdc.new_branch.assert_has_calls( [ call(repo, "rawhide", namespace.rstrip("s")), call(repo, branch, namespace.rstrip("s")), ] ) mock_pdc.new_sla_to_branch.assert_has_calls( [ call("rawhide", "2050-06-01", repo, "rawhide", namespace.rstrip("s")), call(branch, sls[branch], repo, branch, namespace.rstrip("s")), ] ) message = ( "The Pagure repository was created at {0}/{1}/{2}. " 'You may commit to the branch "{3}" in about ' "10 minutes.".format(dist_git_url, namespace, repo, branch) ) self.toddler.pagure_io.close_issue.assert_called_with( 100, namespace=scm_request_processor.PROJECT_NAMESPACE, message=message, reason="Processed", ) mock_bz.comment_on_bug.assert_called_with(bug_id, message) @patch("toddlers.plugins.scm_request_processor.bugzilla_system") @patch("toddlers.plugins.scm_request_processor.pdc") @patch( "toddlers.plugins.scm_request_processor.SCMRequestProcessor.validate_review_bug" ) def test_create_new_repo_default_brach( self, mock_validate_review_bug, mock_pdc, mock_bz ): """ Assert that repo will be created with default branch when everything is in order. """ issue = {"id": 100, "user": {"name": "zlopez"}} repo = "repo" branch = "rawhide" namespace = "rpms" bug_id = "123" action = "new_repo" sls = {branch: "2050-06-01"} monitor = "monitor" exception = False json = { "repo": repo, "branch": branch, "namespace": namespace, "bug_id": bug_id, "action": action, "sls": sls, "monitor": monitor, "exception": exception, } dist_git_url = "https://src.fp.o" self.toddler.dist_git.get_project.return_value = None self.toddler.dist_git._pagure_url = dist_git_url self.toddler.create_new_repo(issue, json) # asserts mock_validate_review_bug.assert_called_with( bug_id, repo, branch, namespace=namespace, pagure_user="zlopez" ) self.toddler.dist_git.get_project.assert_called_with(namespace, repo) self.toddler.dist_git.new_project.assert_called_with( namespace, repo, "", "", branch, initial_commit=True, alias=True ) self.toddler.dist_git.set_monitoring_status.assert_called_with( namespace, repo, monitor ) self.toddler.dist_git.change_project_main_admin.assert_called_with( namespace, repo, "zlopez" ) mock_pdc.new_global_component.assert_called_with( repo, "{0}/{1}/{2}".format(dist_git_url, namespace, repo) ) mock_pdc.new_branch.assert_called_with(repo, branch, namespace.rstrip("s")) mock_pdc.new_sla_to_branch.assert_called_with( branch, sls[branch], repo, branch, namespace.rstrip("s") ) message = "The Pagure repository was created at {0}/{1}/{2}".format( dist_git_url, namespace, repo ) self.toddler.pagure_io.close_issue.assert_called_with( 100, namespace=scm_request_processor.PROJECT_NAMESPACE, message=message, reason="Processed", ) mock_bz.comment_on_bug.assert_called_with(bug_id, message) class TestCreateNewBranch: """ Test class for `toddlers.plugins.scm_request_processor.SCMRequestProcessor.create_new_branch` method. """ def setup(self): """ Initialize toddler. """ self.toddler = scm_request_processor.SCMRequestProcessor() self.toddler.pagure_io = Mock() self.toddler.dist_git = Mock() def test_create_new_branch_missing_required_key(self): """ Assert that ticket will be closed if required key is missing in request. """ issue = { "id": 100, } self.toddler.create_new_branch(issue, {}) self.toddler.pagure_io.close_issue.assert_called_with( 100, namespace=scm_request_processor.PROJECT_NAMESPACE, message="Invalid body, missing required field: action", reason="Invalid", ) def test_create_new_branch_no_contributors(self): """ Assert that ticket will be closed if contributors are not retrieved. """ issue = { "id": 100, } repo = "repo" branch = "rawhide" namespace = "rpms" action = "new_branch" sls = {branch: "2050-06-01"} json = { "repo": repo, "branch": branch, "namespace": namespace, "action": action, "sls": sls, } self.toddler.dist_git.get_project_contributors.return_value = None self.toddler.create_new_branch(issue, json) # Asserts self.toddler.dist_git.get_project_contributors.assert_called_with( namespace, repo ) self.toddler.pagure_io.close_issue.assert_called_with( 100, namespace=scm_request_processor.PROJECT_NAMESPACE, message="The dist git repository doesn't exist", reason="Invalid", ) @patch( "toddlers.plugins.scm_request_processor.SCMRequestProcessor.valid_epel_package" ) def test_create_new_branch_invalid_epel(self, mock_valid_epel_package): """ Assert that ticket will be closed if repo is invalid EPEL repo. """ issue = {"id": 100, "user": {"name": "zlopez"}} repo = "repo" branch = "epel8" namespace = "rpms" action = "new_branch" sls = {"rawhide": "2050-06-01"} json = { "repo": repo, "branch": branch, "namespace": namespace, "action": action, "sls": sls, } mock_valid_epel_package.return_value = False self.toddler.create_new_branch(issue, json) # Asserts self.toddler.dist_git.get_project_contributors.assert_called_with( namespace, repo ) mock_valid_epel_package.assert_called_with(repo, branch) self.toddler.pagure_io.close_issue.assert_called_with( 100, namespace=scm_request_processor.PROJECT_NAMESPACE, message=scm_request_processor.INVALID_EPEL_ERROR, reason="Invalid", ) @patch("toddlers.plugins.scm_request_processor.fedora_account") def test_create_new_branch_requester_is_not_maintainer(self, mock_fedora_account): """ Assert that ticket will be closed if requester is not a maintainer of package. """ issue = {"id": 100, "user": {"name": "zlopez"}} repo = "repo" branch = "rawhide" namespace = "rpms" action = "new_branch" sls = {"rawhide": "2050-06-01"} json = { "repo": repo, "branch": branch, "namespace": namespace, "action": action, "sls": sls, } self.toddler.dist_git.get_project_contributors.return_value = { "users": {"admin": [], "commit": [], "collaborators": []}, "groups": {"admin": ["group"], "commit": [], "collaborators": []}, } mock_fedora_account.user_member_of.return_value = False self.toddler.create_new_branch(issue, json) # Asserts self.toddler.dist_git.get_project_contributors.assert_called_with( namespace, repo ) mock_fedora_account.user_member_of.assert_called_with( mock_fedora_account.get_user_by_username(), "group" ) self.toddler.pagure_io.close_issue.assert_called_with( 100, namespace=scm_request_processor.PROJECT_NAMESPACE, message="zlopez is not a maintainer of the {0} package".format(repo), reason="Invalid", ) @patch("toddlers.plugins.scm_request_processor.TemporaryDirectory") @patch("toddlers.plugins.scm_request_processor.git") @patch("toddlers.plugins.scm_request_processor.fedora_account") @patch("toddlers.plugins.scm_request_processor.pdc") def test_create_new_branch_default_branch_missing( self, mock_pdc, mock_fedora_account, mock_git, mock_temp_dir ): """ Assert that ticket will be closed if there is no default branch in project and new branch in git should be created. """ issue = {"id": 100, "user": {"name": "zlopez"}} repo = "repo" branch = "rawhide" namespace = "rpms" action = "new_branch" sls = {"rawhide": "2050-06-01"} json = { "repo": repo, "branch": branch, "namespace": namespace, "action": action, "sls": sls, } self.toddler.dist_git.get_project_contributors.return_value = { "users": {"admin": [], "commit": [], "collaborators": []}, "groups": {"admin": ["group"], "commit": [], "collaborators": []}, } self.toddler.dist_git.get_default_branch.return_value = None self.toddler.dist_git._pagure_url = "https://fp.o" mock_fedora_account.user_member_of.return_value = True mock_dir = MagicMock() mock_dir.__enter__.return_value = "dir" mock_temp_dir.return_value = mock_dir self.toddler.create_new_branch(issue, json) # Asserts self.toddler.dist_git.get_project_contributors.assert_called_with( namespace, repo ) mock_fedora_account.user_member_of.assert_called_with( mock_fedora_account.get_user_by_username(), "group" ) mock_pdc.new_global_component.assert_called_with( repo, "{0}/{1}/{2}".format(self.toddler.dist_git._pagure_url, namespace, repo), ) mock_pdc.new_branch.assert_called_with( repo, branch, namespace.strip().rstrip("s") ) mock_pdc.new_sla_to_branch.assert_called_with( branch, "2050-06-01", repo, branch, namespace.strip().rstrip("s") ) mock_git.clone_repo.assert_called_with( "{0}/{1}/{2}".format(self.toddler.dist_git._pagure_url, namespace, repo), "dir", ) self.toddler.dist_git.get_default_branch.assert_called_with(namespace, repo) self.toddler.pagure_io.close_issue.assert_called_with( 100, namespace=scm_request_processor.PROJECT_NAMESPACE, message="There is no default branch set for {0}/{1}".format( namespace, repo ), reason="Invalid", ) @patch("toddlers.plugins.scm_request_processor.TemporaryDirectory") @patch("toddlers.plugins.scm_request_processor.git") @patch("toddlers.plugins.scm_request_processor.fedora_account") @patch("toddlers.plugins.scm_request_processor.pdc") def test_create_new_branch_with_git( self, mock_pdc, mock_fedora_account, mock_git, mock_temp_dir ): """ Assert that ticket will be processed and branch created in git. """ issue = {"id": 100, "user": {"name": "zlopez"}} repo = "repo" default_branch = "rawhide" branch = "f36" namespace = "rpms" action = "new_branch" sls = {"rawhide": "2050-06-01"} json = { "repo": repo, "branch": branch, "namespace": namespace, "action": action, "sls": sls, } self.toddler.dist_git.get_project_contributors.return_value = { "users": {"admin": [], "commit": [], "collaborators": []}, "groups": {"admin": ["group"], "commit": [], "collaborators": []}, } self.toddler.dist_git.get_default_branch.return_value = default_branch self.toddler.dist_git._pagure_url = "https://fp.o" mock_fedora_account.user_member_of.return_value = True mock_dir = MagicMock() mock_dir.__enter__.return_value = "dir" mock_temp_dir.return_value = mock_dir mock_git_repo = Mock() mock_git_repo.first_commit.return_value = "SHA256" mock_git.clone_repo.return_value = mock_git_repo self.toddler.create_new_branch(issue, json) # Asserts self.toddler.dist_git.get_project_contributors.assert_called_with( namespace, repo ) mock_fedora_account.user_member_of.assert_called_with( mock_fedora_account.get_user_by_username(), "group" ) mock_pdc.new_global_component.assert_called_with( repo, "{0}/{1}/{2}".format(self.toddler.dist_git._pagure_url, namespace, repo), ) mock_pdc.new_branch.assert_called_with( repo, branch, namespace.strip().rstrip("s") ) mock_pdc.new_sla_to_branch.assert_called_with( default_branch, "2050-06-01", repo, branch, namespace.strip().rstrip("s") ) mock_git.clone_repo.assert_called_with( "{0}/{1}/{2}".format(self.toddler.dist_git._pagure_url, namespace, repo), "dir", ) self.toddler.dist_git.get_default_branch.assert_called_with(namespace, repo) mock_git_repo.first_commit.assert_called_with(default_branch) self.toddler.dist_git.new_branch.assert_called_with( namespace, repo, branch, from_commit="SHA256" ) message = ( "The branch was created in PDC and git. It " "may take up to 10 minutes before you have " "write access on the branch." ) self.toddler.pagure_io.close_issue.assert_called_with( 100, namespace=scm_request_processor.PROJECT_NAMESPACE, message=message, reason="Processed", ) @patch("toddlers.plugins.scm_request_processor.fedora_account") @patch("toddlers.plugins.scm_request_processor.pdc") def test_create_new_branch_only_PDC(self, mock_pdc, mock_fedora_account): """ Assert that ticket will be processed and branch created in PDC. """ issue = {"id": 100, "user": {"name": "zlopez"}} repo = "repo" branch = "rawhide" namespace = "rpms" action = "new_branch" sls = {"rawhide": "2050-06-01"} json = { "repo": repo, "branch": branch, "namespace": namespace, "action": action, "sls": sls, "create_git_branch": False, } self.toddler.dist_git.get_project_contributors.return_value = { "users": {"admin": [], "commit": [], "collaborators": []}, "groups": {"admin": ["group"], "commit": [], "collaborators": []}, } self.toddler.dist_git._pagure_url = "https://fp.o" mock_fedora_account.user_member_of.return_value = True self.toddler.create_new_branch(issue, json) # Asserts self.toddler.dist_git.get_project_contributors.assert_called_with( namespace, repo ) mock_fedora_account.user_member_of.assert_called_with( mock_fedora_account.get_user_by_username(), "group" ) mock_pdc.new_global_component.assert_called_with( repo, "{0}/{1}/{2}".format(self.toddler.dist_git._pagure_url, namespace, repo), ) mock_pdc.new_branch.assert_called_with( repo, branch, namespace.strip().rstrip("s") ) mock_pdc.new_sla_to_branch.assert_called_with( branch, "2050-06-01", repo, branch, namespace.strip().rstrip("s") ) message = ( "The branch in PDC was created. Pagure is still processing " "the request, but in about 10 minutes, you may create the " "branch in Pagure using git." ) self.toddler.pagure_io.close_issue.assert_called_with( 100, namespace=scm_request_processor.PROJECT_NAMESPACE, message=message, reason="Processed", ) @patch("toddlers.plugins.scm_request_processor.bugzilla_system") @patch("toddlers.plugins.scm_request_processor.fedora_account") @patch("toddlers.plugins.scm_request_processor.pdc") def test_create_new_branch_comment_on_bug( self, mock_pdc, mock_fedora_account, mock_bz ): """ Assert that ticket will be processed and comment added to bugzilla bug, if provided. """ issue = {"id": 100, "user": {"name": "zlopez"}} repo = "repo" branch = "rawhide" namespace = "rpms" action = "new_branch" sls = {"rawhide": "2050-06-01"} bug_id = "123" json = { "repo": repo, "branch": branch, "namespace": namespace, "action": action, "bug_id": bug_id, "sls": sls, "create_git_branch": False, } self.toddler.dist_git.get_project_contributors.return_value = { "users": {"admin": [], "commit": [], "collaborators": []}, "groups": {"admin": ["group"], "commit": [], "collaborators": []}, } self.toddler.dist_git._pagure_url = "https://fp.o" mock_fedora_account.user_member_of.return_value = True self.toddler.create_new_branch(issue, json) # Asserts self.toddler.dist_git.get_project_contributors.assert_called_with( namespace, repo ) mock_fedora_account.user_member_of.assert_called_with( mock_fedora_account.get_user_by_username(), "group" ) mock_pdc.new_global_component.assert_called_with( repo, "{0}/{1}/{2}".format(self.toddler.dist_git._pagure_url, namespace, repo), ) mock_pdc.new_branch.assert_called_with( repo, branch, namespace.strip().rstrip("s") ) mock_pdc.new_sla_to_branch.assert_called_with( branch, "2050-06-01", repo, branch, namespace.strip().rstrip("s") ) message = ( "The branch in PDC was created. Pagure is still processing " "the request, but in about 10 minutes, you may create the " "branch in Pagure using git." ) self.toddler.pagure_io.close_issue.assert_called_with( 100, namespace=scm_request_processor.PROJECT_NAMESPACE, message=message, reason="Processed", ) mock_bz.comment_on_bug.assert_called_with(bug_id, message) class TestValidateReviewBug: """ Test class for `toddlers.plugins.scm_request_processor.SCMRequestProcessor.validate_review_bug` method. """ def setup(self): """ Initialize toddler. """ self.toddler = scm_request_processor.SCMRequestProcessor() self.toddler.pagure_io = Mock() self.toddler.dist_git = Mock() @patch("toddlers.plugins.scm_request_processor.bugzilla_system") def test_validate_review_bug_bug_retrieve_error(self, mock_bz): """ Assert that error will be raised when bug couldn't be retrieved. """ # Preparation repo = "repo" branch = "rawhide" bug_id = "123" mock_bz.get_bug.side_effect = Exception("error") error = ( "The Bugzilla bug could not be verified. The following " "error was encountered: error" ) # Method to test with pytest.raises(ValidationError, match=error): self.toddler.validate_review_bug(bug_id, repo, branch) # Asserts mock_bz.get_bug.assert_called_with(bug_id) @patch("toddlers.plugins.scm_request_processor.bugzilla_system") def test_validate_review_bug_no_bug(self, mock_bz): """ Assert that error will be raised when the bug is not found in bugzilla. """ # Preparation repo = "repo" branch = "rawhide" bug_id = "123" mock_bz.get_bug.return_value = None error = "The Bugzilla bug doesn't exist" # Method to test with pytest.raises(ValidationError, match=error): self.toddler.validate_review_bug(bug_id, repo, branch) # Asserts mock_bz.get_bug.assert_called_with(bug_id) @patch("toddlers.plugins.scm_request_processor.bugzilla_system") def test_validate_review_bug_epel_branch_invalid_bug(self, mock_bz): """ Assert that error will be raised when the requested branch is for EPEL, but the review bug isn't. """ # Preparation repo = "repo" branch = "epel8" bug_id = "123" mock_bug = Mock() mock_bug.product = "Fedora" mock_bz.get_bug.return_value = mock_bug error = ( 'The Bugzilla bug is for "Fedora" but the ' "requested branch is an EPEL branch" ) # Method to test with pytest.raises(ValidationError, match=error): self.toddler.validate_review_bug(bug_id, repo, branch) # Asserts mock_bz.get_bug.assert_called_with(bug_id) @patch("toddlers.plugins.scm_request_processor.bugzilla_system") def test_validate_review_bug_not_epel_branch_epel_bug(self, mock_bz): """ Assert that error will be raised when the requested branch isn't for EPEL, but the review bug is. """ # Preparation repo = "repo" branch = "f36" bug_id = "123" mock_bug = Mock() mock_bug.product = "Fedora EPEL" mock_bz.get_bug.return_value = mock_bug error = ( 'The Bugzilla bug is for "Fedora EPEL" but the ' 'requested branch is "{0}"' ).format(branch) # Method to test with pytest.raises(ValidationError, match=error): self.toddler.validate_review_bug(bug_id, repo, branch) # Asserts mock_bz.get_bug.assert_called_with(bug_id) @patch("toddlers.plugins.scm_request_processor.bugzilla_system") def test_validate_review_bug_invalid_component(self, mock_bz): """ Assert that error will be raised when the namespace provided is not known. """ # Preparation repo = "repo" branch = "f36" bug_id = "123" mock_bug = Mock() mock_bug.component = None mock_bz.get_bug.return_value = mock_bug self.toddler.pagure_namespace_to_component = {} error = "The Bugzilla bug provided is not the proper type" # Method to test with pytest.raises(ValidationError, match=error): self.toddler.validate_review_bug(bug_id, repo, branch) # Asserts mock_bz.get_bug.assert_called_with(bug_id) @patch("toddlers.plugins.scm_request_processor.bugzilla_system") def test_validate_review_bug_invalid_product(self, mock_bz): """ Assert that error will be raised when the namespace provided is not a valid product. """ # Preparation repo = "repo" namespace = "rpms" branch = "f36" bug_id = "123" mock_bug = Mock() mock_bug.component = namespace mock_bug.product = "invalid" mock_bz.get_bug.return_value = mock_bug self.toddler.pagure_namespace_to_component = {namespace: namespace} self.toddler.pagure_namespace_to_product = {namespace: [namespace, namespace]} error = 'The Bugzilla bug provided is not for "{0}" or "{0}"'.format(namespace) # Method to test with pytest.raises(ValidationError, match=error): self.toddler.validate_review_bug(bug_id, repo, branch) # Asserts mock_bz.get_bug.assert_called_with(bug_id) @patch("toddlers.plugins.scm_request_processor.bugzilla_system") def test_validate_review_bug_not_assigned(self, mock_bz): """ Assert that error will be raised when the bug is not assigned to anybody. """ # Preparation repo = "repo" namespace = "rpms" branch = "f36" bug_id = "123" mock_bug = Mock() mock_bug.component = namespace mock_bug.product = namespace mock_bug.assigned_to = None mock_bz.get_bug.return_value = mock_bug self.toddler.pagure_namespace_to_component = {namespace: namespace} self.toddler.pagure_namespace_to_product = {namespace: [namespace, namespace]} error = "The Bugzilla bug provided is not assigned to anyone" # Method to test with pytest.raises(ValidationError, match=error): self.toddler.validate_review_bug(bug_id, repo, branch) # Asserts mock_bz.get_bug.assert_called_with(bug_id) @patch("toddlers.plugins.scm_request_processor.fedora_account") @patch("toddlers.plugins.scm_request_processor.bugzilla_system") def test_validate_review_bug_user_not_in_FAS(self, mock_bz, mock_fas): """ Assert that error will be raised when bug creator doesn't exists in FAS. """ # Preparation repo = "repo" namespace = "rpms" branch = "f36" bug_id = "123" user = "zlopez" mock_bug = Mock() mock_bug.component = namespace mock_bug.product = namespace mock_bug.assigned_to = user mock_bug.creator = user mock_bz.get_bug.return_value = mock_bug mock_fas.get_user_by_email.return_value = None self.toddler.pagure_namespace_to_component = {namespace: namespace} self.toddler.pagure_namespace_to_product = {namespace: [namespace, namespace]} error = ( "The Bugzilla review bug creator could not be found in " "FAS. Make sure your FAS email address is the same as in " "Bugzilla." ) # Method to test with pytest.raises(ValidationError, match=error): self.toddler.validate_review_bug(bug_id, repo, branch, pagure_user=user) # Asserts mock_bz.get_bug.assert_called_with(bug_id) mock_fas.get_user_by_email.assert_called_with(user) @patch("toddlers.plugins.scm_request_processor.fedora_account") @patch("toddlers.plugins.scm_request_processor.bugzilla_system") def test_validate_review_bug_user_not_matching_requester(self, mock_bz, mock_fas): """ Assert that error will be raised when bug creator doesn't match the ticket requester. """ # Preparation repo = "repo" namespace = "rpms" branch = "f36" bug_id = "123" user = "zlopez" mock_bug = Mock() mock_bug.component = namespace mock_bug.product = namespace mock_bug.assigned_to = user mock_bug.creator = user mock_bz.get_bug.return_value = mock_bug mock_fas.get_user_by_email.return_value = {"username": "not_zlopez"} self.toddler.pagure_namespace_to_component = {namespace: namespace} self.toddler.pagure_namespace_to_product = {namespace: [namespace, namespace]} error = ( "The Bugzilla review bug creator " "didn't match the requester in Pagure." ) # Method to test with pytest.raises(ValidationError, match=error): self.toddler.validate_review_bug(bug_id, repo, branch, pagure_user=user) # Asserts mock_bz.get_bug.assert_called_with(bug_id) mock_fas.get_user_by_email.assert_called_with(user) @patch("toddlers.plugins.scm_request_processor.fedora_account") @patch("toddlers.plugins.scm_request_processor.bugzilla_system") def test_validate_review_bug_reviewer_not_in_FAS(self, mock_bz, mock_fas): """ Assert that error will be raised when bug reviewer isn't in FAS. """ # Preparation repo = "repo" namespace = "rpms" branch = "f36" bug_id = "123" user = "zlopez" mock_bug = Mock() mock_bug.component = namespace mock_bug.product = namespace mock_bug.assigned_to = user mock_bug.creator = user mock_bug.flags = [{"name": "fedora-review", "status": ""}] mock_bz.get_bug.return_value = mock_bug mock_fas.get_user_by_email.side_effect = [{"username": user}, None] self.toddler.pagure_namespace_to_component = {namespace: namespace} self.toddler.pagure_namespace_to_product = {namespace: [namespace, namespace]} error = ( 'The email address "{0}" of the Bugzilla reviewer ' "is not tied to a user in FAS or FAS check failed. " "Group membership can't be validated.".format(user) ) # Method to test with pytest.raises(ValidationError, match=error): self.toddler.validate_review_bug(bug_id, repo, branch, pagure_user=user) # Asserts mock_bz.get_bug.assert_called_with(bug_id) mock_fas.get_user_by_email.assert_has_calls([call(user), call(user)]) @patch("toddlers.plugins.scm_request_processor.fedora_account") @patch("toddlers.plugins.scm_request_processor.bugzilla_system") def test_validate_review_bug_reviewer_not_packager(self, mock_bz, mock_fas): """ Assert that error will be raised when bug reviewer is not in packager group. """ # Preparation repo = "repo" namespace = "rpms" branch = "f36" bug_id = "123" user = "zlopez" mock_bug = Mock() mock_bug.component = namespace mock_bug.product = namespace mock_bug.assigned_to = user mock_bug.creator = user mock_bug.flags = [{"name": "fedora-review", "status": ""}] mock_bz.get_bug.return_value = mock_bug mock_fas.get_user_by_email.return_value = {"username": user} mock_fas.user_member_of.return_value = False self.toddler.pagure_namespace_to_component = {namespace: namespace} self.toddler.pagure_namespace_to_product = {namespace: [namespace, namespace]} error = ( "The Bugzilla bug's review " 'is approved by a user "{0}" that is ' "not a packager or FAS check failed".format(user) ) # Method to test with pytest.raises(ValidationError, match=error): self.toddler.validate_review_bug(bug_id, repo, branch, pagure_user=user) # Asserts mock_bz.get_bug.assert_called_with(bug_id) mock_fas.get_user_by_email.assert_has_calls([call(user), call(user)]) mock_fas.user_member_of.assert_called_with({"username": user}, "packager") @patch("toddlers.plugins.scm_request_processor.fedora_account") @patch("toddlers.plugins.scm_request_processor.bugzilla_system") def test_validate_review_bug_submitter_not_packager(self, mock_bz, mock_fas): """ Assert that error will be raised when bug submitter is not in packager group. """ # Preparation repo = "repo" namespace = "rpms" branch = "f36" bug_id = "123" user = "zlopez" mock_bug = Mock() mock_bug.component = namespace mock_bug.product = namespace mock_bug.assigned_to = user mock_bug.creator = user mock_bug.flags = [{"name": "fedora-review", "status": ""}] mock_bz.get_bug.return_value = mock_bug mock_fas.get_user_by_email.return_value = {"username": user} mock_fas.user_member_of.side_effect = [True, False] self.toddler.pagure_namespace_to_component = {namespace: namespace} self.toddler.pagure_namespace_to_product = {namespace: [namespace, namespace]} error = 'The Bugzilla reporter "{0}"' "is not a packager".format(user) # Method to test with pytest.raises(ValidationError, match=error): self.toddler.validate_review_bug(bug_id, repo, branch, pagure_user=user) # Asserts mock_bz.get_bug.assert_called_with(bug_id) mock_fas.get_user_by_email.assert_has_calls([call(user), call(user)]) mock_fas.user_member_of.assert_has_calls( [ call({"username": user}, "packager"), call({"username": user}, "packager"), ] ) @patch("toddlers.plugins.scm_request_processor.fedora_account") @patch("toddlers.plugins.scm_request_processor.bugzilla_system") def test_validate_review_bug_approved_by_creator(self, mock_bz, mock_fas): """ Assert that error will be raised when bug is approved by creator. """ # Preparation repo = "repo" namespace = "rpms" branch = "f36" bug_id = "123" user = "zlopez" mock_bug = Mock() mock_bug.component = namespace mock_bug.product = namespace mock_bug.assigned_to = user mock_bug.creator = user mock_bug.flags = [{"name": "fedora-review", "status": "", "setter": user}] mock_bz.get_bug.return_value = mock_bug mock_fas.get_user_by_email.return_value = {"username": user} mock_fas.user_member_of.return_value = True self.toddler.pagure_namespace_to_component = {namespace: namespace} self.toddler.pagure_namespace_to_product = {namespace: [namespace, namespace]} error = ( "The Bugzilla bug's review is approved " "by the person creating the bug. This is " "not allowed." ) # Method to test with pytest.raises(ValidationError, match=error): self.toddler.validate_review_bug(bug_id, repo, branch, pagure_user=user) # Asserts mock_bz.get_bug.assert_called_with(bug_id) mock_fas.get_user_by_email.assert_has_calls([call(user), call(user)]) mock_fas.user_member_of.assert_has_calls( [ call({"username": user}, "packager"), call({"username": user}, "packager"), ] ) @patch("toddlers.plugins.scm_request_processor.fedora_account") @patch("toddlers.plugins.scm_request_processor.bugzilla_system") def test_validate_review_bug_not_approved_by_assignee(self, mock_bz, mock_fas): """ Assert that error will be raised when bug is not approved by assignee. """ # Preparation repo = "repo" namespace = "rpms" branch = "f36" bug_id = "123" user = "zlopez" mock_bug = Mock() mock_bug.component = namespace mock_bug.product = namespace mock_bug.assigned_to = user mock_bug.creator = user mock_bug.flags = [ {"name": "fedora-review", "status": "", "setter": "not@setter.com"} ] mock_bz.get_bug.return_value = mock_bug mock_fas.get_user_by_email.return_value = {"username": user, "emails": []} mock_fas.user_member_of.return_value = True self.toddler.pagure_namespace_to_component = {namespace: namespace} self.toddler.pagure_namespace_to_product = {namespace: [namespace, namespace]} error = "The review is not approved by " "the assignee of the Bugzilla " "bug" # Method to test with pytest.raises(ValidationError, match=error): self.toddler.validate_review_bug(bug_id, repo, branch, pagure_user=user) # Asserts mock_bz.get_bug.assert_called_with(bug_id) mock_fas.get_user_by_email.assert_has_calls([call(user), call(user)]) mock_fas.user_member_of.assert_has_calls( [ call({"username": user, "emails": []}, "packager"), call({"username": user, "emails": []}, "packager"), ] ) @patch("toddlers.plugins.scm_request_processor.fedora_account") @patch("toddlers.plugins.scm_request_processor.bugzilla_system") def test_validate_review_bug_approved_in_past(self, mock_bz, mock_fas): """ Assert that error will be raised when bug was approved more than 60 days ago. """ # Preparation repo = "repo" namespace = "rpms" branch = "f36" bug_id = "123" user = "zlopez" assignee = "assignee" mock_bug = Mock() mock_bug.component = namespace mock_bug.product = namespace mock_bug.assigned_to = assignee mock_bug.creator = user mock_bug.flags = [ { "name": "fedora-review", "status": "", "setter": assignee, "modification_date": xmlrpc.client.DateTime( arrow.utcnow().shift(days=-61).datetime ), } ] mock_bz.get_bug.return_value = mock_bug mock_fas.get_user_by_email.return_value = { "username": user, } mock_fas.user_member_of.return_value = True self.toddler.pagure_namespace_to_component = {namespace: namespace} self.toddler.pagure_namespace_to_product = {namespace: [namespace, namespace]} error = "The Bugzilla bug's review " "was approved over 60 days ago" # Method to test with pytest.raises(ValidationError, match=error): self.toddler.validate_review_bug(bug_id, repo, branch, pagure_user=user) # Asserts mock_bz.get_bug.assert_called_with(bug_id) mock_fas.get_user_by_email.assert_has_calls([call(user), call(assignee)]) mock_fas.user_member_of.assert_has_calls( [ call({"username": user}, "packager"), call({"username": user}, "packager"), ] ) @patch("toddlers.plugins.scm_request_processor.fedora_account") @patch("toddlers.plugins.scm_request_processor.bugzilla_system") def test_validate_review_bug_flags_missing(self, mock_bz, mock_fas): """ Assert that error will be raised when the bug flags are missing. """ # Preparation repo = "repo" namespace = "rpms" branch = "f36" bug_id = "123" user = "zlopez" mock_bug = Mock() mock_bug.component = namespace mock_bug.product = namespace mock_bug.creator = user mock_bug.flags = [] mock_bz.get_bug.return_value = mock_bug mock_fas.get_user_by_email.return_value = { "username": user, } self.toddler.pagure_namespace_to_component = {namespace: namespace} self.toddler.pagure_namespace_to_product = {namespace: [namespace, namespace]} error = "The Bugzilla bug is not approved yet" # Method to test with pytest.raises(ValidationError, match=error): self.toddler.validate_review_bug(bug_id, repo, branch, pagure_user=user) # Asserts mock_bz.get_bug.assert_called_with(bug_id) mock_fas.get_user_by_email.assert_has_calls([call(user)]) @pytest.mark.parametrize( "summary, error", [ ("", 'Invalid title for this Bugzilla bug (no ":" present)'), ("Review: package", 'Invalid title for this Bugzilla bug (no "-" present)'), ( "Review: package - 1.0", ( 'The package in the Bugzilla bug "package" doesn\'t match ' 'the one provided "repo"' ), ), ], ) @patch("toddlers.plugins.scm_request_processor.fedora_account") @patch("toddlers.plugins.scm_request_processor.bugzilla_system") def test_validate_review_bug_invalid_summary( self, mock_bz, mock_fas, summary, error ): """ Assert that error will be raised when bug summary is not correct. """ # Preparation repo = "repo" namespace = "rpms" branch = "f36" bug_id = "123" user = "zlopez" assignee = "assignee" mock_bug = Mock() mock_bug.component = namespace mock_bug.product = namespace mock_bug.assigned_to = assignee mock_bug.creator = user mock_bug.summary = summary mock_bug.flags = [ { "name": "fedora-review", "status": "+", "setter": assignee, "modification_date": xmlrpc.client.DateTime(arrow.utcnow().datetime), } ] mock_bz.get_bug.return_value = mock_bug mock_fas.get_user_by_email.return_value = { "username": user, } mock_fas.user_member_of.return_value = True self.toddler.pagure_namespace_to_component = {namespace: namespace} self.toddler.pagure_namespace_to_product = {namespace: [namespace, namespace]} # Method to test with pytest.raises(ValidationError, match=re.escape(error)): self.toddler.validate_review_bug(bug_id, repo, branch, pagure_user=user) # Asserts mock_bz.get_bug.assert_called_with(bug_id) mock_fas.get_user_by_email.assert_has_calls([call(user), call(assignee)]) mock_fas.user_member_of.assert_has_calls( [ call({"username": user}, "packager"), call({"username": user}, "packager"), ] ) @patch("toddlers.plugins.scm_request_processor.fedora_account") @patch("toddlers.plugins.scm_request_processor.bugzilla_system") def test_validate_review_bug_valid(self, mock_bz, mock_fas): """ Assert that no error will be raised when bug is valid. """ # Preparation repo = "repo" namespace = "rpms" branch = "f36" bug_id = "123" user = "zlopez" assignee = "assignee" mock_bug = Mock() mock_bug.component = namespace mock_bug.product = namespace mock_bug.assigned_to = assignee mock_bug.creator = user mock_bug.summary = "Review: repo - Project for everyday work" mock_bug.flags = [ { "name": "fedora-review", "status": "+", "setter": assignee, "modification_date": xmlrpc.client.DateTime(arrow.utcnow().datetime), }, { "name": "random-flag", }, ] mock_bz.get_bug.return_value = mock_bug mock_fas.get_user_by_email.return_value = { "username": user, } mock_fas.user_member_of.return_value = True self.toddler.pagure_namespace_to_component = {namespace: namespace} self.toddler.pagure_namespace_to_product = {namespace: [namespace, namespace]} # Method to test self.toddler.validate_review_bug(bug_id, repo, branch, pagure_user=user) # Asserts mock_bz.get_bug.assert_called_with(bug_id) mock_fas.get_user_by_email.assert_has_calls([call(user), call(assignee)]) mock_fas.user_member_of.assert_has_calls( [ call({"username": user}, "packager"), call({"username": user}, "packager"), ] ) class TestValidEpelPackage: """ Test class for `toddlers.plugins.scm_request_processor.SCMRequestProcessor.valid_epel_package` method. """ def setup(self): """ Initialize toddler. """ self.toddler = scm_request_processor.SCMRequestProcessor() self.toddler.pagure_io = Mock() self.toddler.dist_git = Mock() self.toddler.requests_session = Mock() def test_valid_epel_package_bad_response(self): """ Assert that method will raise error when bad response is retrieved. """ # Preparation name = "package" branch = "epel8" mock_response = Mock() mock_response.status_code = 500 self.toddler.requests_session.get.return_value = mock_response error = "Couldn't retrieve the list of official EL packages" # Method to test with pytest.raises(ValidationError, match=error): self.toddler.valid_epel_package(name=name, branch=branch) # Asserts self.toddler.requests_session.get.assert_called_with( "https://infrastructure.fedoraproject.org/repo/json/pkg_el8.json" ) def test_valid_epel_package_epel6_noarch(self): """ Assert that method will return False when epel6 noarch package is introduced. """ # Preparation name = "package" branch = "epel6" mock_response = Mock() mock_response.status_code = 200 mock_response.json.return_value = { "arches": ["noarch", "ppc", "i386"], "packages": {name: {"arch": ["noarch"]}}, } self.toddler.requests_session.get.return_value = mock_response # Method to test result = self.toddler.valid_epel_package(name=name, branch=branch) # Asserts self.toddler.requests_session.get.assert_called_with( "https://infrastructure.fedoraproject.org/repo/json/pkg_el6.json" ) assert result is False def test_valid_epel_package_epel7_noarch(self): """ Assert that method will return False when epel7 noarch package is introduced. """ # Preparation name = "package" branch = "epel7" mock_response = Mock() mock_response.status_code = 200 mock_response.json.return_value = { "arches": ["noarch", "ppc", "i386"], "packages": {name: {"arch": ["noarch"]}}, } self.toddler.requests_session.get.return_value = mock_response # Method to test result = self.toddler.valid_epel_package(name=name, branch=branch) # Asserts self.toddler.requests_session.get.assert_called_with( "https://infrastructure.fedoraproject.org/repo/json/pkg_el7.json" ) assert result is False def test_valid_epel_package_valid_package(self): """ Assert that method will return True when package is valid. """ # Preparation name = "package" branch = "epel8" mock_response = Mock() mock_response.status_code = 200 mock_response.json.return_value = { "arches": ["noarch", "ppc", "i386"], "packages": {name: {"arch": ["x86_64"]}}, } self.toddler.requests_session.get.return_value = mock_response # Method to test result = self.toddler.valid_epel_package(name=name, branch=branch) # Asserts self.toddler.requests_session.get.assert_called_with( "https://infrastructure.fedoraproject.org/repo/json/pkg_el8.json" ) assert result is True class TestMain: """ Test class for `toddlers.plugins.scm_request_processor.main` function. """ def test_main_no_args(self, capsys): """Assert that help is printed if no arg is provided.""" with pytest.raises(SystemExit): scm_request_processor.main([]) out, err = capsys.readouterr() assert out == "" # Expecting something along these lines, but don't make the test too tight: # # usage: pytest [-h] [--dry-run] [-q | --debug] conf [username] # pytest: error: the following arguments are required: conf assert err.startswith("usage:") assert "error: the following arguments are required:" in err @patch("toml.load") @patch("toddlers.plugins.scm_request_processor.SCMRequestProcessor.process") @patch("toddlers.plugins.scm_request_processor.pagure") def test_main(self, mock_pagure, mock_process, mock_toml): """Assert that main is initializing config and message correctly.""" # Preparation mock_toml.return_value = {} mock_pagure_io = Mock() mock_issue = {"id": 100} mock_pagure_io.get_issue.return_value = mock_issue mock_pagure.set_pagure.return_value = mock_pagure_io # Function to test scm_request_processor.main(["--config", "test.cfg", "100"]) # Assertions mock_toml.assert_called_with("test.cfg") mock_pagure.set_pagure.assert_called_with({}) mock_pagure_io.get_issue.assert_called_with( 100, scm_request_processor.PROJECT_NAMESPACE ) message = IssueNewV1() message.body["issue"] = mock_issue message.body["project"] = {"fullname": scm_request_processor.PROJECT_NAMESPACE} mock_process.assert_called_with(config={}, message=message)