cede6e9ab6
Currently process, process_ticket and accepts_topic are covered. Signed-off-by: Michal Konečný <mkonecny@redhat.com>
350 lines
11 KiB
Python
350 lines
11 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,
|
|
)
|