ariadne/tests/test_hermes_sonar_sweep.py

"""Tests for the scheduled SonarQube quality sweep."""

from __future__ import annotations

from types import SimpleNamespace

import pytest

from ariadne.services import hermes_sonar_sweep as module


REPO_CFG = {
    "owner": "bstein",
    "repo": "ariadne",
    "allowed_path_prefixes": ["ariadne/"],
    "allowed_suffixes": [".py"],
}


def _issue(**overrides):
    issue = {
        "key": "AZ-1",
        "path": "ariadne/services/hermes_code_defects.py",
        "line": 160,
        "rule": "python:S1172",
        "severity": "MAJOR",
        "type": "CODE_SMELL",
        "effort": "5min",
        "message": "Remove the unused function parameter.",
    }
    issue.update(overrides)
    return issue


def _config(**overrides):
    values = {
        "hermes_sonar_projects": {"ariadne": "ariadne"},
        "hermes_sonar_url": "http://sonar:9000",
        "hermes_sonar_token": "t",
        "hermes_sonar_types": ["CODE_SMELL"],
        "hermes_sonar_severities": [],
        "hermes_sonar_max_per_sweep": 1,
        "hermes_sonar_max_effort_minutes": 20,
        "hermes_sonar_timeout_seconds": 5,
    }
    values.update(overrides)
    return SimpleNamespace(**values)


class _Storage:
    def __init__(self):
        self.events = []

    def record_event(self, event_type, detail):
        self.events.append((event_type, detail))


@pytest.fixture
def wiring(monkeypatch):
    """Stub the repo resolution, the client and the proposal flow."""

    calls = {"proposals": [], "issues": [_issue()], "error": None, "result": {"status": "pr_opened"}}
    monkeypatch.setattr(module.hermes_code_flow, "code_config", lambda config: {})
    monkeypatch.setattr(module.hermes_code_flow, "resolve_repo_config", lambda job, cfg: REPO_CFG)
    monkeypatch.setattr(
        module.hermes_sonar_client, "fetch_issues",
        lambda cfg, project: (calls["issues"], calls["error"]),
    )

    def _propose(storage, **kwargs):
        calls["proposals"].append(kwargs)
        return calls["result"]

    monkeypatch.setattr(module.hermes_code_flow, "propose_code_fix", _propose)
    return calls


@pytest.mark.parametrize(
    ("effort", "expected"),
    [
        ("5min", 5),
        ("1h", 60),
        ("1h30min", 90),
        ("2d", 960),
        ("", None),
        ("soon", None),
        ("30", None),
        (None, None),
    ],
)
def test_the_effort_estimate_is_parsed_into_minutes(effort, expected) -> None:
    assert module.effort_minutes(effort) == expected


def test_the_cheapest_finding_is_chosen_first() -> None:
    """Effort is the closest proxy for a change the validator can check."""

    issues = [
        _issue(key="a", effort="15min"),
        _issue(key="b", effort="5min"),
        _issue(key="c", effort="10min"),
    ]

    assert module.select_issue(issues, REPO_CFG, _config())["key"] == "b"


def test_severity_breaks_a_tie_on_effort() -> None:
    issues = [_issue(key="a", severity="MINOR"), _issue(key="b", severity="CRITICAL")]

    assert module.select_issue(issues, REPO_CFG, _config())["key"] == "b"


def test_the_key_breaks_a_full_tie_so_the_pick_is_stable() -> None:
    """A sweep that proposes nothing today must retry the same finding."""

    issues = [_issue(key="zz"), _issue(key="aa")]

    assert module.select_issue(issues, REPO_CFG, _config())["key"] == "aa"
    assert module.select_issue(list(reversed(issues)), REPO_CFG, _config())["key"] == "aa"


def test_an_expensive_finding_is_left_to_a_person() -> None:
    issues = [_issue(effort="4h")]

    assert module.select_issue(issues, REPO_CFG, _config()) is None


def test_an_unparseable_effort_is_not_treated_as_free() -> None:
    """An unreadable estimate is not evidence that the work is small."""

    assert module.select_issue([_issue(effort="dunno")], REPO_CFG, _config()) is None


@pytest.mark.parametrize(
    "path",
    ["docs/readme.md", "ariadne/thing.txt", "scripts/tool.py", ""],
)
def test_a_finding_outside_the_write_allowlist_is_never_chosen(path) -> None:
    assert module.select_issue([_issue(path=path)], REPO_CFG, _config()) is None


