toddlers/tests/plugins/test_flag_ci_pr.py

import logging
from unittest.mock import MagicMock, patch

import fedora_messaging.api
import pytest

from toddlers.plugins.flag_ci_pr import FlagCIPR


class TestFlagCIPRToddler:

    toddler_cls = FlagCIPR

    def test_accepts_topic_invalid(self, toddler):
        assert toddler.accepts_topic("foo.bar") is False

    @pytest.mark.parametrize(
        "topic",
        [
            "org.centos.#.ci.dist-git-pr.test.error",
            "org.centos.prod.ci.dist-git-pr.test.error",
            "org.centos.stage.ci.dist-git-pr.test.error",
            "org.centos.#.ci.dist-git-pr.test.complete"
            "org.centos.prod.ci.dist-git-pr.test.complete"
            "org.centos.stg.ci.dist-git-pr.test.complete"
            "org.centos.#.ci.dist-git-pr.test.running",
            "org.centos.prod.ci.dist-git-pr.test.running",
            "org.centos.stg.ci.dist-git-pr.test.running",
        ],
    )
    def test_accepts_topic_valid(self, toddler, topic):
        assert toddler.accepts_topic(topic)

    def test_process_invalid(self, toddler):
        msg = fedora_messaging.api.Message()
        msg.id = 123
        msg.topic = "toddlers.test.topic"
        assert toddler.process(config={}, message=msg) is None

    def test_process_invalid_status(self, toddler, caplog):
        caplog.set_level(logging.INFO)
        msg = fedora_messaging.api.Message()
        msg.id = 123
        msg.topic = "org.centos.stg.ci.dist-git-pr.test.invalid"
        assert toddler.process(config={}, message=msg) is None
        assert (
            caplog.records[-1].message
            == "Pipeline state is not 'complete' or 'running' or 'error'."
        )

    def test_process_no_version(self, toddler, caplog):
        caplog.set_level(logging.INFO)
        msg = fedora_messaging.api.Message()
        msg.id = 123
        msg.topic = "org.centos.stg.ci.dist-git-pr.test.complete"
        assert toddler.process(config={}, message=msg) is None
        assert caplog.records[-1].message == "Unsupported msg version, ignoring"

    def test_process_invalid_complete_status(self, toddler, caplog):
        caplog.set_level(logging.INFO)
        msg = fedora_messaging.api.Message()
        msg.id = 123
        msg.topic = "org.centos.stg.ci.dist-git-pr.test.complete"
        msg.body = {"version": "0.2.1", "test": {"result": "invalid"}}
        assert toddler.process(config={}, message=msg) is None
        assert (
            caplog.records[-1].message
            == "Build is not in one of the expected status, ignoring"
        )

    def test_process_no_artifact(self, toddler, caplog):
        caplog.set_level(logging.INFO)
        msg = fedora_messaging.api.Message()
        msg.id = 123
        msg.topic = "org.centos.stg.ci.dist-git-pr.test.complete"
        msg.body = {
            "version": "0.2.1",
            "test": {"result": "passed"},
            "artifact": {"commit_hash": "abcdefghijklmnopqrst"},
        }
        assert toddler.process(config={}, message=msg) is None
        assert (
            caplog.records[-1].message
            == "Invalid message: {'commit_hash': 'abcdefghijklmnopqrst'}, could not extract "
            "the PR id from it"
        )

    @patch("toddlers.plugins.flag_ci_pr.requests_session")
    def test_process_request_failed(self, mock_requests, toddler, caplog):
        mock_requests.request.return_value = MagicMock(
            ok=False, status_code=401, text="invalid"
        )
        caplog.set_level(logging.INFO)
        msg = fedora_messaging.api.Message()
        msg.id = 123
        msg.topic = "org.centos.stg.ci.dist-git-pr.test.running"
        msg.body = {
            "version": "0.2.1",
            "test": {"result": "passed"},
            "artifact": {
                "id": 456,
                "commit_hash": "abcdefghijklmnopqrst",
                "repository": "namespace/name",
            },
            "run": {"url": "https://example.com/testing"},
        }
        config = {
            "pagure_token_seed": "example_seed",
            "pagure_url": "https://pagure.io",
            "pagure_token": "ahah",
        }

        assert toddler.process(config=config, message=msg) == 1
        assert (
            caplog.records[-1].message == "Request to https://pagure.io returned: 401"
        )

    @patch("toddlers.plugins.flag_ci_pr.requests_session")
    def test_process_valid(self, mock_requests, toddler, caplog):
        mock_requests.request.return_value = MagicMock(
            ok=True, status_code=200, text="invalid"
        )
        caplog.set_level(logging.INFO)
        msg = fedora_messaging.api.Message()
        msg.id = 123
        msg.topic = "org.centos.stg.ci.dist-git-pr.test.error"
        msg.body = {
            "version": "0.2.1",
            "test": {"result": "passed"},
            "artifact": {
                "id": 456,
                "commit_hash": "abcdefghijklmnopqrst",
                "repository": "https://host.c/namespace/name",
            },
            "run": {"url": "https://example.com/testing"},
        }
        config = {
            "pagure_token_seed": "example_seed",
            "pagure_url": "https://pagure.io",
            "pagure_token": "ahah",
        }

        assert toddler.process(config=config, message=msg) is None
        assert (
            caplog.records[-3].message == "Request to https://pagure.io returned: 200"
        )
        assert caplog.records[-2].message == "All clear"
        assert (
            caplog.records[-1].message
            == "User-URL: https://host.c/namespace/name/pull-request/456"
        )