ariadne/tests/test_hermes_incident_issue.py

from __future__ import annotations

from types import SimpleNamespace

from ariadne.services import hermes_incident_body as body
from ariadne.services import hermes_incident_issue as module


JOB = "metis"
INCIDENT_ID = f"{JOB}/12"
CLASSIFICATION = "dependency_resolution_failure"
TOKEN = "super-secret-token"
PROPOSAL_URL = "https://scm.example/pulls/8"


class FakeResponse:
    def __init__(self, status_code: int, payload=None) -> None:  # type: ignore[no-untyped-def]
        self.status_code = status_code
        self._payload = payload

    def json(self):  # type: ignore[no-untyped-def]
        if isinstance(self._payload, Exception):
            raise self._payload
        return self._payload


class FakeStorage:
    def __init__(self, explode: bool = False) -> None:
        self.events: list[tuple[str, dict]] = []
        self.explode = explode

    def record_event(self, event_type, detail) -> None:  # type: ignore[no-untyped-def]
        if self.explode:
            raise RuntimeError("storage is down")
        self.events.append((event_type, detail))


def _install_http(monkeypatch, responses=None) -> dict:  # type: ignore[no-untyped-def]
    calls: dict = {"requests": [], "kwargs": None}
    queue = list(responses or [])

    class FakeClient:
        def __init__(self, **kwargs) -> None:  # type: ignore[no-untyped-def]
            calls["kwargs"] = kwargs

        def __enter__(self):  # type: ignore[no-untyped-def]
            return self

        def __exit__(self, *args) -> None:  # type: ignore[no-untyped-def]
            return None

        def _next(self, method, url, kwargs):  # type: ignore[no-untyped-def]
            calls["requests"].append((method, url, kwargs))
            item = queue.pop(0)
            if isinstance(item, Exception):
                raise item
            return item

        def get(self, url, **kwargs):  # type: ignore[no-untyped-def]
            return self._next("GET", url, kwargs)

        def post(self, url, **kwargs):  # type: ignore[no-untyped-def]
            return self._next("POST", url, kwargs)

    monkeypatch.setattr(module.httpx, "Client", FakeClient)
    return calls


def _cfg(**overrides) -> dict:  # type: ignore[no-untyped-def]
    values = {
        "gitea_base_url": "https://scm.example",
        "gitea_token": TOKEN,
        "owner": "bstein",
        "repo": JOB,
        "timeout_seconds": 7.5,
    }
    values.update(overrides)
    return values


def _context(**overrides) -> dict:  # type: ignore[no-untyped-def]
    values = {
        "incident_id": INCIDENT_ID,
        "job": JOB,
        "build_number": 12,
        "build_url": "https://ci.example/job/metis/12/",
        "classification": CLASSIFICATION,
        "confidence": 0.91,
        "first_failed_gate": "dependencies",
        "reason": "no allowlisted action fits this failure",
        "facts": [{"statement": "pip could not resolve urllib3", "source": "jenkins", "reference": "console"}],
        "inferences": ["an upstream index published a broken constraint"],
        "authorize_reason": "classification_not_actionable",
        "run_id": "run-9",
    }
    values.update(overrides)
    return values


def _issue(number: int, job: str = JOB, classification: str = CLASSIFICATION, incident: str = INCIDENT_ID) -> dict:
    return {
        "number": number,
        "html_url": f"https://scm.example/bstein/{job}/issues/{number}",
        "body": f"diagnosis\n\n{body.issue_marker(job, classification, incident)}",
    }


def _settings(**overrides) -> SimpleNamespace:  # type: ignore[no-untyped-def]
    values = {
        "hermes_issues_enabled": True,
        "hermes_issue_repos": {JOB: ("bstein", JOB)},
        "hermes_issue_dedupe_scope": "classification",
        "hermes_issue_max_per_tick": 2,
        "hermes_gitea_base_url": "https://scm.example",
        "hermes_gitea_token": TOKEN,
    }
    values.update(overrides)
    return SimpleNamespace(**values)


