toddlers/tests/plugins/test_scm_request_processor.py
Michal Konečný 92497811d2 Add unit tests for scm_request_processor module
Add unit tests for verify_slas method and continue adding tests for
create_new_repo method.

Signed-off-by: Michal Konečný <mkonecny@redhat.com>
2022-03-16 16:55:58 +01:00

768 lines
23 KiB
Python

"""
Unit tests for `toddlers.plugins.scm_request_processor`
"""
import json
from unittest.mock import call, patch, Mock
import logging
from pagure_messages.issue_schema import IssueNewV1
import pytest
import toddlers.plugins.scm_request_processor as scm_request_processor
from toddlers.exceptions import ValidationError
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",
[
"org.fedoraproject.*.pagure.issue.new",
"org.fedoraproject.*.pagure.issue.edit",
"org.fedoraproject.stg.pagure.issue.new",
"org.fedoraproject.stg.pagure.issue.edit",
"org.fedoraproject.prod.pagure.issue.new",
"org.fedoraproject.prod.pagure.issue.edit",
]
)
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,
"close_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."
)
@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,
"close_status": "Open"
}
msg.body = {
"project": {
"fullname": scm_request_processor.PROJECT_NAMESPACE
},
"issue": issue
}
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_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")
def test_create_new_repo_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"
bug_id = "123"
action = "new_repo"
sls = {
"rawhide": "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
}
mock_valid_epel_package.return_value = False
self.toddler.create_new_repo(issue, json)
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"
)
def test_create_new_repo_requester_not_in_dist_git(self):
"""
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 = True
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)
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,
)
def test_create_new_repo_project_exists(self):
"""
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 = True
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)
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"
)