def test_a_sweep_opens_one_proposal_for_the_chosen_finding(wiring) -> None:
    storage = _Storage()

    result = module.sweep(storage, _config(), {})

    assert result == {"proposed": 1, "skipped": [], "projects": 1}
    proposal = wiring["proposals"][0]
    assert proposal["incident_id"] == "sonar/ariadne/AZ-1"
    assert proposal["job"] == "ariadne"
    assert proposal["build_number"] == "sonar-AZ-1"


def test_the_bundle_offers_only_the_file_the_finding_names() -> None:
    """The finding states the file; ranking console text alongside adds noise."""

    bundle = module.bundle_for("ariadne", _issue())

    assert bundle["sonarqube"] == {"project": "ariadne", "issues": [_issue()]}
    assert bundle["jenkins"]["console_failures"] == []
    assert bundle["jenkins"]["console_tail"] == ""
    assert bundle["incident_id"] == "sonar/ariadne/AZ-1"


def test_the_sweep_budget_stops_after_its_quota(wiring) -> None:
    """A backlog must not become a pull request queue nobody drains."""

    config = _config(hermes_sonar_projects={"a": "a", "b": "b", "c": "c"})

    result = module.sweep(_Storage(), config, {})

    assert result["proposed"] == 1
    assert len(wiring["proposals"]) == 1
    assert any("budget spent" in item for item in result["skipped"])


def test_a_declined_proposal_is_reported_and_does_not_spend_budget(wiring) -> None:
    wiring["result"] = {"status": "human_required", "reason": "open_proposal_limit_reached"}
    config = _config(hermes_sonar_projects={"a": "a", "b": "b"})

    result = module.sweep(_Storage(), config, {})

    assert result["proposed"] == 0
    assert len(wiring["proposals"]) == 2
    assert "a: open_proposal_limit_reached" in result["skipped"]


def test_no_configured_projects_makes_no_call(wiring) -> None:
    result = module.sweep(_Storage(), _config(hermes_sonar_projects={}), {})

    assert result == {"proposed": 0, "skipped": ["no_projects_configured"], "projects": 0}
    assert wiring["proposals"] == []


def test_a_project_naming_no_repository_is_skipped(monkeypatch, wiring) -> None:
    monkeypatch.setattr(module.hermes_code_flow, "resolve_repo_config", lambda job, cfg: None)

    result = module.sweep(_Storage(), _config(), {})

    assert result["proposed"] == 0
    assert "maps to no repository" in result["skipped"][0]
    assert wiring["proposals"] == []


def test_a_fetch_error_is_reported_not_raised(wiring) -> None:
    wiring["error"] = "sonar fetch http 503"

    result = module.sweep(_Storage(), _config(), {})

    assert result["proposed"] == 0
    assert result["skipped"] == ["ariadne: sonar fetch http 503"]


def test_a_project_with_nothing_mechanical_is_reported(wiring) -> None:
    wiring["issues"] = [_issue(effort="8h")]

    result = module.sweep(_Storage(), _config(), {})

    assert result["skipped"] == ["ariadne: no mechanically fixable finding"]


def test_a_sweep_never_takes_down_the_scheduler(monkeypatch) -> None:
    """Triage shares this scheduler; a background sweep must not stop it."""

    monkeypatch.setattr(
        module.hermes_code_flow, "code_config",
        lambda config: (_ for _ in ()).throw(RuntimeError("boom")),
    )

    result = module.sweep(_Storage(), _config(), {})

    assert result["proposed"] == 0
    assert "sweep_failed" in result["skipped"][0]


def test_a_project_with_no_job_maps_to_itself() -> None:
    assert module._projects(_config(hermes_sonar_projects={"ariadne": ""})) == {
        "ariadne": "ariadne"
    }
    assert module._projects(SimpleNamespace(hermes_sonar_projects="nope")) == {}


def test_the_client_config_is_built_from_settings() -> None:
    cfg = module.client_config(_config())

    assert cfg["sonar_base_url"] == "http://sonar:9000"
    assert cfg["sonar_token"] == "t"
    assert cfg["sonar_types"] == ["CODE_SMELL"]