def _base(build_number: int = 12) -> dict:
    return {"incident_id": f"{JOB}/{build_number}", "job": JOB, "build_number": build_number}


def _diagnosis(**overrides) -> dict:  # type: ignore[no-untyped-def]
    decision = SimpleNamespace(
        classification=CLASSIFICATION,
        confidence=0.91,
        first_failed_gate="dependencies",
        reason="the failure has no known signature",
        facts=[SimpleNamespace(statement="pip failed", source="jenkins", reference="console")],
        inferences=["upstream index broke"],
    )
    values = {
        "bundle": {"jenkins": {"url": "https://ci.example/job/metis/12/"}},
        "outcome": SimpleNamespace(decision=decision),
        "authorize_reason": "classification_not_actionable",
        "run_id": "run-9",
    }
    values.update(overrides)
    return values


def test_marker_round_trips_through_the_body() -> None:
    marker = body.issue_marker(JOB, CLASSIFICATION, INCIDENT_ID)

    assert body.parse_issue_marker(f"text\n{marker}\nmore") == {
        "job": JOB,
        "classification": CLASSIFICATION,
        "incident": INCIDENT_ID,
    }


def test_marker_flattens_values_that_would_break_the_comment() -> None:
    marker = body.issue_marker("job\nname", "a --> b", INCIDENT_ID)

    assert "\n" not in marker
    assert body.parse_issue_marker(marker) == {
        "job": "job name",
        "classification": "a b",
        "incident": INCIDENT_ID,
    }


def test_marker_parsing_tolerates_missing_and_non_string_bodies() -> None:
    assert body.parse_issue_marker(None) is None
    assert body.parse_issue_marker(12) is None
    assert body.parse_issue_marker("a plain human issue") is None


def test_classification_scope_matches_a_different_incident_of_the_same_job(monkeypatch) -> None:
    calls = _install_http(monkeypatch, [FakeResponse(200, [_issue(7, incident=f"{JOB}/9")])])

    found = module.find_open_incident_issue(_cfg(), JOB, CLASSIFICATION, INCIDENT_ID)

    assert found == {
        "found": True,
        "issue_number": 7,
        "url": "https://scm.example/bstein/metis/issues/7",
        "error": None,
    }
    method, url, kwargs = calls["requests"][0]
    assert (method, url) == ("GET", "https://scm.example/api/v1/repos/bstein/metis/issues")
    assert kwargs["params"] == {"state": "open", "limit": 50}
    assert kwargs["headers"] == {"Authorization": f"token {TOKEN}"}


def test_classification_scope_ignores_other_jobs_and_classifications(monkeypatch) -> None:
    _install_http(
        monkeypatch,
        [FakeResponse(200, [_issue(7, job="lesavka"), _issue(8, classification="flaky_test")])],
    )

    found = module.find_open_incident_issue(_cfg(), JOB, CLASSIFICATION, INCIDENT_ID)

    assert found["found"] is False
    assert found["error"] is None


def test_incident_scope_matches_only_the_same_incident(monkeypatch) -> None:
    _install_http(
        monkeypatch,
        [FakeResponse(200, [_issue(7, incident=f"{JOB}/9"), _issue(9, incident=INCIDENT_ID)])],
    )

    found = module.find_open_incident_issue(
        _cfg(dedupe_scope="incident"), JOB, CLASSIFICATION, INCIDENT_ID
    )

    assert found["found"] is True
    assert found["issue_number"] == 9


def test_lookup_returns_the_lowest_numbered_match(monkeypatch) -> None:
    _install_http(monkeypatch, [FakeResponse(200, [_issue(31), _issue(12), _issue(20)])])

    found = module.find_open_incident_issue(_cfg(), JOB, CLASSIFICATION, INCIDENT_ID)

    assert found["issue_number"] == 12


