ariadne/tests/test_jenkins_build_weather.py

from __future__ import annotations

from datetime import datetime, timezone
import types

import httpx
import pytest
from prometheus_client import REGISTRY

from ariadne.services import jenkins_build_weather as weather_module


class _DummyResponse:
    def __init__(self, payload: dict[str, object], status_code: int = 200) -> None:
        self._payload = payload
        self.status_code = status_code

    def raise_for_status(self) -> None:
        if self.status_code >= 400:
            request = httpx.Request("GET", "https://ci.bstein.dev/api/json")
            response = httpx.Response(self.status_code, request=request)
            raise httpx.HTTPStatusError("boom", request=request, response=response)

    def json(self) -> dict[str, object]:
        return self._payload


class _DummyClient:
    def __init__(self, payload: dict[str, object]) -> None:
        self._payload = payload
        self.called = False

    def __enter__(self) -> _DummyClient:
        return self

    def __exit__(self, exc_type, exc, tb) -> bool:
        return False

    def get(self, url: str, params: dict[str, str] | None = None) -> _DummyResponse:
        self.called = True
        assert url == "https://ci.bstein.dev/api/json"
        assert isinstance(params, dict)
        assert "tree" in params
        return _DummyResponse(self._payload)


def _metric_value(name: str, labels: dict[str, str] | None = None) -> float | None:
    value = REGISTRY.get_sample_value(name, labels or {})
    return float(value) if value is not None else None


def _dummy_settings(base_url: str = "https://ci.bstein.dev") -> types.SimpleNamespace:
    return types.SimpleNamespace(
        jenkins_base_url=base_url,
        jenkins_api_user="",
        jenkins_api_token="",
        jenkins_api_timeout_sec=5.0,
    )


def test_collect_jenkins_build_weather_records_metrics(monkeypatch) -> None:
    weather_module._JOB_SERIES = set()
    monkeypatch.setattr(weather_module, "settings", _dummy_settings())

    payload = {
        "jobs": [
            {
                "name": "ariadne",
                "url": "https://ci.bstein.dev/job/ariadne/",
                "color": "blue",
                "healthReport": [{"score": 93}],
                "lastBuild": {"result": "SUCCESS", "timestamp": 1713000000000, "duration": 186000},
                "lastSuccessfulBuild": {"timestamp": 1713000000000},
                "lastFailedBuild": {"timestamp": 1712000000000},
            },
            {
                "name": "titan-iac",
                "url": "https://ci.bstein.dev/job/titan-iac/",
                "color": "red",
                "healthReport": [{"score": 11}],
                "lastBuild": {"result": "FAILURE", "timestamp": 1712990000000, "duration": 126000},
                "lastSuccessfulBuild": {"timestamp": 1711000000000},
                "lastFailedBuild": {"timestamp": 1712990000000},
            },
        ]
    }

    monkeypatch.setattr(weather_module.httpx, "Client", lambda **_kwargs: _DummyClient(payload))

    before = _metric_value("ariadne_jenkins_build_weather_runs_total", {"status": "ok"}) or 0.0
    summary = weather_module.collect_jenkins_build_weather()

    assert summary.jobs_total == 2
    assert summary.success_total == 1
    assert summary.failure_total == 1
    assert summary.running_total == 0
    assert summary.unknown_total == 0

    assert (_metric_value("ariadne_jenkins_build_weather_runs_total", {"status": "ok"}) or 0.0) == before + 1
    assert _metric_value(
        "ariadne_jenkins_build_weather_job_last_status",
        {
            "job": "ariadne",
            "job_url": "https://ci.bstein.dev/job/ariadne/",
            "weather_icon": "☀️",
        },
    ) == 1.0
    assert _metric_value(
        "ariadne_jenkins_build_weather_job_last_status",
        {
            "job": "titan-iac",
            "job_url": "https://ci.bstein.dev/job/titan-iac/",
            "weather_icon": "⛈️",
        },
    ) == 0.0
    assert _metric_value(
        "ariadne_jenkins_build_weather_job_last_duration_seconds",
        {
            "job": "ariadne",
            "job_url": "https://ci.bstein.dev/job/ariadne/",
            "weather_icon": "☀️",
        },
    ) == 186.0