@pytest.mark.parametrize(
    ("attr", "func", "bad", "default"),
    [
        ("hermes_sonar_max_per_sweep", "_max_per_sweep", "junk", module.DEFAULT_MAX_PER_SWEEP),
        (
            "hermes_sonar_max_effort_minutes",
            "_max_effort_minutes",
            "junk",
            module.DEFAULT_MAX_EFFORT_MINUTES,
        ),
    ],
)
def test_an_unparseable_bound_falls_back_to_its_default(attr, func, bad, default) -> None:
    assert getattr(module, func)(_config(**{attr: bad})) == default


def test_the_bounds_are_floored() -> None:
    assert module._max_per_sweep(_config(hermes_sonar_max_per_sweep=-4)) == 0
    assert module._max_effort_minutes(_config(hermes_sonar_max_effort_minutes=0)) == 1


def test_the_sweep_is_disabled_until_an_operator_turns_it_on(monkeypatch) -> None:
    """It opens pull requests nobody asked for; that is a deliberate choice."""

    monkeypatch.setattr(module, "settings", SimpleNamespace(hermes_sonar_enabled=False))

    assert module.run_hermes_sonar_sweep(_Storage()) == {"status": "disabled"}


def test_the_default_deployment_leaves_it_off() -> None:
    """A fresh deployment must not start opening pull requests on its own."""

    from ariadne.settings import settings as real_settings

    assert real_settings.hermes_sonar_enabled is False


def test_the_scheduled_entry_point_runs_the_sweep_when_enabled(monkeypatch) -> None:
    seen = {}
    monkeypatch.setattr(
        module,
        "settings",
        SimpleNamespace(
            hermes_sonar_enabled=True,
            hermes_api_url="http://hermes:8642",
            hermes_api_key="k",
            hermes_run_timeout_seconds=420.0,
        ),
    )
    monkeypatch.setattr(
        module, "sweep", lambda storage, config, cfg: seen.update(cfg=cfg) or {"proposed": 0}
    )

    assert module.run_hermes_sonar_sweep(_Storage()) == {"proposed": 0}
    assert seen["cfg"] == {
        "base_url": "http://hermes:8642",
        "api_key": "k",
        "total_timeout_seconds": 420.0,
    }


def test_the_sweep_prompt_never_claims_a_build_failed() -> None:
    """A green build framed as a failing test is wrong in the first line read."""

    from ariadne.services import hermes_code_prompt

    cfg = {"owner": "bstein", "repo": "ariadne", "base_branch": "master"}
    prompt = hermes_code_prompt.build_patch_prompt(
        "sonar/ariadne/AZ-1",
        module.bundle_for("ariadne", _issue()),
        {**cfg, "allowed_path_prefixes": ["ariadne/"], "allowed_suffixes": [".py"]},
        {"ariadne/services/hermes_code_defects.py": "x"},
    )

    assert prompt.startswith("A static-analysis finding, not a build failure.")
    assert "Failing test evidence bundle" not in prompt
    assert "Static analysis evidence bundle:" in prompt
    assert "The build reported these specific defects" not in prompt
    # The triage skill investigates a failed build; there isn't one.
    assert "$triage-titan-test-failures" not in prompt


def test_a_build_driven_prompt_is_unchanged() -> None:
    from ariadne.services import hermes_code_prompt

    bundle = {"jenkins": {"console_tail": "AssertionError", "console_failures": [], "failed_tests": []}}
    prompt = hermes_code_prompt.build_patch_prompt(
        "ariadne/9", bundle, {"owner": "bstein", "repo": "ariadne"}, {"ariadne/a.py": "x"}
    )

    assert prompt.startswith("Use $triage-titan-test-failures.")
    assert "Failing test evidence bundle:" in prompt


@pytest.mark.parametrize(
    ("bundle", "expected"),
    [
        ({"sonarqube": {"issues": [{"path": "a.py"}]}, "jenkins": {}}, True),
        ({"sonarqube": {"issues": []}, "jenkins": {}}, False),
        ({"jenkins": {"console_tail": "boom"}}, False),
        # Both present: a real failure is the more urgent framing.
        (
            {"sonarqube": {"issues": [{"path": "a.py"}]}, "jenkins": {"console_tail": "boom"}},
            False,
        ),
        ("not-a-dict", False),
    ],
)
def test_a_bundles_origin_decides_the_framing(bundle, expected) -> None:
    from ariadne.services import hermes_code_prompt

    assert hermes_code_prompt.is_quality_sweep(bundle) is expected