def test_lookup_ignores_unmarked_and_malformed_issues(monkeypatch) -> None:
    payload = [
        {"number": 3, "body": "a human filed this"},
        {"number": True, "body": _issue(4)["body"]},
        {"number": "5", "body": _issue(5)["body"]},
        "not an issue",
    ]
    _install_http(monkeypatch, [FakeResponse(200, payload)])

    assert module.find_open_incident_issue(_cfg(), JOB, CLASSIFICATION, INCIDENT_ID)["found"] is False


def test_lookup_fails_open_on_http_parse_and_transport_errors(monkeypatch) -> None:
    _install_http(
        monkeypatch,
        [
            FakeResponse(503, None),
            FakeResponse(200, ValueError("no json")),
            FakeResponse(200, {"issues": []}),
            RuntimeError("connection reset"),
        ],
    )

    statuses = [
        module.find_open_incident_issue(_cfg(), JOB, CLASSIFICATION, INCIDENT_ID) for _ in range(4)
    ]

    assert [result["found"] for result in statuses] == [False, False, False, False]
    assert statuses[0]["error"] == "open issue lookup http 503"
    assert "parse failed" in statuses[1]["error"]
    assert statuses[2]["error"] == "open issue payload is not a list"
    assert "connection reset" in statuses[3]["error"]


def test_lookup_without_a_base_url_never_calls_gitea(monkeypatch) -> None:
    calls = _install_http(monkeypatch, [])

    found = module.find_open_incident_issue(_cfg(gitea_base_url=" "), JOB, CLASSIFICATION, INCIDENT_ID)

    assert found == {"found": False, "issue_number": None, "url": None, "error": "gitea base url is empty"}
    assert calls["requests"] == []


def test_create_posts_the_issue_and_returns_its_identity(monkeypatch) -> None:
    payload = {"number": 41, "html_url": "https://scm.example/bstein/metis/issues/41"}
    calls = _install_http(monkeypatch, [FakeResponse(201, payload)])

    created = module.create_incident_issue(_cfg(), _context())

    assert created == {"issue_number": 41, "url": payload["html_url"], "error": None}
    method, url, kwargs = calls["requests"][0]
    assert (method, url) == ("POST", "https://scm.example/api/v1/repos/bstein/metis/issues")
    assert kwargs["headers"] == {"Authorization": f"token {TOKEN}"}
    assert kwargs["json"]["title"] == f"[hermes] {JOB} #12: {CLASSIFICATION}"


def test_create_reports_http_errors_exceptions_and_missing_numbers(monkeypatch) -> None:
    _install_http(
        monkeypatch,
        [FakeResponse(422, None), RuntimeError("tls handshake failed"), FakeResponse(201, {"html_url": "u"})],
    )

    results = [module.create_incident_issue(_cfg(), _context()) for _ in range(3)]

    assert results[0] == {"issue_number": None, "url": None, "error": "issue create http 422"}
    assert "tls handshake failed" in results[1]["error"]
    assert results[2] == {"issue_number": None, "url": "u", "error": "issue create response had no number"}


def test_create_without_a_base_url_never_calls_gitea(monkeypatch) -> None:
    calls = _install_http(monkeypatch, [])

    created = module.create_incident_issue(_cfg(gitea_base_url=""), _context())

    assert created["error"] == "gitea base url is empty"
    assert calls["requests"] == []


def test_title_truncates_a_long_classification() -> None:
    title = body.issue_title(_context(classification="x" * 400))

    assert len(title) == 120
    assert title.startswith(f"[hermes] {JOB} #12: ")
    assert title.endswith("...")


def test_body_carries_the_marker_the_build_url_and_the_no_write_footer() -> None:
    rendered = body.issue_body(_context())

    assert rendered.rstrip().endswith(body.issue_marker(JOB, CLASSIFICATION, INCIDENT_ID))
    assert "https://ci.example/job/metis/12/" in rendered
    assert "Hermes has no write access to this repository" in rendered
    assert "no files or infrastructure were changed" in rendered
    assert "run `run-9`" in rendered
    assert "/api/admin/audit/events" in rendered
    assert "## Why a human is needed" in rendered
    assert "classification_not_actionable" in rendered
    assert "- **jenkins** — pip could not resolve urllib3 (`console`)" in rendered