def test_collect_jenkins_build_weather_removes_deleted_job_series(monkeypatch) -> None:
    weather_module._JOB_SERIES = set()
    monkeypatch.setattr(weather_module, "settings", _dummy_settings())

    first_payload = {
        "jobs": [
            {
                "name": "ariadne",
                "url": "https://ci.bstein.dev/job/ariadne/",
                "color": "blue",
                "healthReport": [{"score": 90}],
                "lastBuild": {"result": "SUCCESS", "timestamp": 1713000000000, "duration": 186000},
                "lastSuccessfulBuild": {"timestamp": 1713000000000},
                "lastFailedBuild": {"timestamp": 1712000000000},
            },
            {
                "name": "pegasus",
                "url": "https://ci.bstein.dev/job/pegasus/",
                "color": "yellow",
                "healthReport": [{"score": 50}],
                "lastBuild": {"result": "FAILURE", "timestamp": 1712980000000, "duration": 120000},
                "lastSuccessfulBuild": {"timestamp": 1710000000000},
                "lastFailedBuild": {"timestamp": 1712980000000},
            },
        ]
    }
    second_payload = {
        "jobs": [
            {
                "name": "ariadne",
                "url": "https://ci.bstein.dev/job/ariadne/",
                "color": "blue",
                "healthReport": [{"score": 90}],
                "lastBuild": {"result": "SUCCESS", "timestamp": 1713010000000, "duration": 184000},
                "lastSuccessfulBuild": {"timestamp": 1713010000000},
                "lastFailedBuild": {"timestamp": 1712000000000},
            }
        ]
    }
    payloads = [first_payload, second_payload]

    monkeypatch.setattr(
        weather_module.httpx,
        "Client",
        lambda **_kwargs: _DummyClient(payloads.pop(0)),
    )

    weather_module.collect_jenkins_build_weather()
    weather_module.collect_jenkins_build_weather()

    assert _metric_value(
        "ariadne_jenkins_build_weather_job_last_status",
        {
            "job": "pegasus",
            "job_url": "https://ci.bstein.dev/job/pegasus/",
            "weather_icon": "☁️",
        },
    ) is None


def test_collect_jenkins_build_weather_skips_when_base_url_empty(monkeypatch) -> None:
    weather_module._JOB_SERIES = set()
    monkeypatch.setattr(weather_module, "settings", _dummy_settings(base_url=""))

    before = _metric_value("ariadne_jenkins_build_weather_runs_total", {"status": "skipped"}) or 0.0
    summary = weather_module.collect_jenkins_build_weather()

    assert summary.jobs_total == 0
    assert (_metric_value("ariadne_jenkins_build_weather_runs_total", {"status": "skipped"}) or 0.0) == before + 1


def test_fetch_jobs_flattens_folder_jobs(monkeypatch) -> None:
    weather_module._JOB_SERIES = set()
    monkeypatch.setattr(weather_module, "settings", _dummy_settings())

    payload = {
        "jobs": [
            {
                "name": "folder",
                "url": "https://ci.bstein.dev/job/folder/",
                "jobs": [
                    {
                        "name": "child",
                        "url": "https://ci.bstein.dev/job/folder/job/child/",
                        "color": "blue",
                        "healthReport": [{"score": 100}],
                        "lastBuild": {"result": "SUCCESS", "timestamp": 1713000000000, "duration": 1000},
                        "lastSuccessfulBuild": {"timestamp": 1713000000000},
                        "lastFailedBuild": {"timestamp": 1712000000000},
                    }
                ],
            }
        ]
    }

    monkeypatch.setattr(weather_module.httpx, "Client", lambda **_kwargs: _DummyClient(payload))

    jobs = weather_module._fetch_jobs()
    assert len(jobs) == 1
    assert jobs[0].job == "folder/child"
    assert jobs[0].status == "success"
    assert jobs[0].last_duration_seconds == 1.0
    assert datetime.fromtimestamp(jobs[0].last_run_ts, tz=timezone.utc).year == 2024


