toddlers/tests/plugins/test_distgit_commit_processor.py

from contextlib import nullcontext
import datetime as dt
import random
from typing import ContextManager, Optional
from unittest import mock

from fedora_messaging.message import Message
from fedora_messaging_git_hook_messages import CommitV1
import pytest

from toddlers.exceptions import ConfigError
from toddlers.plugins import distgit_commit_processor


class TestDistGitCommitProcessor:
    """Test the DistGitCommitProcessor toddler plugin."""

    toddler_cls = distgit_commit_processor.DistGitCommitProcessor

    @pytest.mark.parametrize(
        "topic, valid",
        (
            pytest.param("org.fedoraproject.prod.git.receive", True, id="prod-valid"),
            pytest.param("org.fedoraproject.stg.git.receive", True, id="stg-valid"),
            pytest.param(
                "org.fedoraproject.prod.pagure.git.receive",
                False,
                id="pagure-prod-invalid",
            ),
            pytest.param(
                "org.fedoraproject.stg.pagure.git.receive",
                False,
                id="pagure-stg-invalid",
            ),
            pytest.param(
                "pagure.io.prod.pagure.git.receive", False, id="pagure-io-invalid"
            ),
        ),
    )
    def test_accepts_topic(
        self,
        topic: str,
        valid: bool,
        toddler: distgit_commit_processor.DistGitCommitProcessor,
    ) -> None:
        assert toddler.accepts_topic(topic) is valid

    @pytest.mark.parametrize(
        "namespace, repo, path, result",
        (
            ("rpms", "kernel", "/some/root/rpms/kernel.git/", False),
            ("rpms", "kernel", "/some/root/rpms/kernel/", False),
            ("frobozzniks", "blomp", "/some/root/rpms/kernel.git/", False),
            ("rpms", "kernel", "/some/root/fork/foo/rpms/kernel.git/", True),
            ("rpms", "kernel", "/some/root/fork/foo/rpms/kernel", True),
            (None, "project", "/some/root/project.git/", False),
        ),
    )
    def test_ignore_commit(
        self,
        namespace: Optional[str],
        repo: str,
        path: str,
        result: bool,
        toddler: distgit_commit_processor.DistGitCommitProcessor,
        caplog: pytest.LogCaptureFixture,
    ):
        if namespace:
            repo_with_ns = namespace + "/" + repo
        else:
            repo_with_ns = repo

        commit = {
            "namespace": namespace,
            "repo": repo,
            "path": path,
        }
        body = {"id": "BOO", "agent": "m0rk", "commit": commit}

        message = CommitV1(body=body)

        assert toddler.ignore_commit(message) is result

        if repo_with_ns in path:
            assert f"Message {message.id} mismatch" not in caplog.text
        else:
            assert f"Message {message.id} mismatch" in caplog.text

    @pytest.mark.parametrize(
        "testcase",
        (
            "success",
            "success-loglevel-info",
            "success-default-subject",
            "success-default-content",
            "failure-wrong-msg-type",
            "failure-wrong-msg-type-loglevel-info",
            "failure-fork-commit",
            "failure-config-error",
        ),
    )
    def test_process(
        self,
        testcase: str,
        toddler: distgit_commit_processor.DistGitCommitProcessor,
        caplog: pytest.LogCaptureFixture,
    ) -> None:
        success = "success" in testcase
        wrong_msg_type = "wrong-msg-type" in testcase
        fork_commit = "fork-commit" in testcase
        config_error = "config-error" in testcase
        loglevel_info = "loglevel-info" in testcase
        default_subject = "default-subject" in testcase
        default_content = "default-content" in testcase

        # Appease mypy
        exception_ctx: ContextManager

        now = dt.datetime.now(tz=dt.timezone.utc).isoformat()

        commit = {
            "namespace": "rpms",
            "repo": "kernel",
            "path": "/some/root/rpms/kernel.git/",
            "branch": "rawhide",
            "rev": "deadbeef",
            "name": "Mork",
            "email": "mork@ork.org",
            "date": now,
            "summary": "Did the thing",
            "message": "Did the thing\n\nAmazing.",
            "patch": "No, I won’t fake a diff here.",
        }
        body = {"agent": "m0rk", "commit": commit}

        if fork_commit:
            commit["path"] = "/some/root/fork/foo/rpms/kernel.git/"

        msg = CommitV1(body=body)
        if wrong_msg_type:
            msg = Message(body=body)

        # Config items which must be set
        config = {
            "mail_server": "bastion.fedoraproject.org",
            "mail_from": "notifications@fedoraproject.org",
            "mail_to": "scm-commits@lists.fedoraproject.org",
        }
        if config_error:
            # Nuke a random configuration item
            del config[list(config)[random.randint(0, len(config) - 1)]]
            exception_ctx = pytest.raises(ConfigError)
        else:
            exception_ctx = nullcontext()

        if not default_subject:
            config["mail_subject_tmpl"] = "SUBJECT-IS-SET: {message.summary}"

        if not default_content:
            config["mail_content_tmpl"] = "CONTENT-IS-SET\n{message}"

        with caplog.at_level(
            "INFO" if loglevel_info else "DEBUG"
        ), exception_ctx, mock.patch.object(
            distgit_commit_processor, "send_email"
        ) as send_email:
            toddler.process(config, msg)

        if not loglevel_info:
            assert "Processing message:" in caplog.text

        if success:
            send_email.assert_called_with(
                to_addresses=[config["mail_to"]],
                from_address=config["mail_from"],
                subject=mock.ANY,
                content=mock.ANY,
                mail_server=config["mail_server"],
            )

            subject = send_email.call_args.kwargs["subject"]
            assert (
                f"{body['agent']} pushed to {commit['namespace']}/{commit['repo']}"
                in subject
            )
            assert f"({commit['branch']})" in subject
            assert f"\"{commit['summary']}\"" in subject
            if default_subject:
                assert "SUBJECT-IS-SET" not in subject
            else:
                assert "SUBJECT-IS-SET" in subject

            content = send_email.call_args.kwargs["content"]
            assert f"From {commit['rev']}" in content
            assert f"From: {commit['name']} <{commit['email']}>" in content
            assert f"Date: {now}" in content
            assert f"Subject: {commit['summary']}" in content
            assert commit["message"] in content
            if default_content:
                assert "CONTENT-IS-SET" not in content
            else:
                assert "CONTENT-IS-SET" in content
        else:
            send_email.assert_not_called()

        if wrong_msg_type and not loglevel_info:
            assert "Skipping message" in caplog.text
        else:
            assert "Skipping message" not in caplog.text

        if fork_commit:
            assert "Ignoring message" in caplog.text
        else:
            assert "Ignoring message" not in caplog.text