def test_body_never_contains_the_gitea_token() -> None:
    rendered = body.issue_body(_context(reason=f"failure while using {TOKEN[:4]}"))

    assert TOKEN not in rendered


def test_body_caps_facts_inferences_and_statement_length() -> None:
    rendered = body.issue_body(
        _context(
            facts=[{"statement": "s" * 500, "source": "jenkins", "reference": f"r{i}"} for i in range(20)],
            inferences=[f"inference {i}" for i in range(12)],
        )
    )

    assert rendered.count("- **jenkins**") == 10
    assert rendered.count("- inference ") == 6
    assert "s" * 500 not in rendered
    assert "s" * 297 + "..." in rendered


def test_body_truncates_to_the_configured_cap_and_keeps_the_marker() -> None:
    rendered = body.issue_body(_context(reason="r" * 20000), max_chars=1200)

    assert len(rendered) <= 1200
    assert "Truncated by Ariadne" in rendered
    assert rendered.rstrip().endswith(body.issue_marker(JOB, CLASSIFICATION, INCIDENT_ID))


def test_a_proposed_fix_flows_from_the_diagnosis_into_the_issue_body() -> None:
    proposal = {"branch": "hermes-repair/12", "pr_number": 8, "url": PROPOSAL_URL}
    context = module.issue_context(_base(), _diagnosis(code_proposal=proposal))
    assert context["code_proposal_url"] == PROPOSAL_URL
    assert f"- Proposed fix awaiting review: {PROPOSAL_URL}" in body.issue_body(context)
    assert module.issue_context(_base(), _diagnosis())["code_proposal_url"] == ""
    assert module.issue_context(_base(), _diagnosis(code_proposal={}))["code_proposal_url"] == ""
    assert "Proposed fix awaiting review" not in body.issue_body(_context())


def test_context_defaults_to_undiagnosed_when_no_decision_was_parsed() -> None:
    context = module.issue_context(_base(), _diagnosis(outcome=None, run_id=None))

    assert context["classification"] == "undiagnosed"
    assert context["reason"] == "classification_not_actionable"
    assert context["facts"] == []
    assert context["build_url"] == "https://ci.example/job/metis/12/"


def test_maybe_file_issue_files_and_records_one_event(monkeypatch) -> None:
    payload = {"number": 5, "html_url": "https://scm.example/bstein/metis/issues/5"}
    calls = _install_http(monkeypatch, [FakeResponse(200, []), FakeResponse(201, payload)])
    storage, tick = FakeStorage(), {}

    detail = module.maybe_file_issue(storage, _settings(), _base(), _diagnosis(), tick)

    assert detail == {
        "incident_id": INCIDENT_ID,
        "job": JOB,
        "build_number": 12,
        "classification": CLASSIFICATION,
        "issue_number": 5,
        "url": payload["html_url"],
        "skipped": False,
        "error": None,
    }
    assert storage.events == [(module.ISSUE_EVENT_TYPE, detail)]
    assert tick == {"issues_filed": 1}
    assert [call[0] for call in calls["requests"]] == ["GET", "POST"]


def test_maybe_file_issue_skips_when_an_open_issue_already_exists(monkeypatch) -> None:
    calls = _install_http(monkeypatch, [FakeResponse(200, [_issue(7, incident=f"{JOB}/9")])])
    storage, tick = FakeStorage(), {}

    detail = module.maybe_file_issue(storage, _settings(), _base(), _diagnosis(), tick)

    assert detail["skipped"] is True
    assert detail["issue_number"] == 7
    assert detail["error"] is None
    assert [call[0] for call in calls["requests"]] == ["GET"]
    assert tick == {}