def test_weather_helper_edges(monkeypatch) -> None:
    assert weather_module._metric_number(True) == 0.0
    assert weather_module._metric_number(object()) == 0.0
    assert weather_module._millis_to_seconds(0) == 0.0

    monkeypatch.setattr(
        weather_module,
        "settings",
        types.SimpleNamespace(jenkins_api_user=" user ", jenkins_api_token=" token "),
    )
    assert weather_module._jenkins_auth() == ("user", "token")

    assert weather_module._jenkins_status({"color": "blue_anime"}) == "running"
    assert weather_module._jenkins_status({"color": "green"}) == "success"
    assert weather_module._jenkins_status({"color": "yellow"}) == "failure"
    assert weather_module._jenkins_status({}) == "unknown"

    assert weather_module._health_score({"healthReport": ["bad"]}, "success") == 100.0
    assert weather_module._health_score({}, "running") == 60.0
    assert weather_module._health_score({}, "failure") == 10.0
    assert weather_module._health_score({}, "unknown") == -1.0

    assert weather_module._weather_icon(-1) == "❔"
    assert weather_module._weather_icon(60) == "⛅"
    assert weather_module._weather_icon(20) == "🌧️"


def test_flatten_parse_and_fetch_edges(monkeypatch) -> None:
    flattened = weather_module._flatten_jobs(
        [
            "bad",
            {"name": ""},
            {"name": "folder", "jobs": [{"name": "child", "url": "https://ci/job/child/", "lastBuild": {"result": "SUCCESS"}}]},
            {"name": "folder-without-build", "jobs": []},
        ]
    )
    assert [job["name"] for job in flattened] == ["folder/child", "folder-without-build"]
    assert weather_module._parse_job({"name": "missing-url"}) is None

    monkeypatch.setattr(weather_module, "settings", _dummy_settings(base_url=""))
    assert weather_module._fetch_jobs() == []

    captured = {}

    class CapturingClient(_DummyClient):
        def __init__(self, **kwargs):
            captured.update(kwargs)
            super().__init__({"jobs": [{"name": "bad"}]})

    monkeypatch.setattr(
        weather_module,
        "settings",
        types.SimpleNamespace(
            jenkins_base_url="https://ci.bstein.dev/",
            jenkins_api_user="user",
            jenkins_api_token="token",
            jenkins_api_timeout_sec=7.0,
        ),
    )
    monkeypatch.setattr(weather_module.httpx, "Client", CapturingClient)
    assert weather_module._fetch_jobs() == []
    assert captured["auth"] == ("user", "token")
    assert captured["timeout"] == 7.0

    class NonObjectClient(_DummyClient):
        def __init__(self, **kwargs):
            super().__init__(["bad"])

    monkeypatch.setattr(weather_module.httpx, "Client", NonObjectClient)
    with pytest.raises(ValueError, match="non-object"):
        weather_module._fetch_jobs()


def test_remove_missing_series_ignores_missing_metric_labels(monkeypatch) -> None:
    class MissingMetric:
        def remove(self, *labels):
            raise KeyError(labels)

    weather_module._JOB_SERIES = {("old", "https://ci/job/old/", "☀️")}
    monkeypatch.setattr(weather_module, "_JOB_METRICS", (MissingMetric(),))

    weather_module._remove_missing_series(set())

    assert weather_module._JOB_SERIES == set()


def test_collect_jenkins_build_weather_records_error(monkeypatch) -> None:
    monkeypatch.setattr(weather_module, "settings", _dummy_settings())
    before = _metric_value("ariadne_jenkins_build_weather_runs_total", {"status": "error"}) or 0.0
    monkeypatch.setattr(weather_module, "_fetch_jobs", lambda: (_ for _ in ()).throw(RuntimeError("jenkins down")))

    with pytest.raises(RuntimeError, match="jenkins down"):
        weather_module.collect_jenkins_build_weather()

    assert (_metric_value("ariadne_jenkins_build_weather_runs_total", {"status": "error"}) or 0.0) == before + 1