def test_disabled_flag_files_nothing_and_calls_nothing(monkeypatch) -> None:
    calls = _install_http(monkeypatch, [])
    storage = FakeStorage()

    assert module.maybe_file_issue(storage, _settings(hermes_issues_enabled=False), _base(), _diagnosis(), {}) is None
    assert calls["requests"] == []
    assert storage.events == []


def test_unmapped_job_files_nothing_and_calls_nothing(monkeypatch) -> None:
    calls = _install_http(monkeypatch, [])
    storage = FakeStorage()

    result = module.maybe_file_issue(
        storage, _settings(hermes_issue_repos={"other-job": ("bstein", "other")}), _base(), _diagnosis(), {}
    )

    assert result is None
    assert calls["requests"] == []
    assert storage.events == []


def test_max_per_tick_stops_a_burst_of_failures(monkeypatch) -> None:
    payload = {"number": 5, "html_url": "u"}
    calls = _install_http(
        monkeypatch,
        [FakeResponse(200, []), FakeResponse(201, payload), FakeResponse(200, []), FakeResponse(201, payload)],
    )
    storage, tick = FakeStorage(), {}
    config = _settings(hermes_issue_max_per_tick=2)

    results = [
        module.maybe_file_issue(storage, config, _base(build), _diagnosis(), tick) for build in (12, 13, 14)
    ]

    assert [result is None for result in results] == [False, False, True]
    assert tick == {"issues_filed": 2}
    assert len(calls["requests"]) == 4


def test_lookup_failure_fails_open_and_still_files(monkeypatch) -> None:
    calls = _install_http(monkeypatch, [FakeResponse(500, None), FakeResponse(201, {"number": 6})])
    storage, tick = FakeStorage(), {}

    detail = module.maybe_file_issue(storage, _settings(), _base(), _diagnosis(), tick)

    assert detail["skipped"] is False
    assert detail["issue_number"] == 6
    assert [call[0] for call in calls["requests"]] == ["GET", "POST"]


def test_create_failure_is_recorded_on_the_event(monkeypatch) -> None:
    _install_http(monkeypatch, [FakeResponse(200, []), FakeResponse(403, None)])
    storage, tick = FakeStorage(), {}

    detail = module.maybe_file_issue(storage, _settings(), _base(), _diagnosis(), tick)

    assert detail["issue_number"] is None
    assert detail["error"] == "issue create http 403"
    assert detail["skipped"] is False
    assert tick == {"issues_filed": 1}


def test_maybe_file_issue_never_raises(monkeypatch) -> None:
    _install_http(monkeypatch, [FakeResponse(200, []), FakeResponse(201, {"number": 5})])

    assert module.maybe_file_issue(FakeStorage(explode=True), _settings(), _base(), _diagnosis(), {}) is None


def test_repo_map_entries_accept_pairs_dicts_and_strings(monkeypatch) -> None:
    calls = _install_http(monkeypatch, [FakeResponse(200, []), FakeResponse(201, {"number": 5})] * 2)
    for repos in ({JOB: {"owner": "bstein", "repo": "metis"}}, {JOB: "bstein/metis"}):
        module.maybe_file_issue(
            FakeStorage(), _settings(hermes_issue_repos=repos), _base(), _diagnosis(), {}
        )

    assert [call[1] for call in calls["requests"]] == [
        "https://scm.example/api/v1/repos/bstein/metis/issues"
    ] * 4


def test_broken_repo_map_entries_file_nothing(monkeypatch) -> None:
    calls = _install_http(monkeypatch, [])

    for repos in ({JOB: "bstein"}, {JOB: ("bstein", "")}, {JOB: 12}, {JOB: None}):
        assert (
            module.maybe_file_issue(
                FakeStorage(), _settings(hermes_issue_repos=repos), _base(), _diagnosis(), {}
            )
            is None
        )
    assert calls["requests"] == []