ariadne/tests/test_hermes_sonar_advice.py

"""Tests for filing a suggested fix when a finding cannot become a patch."""

from __future__ import annotations

import json
from types import SimpleNamespace

import pytest

from ariadne.services import hermes_sonar_advice as module


ISSUE = {
    "key": "AZ1",
    "rule": "python:S3776",
    "severity": "CRITICAL",
    "type": "CODE_SMELL",
    "effort": "11min",
    "path": "ariadne/services/thing.py",
    "line": 42,
    "message": "Refactor this function to reduce its Cognitive Complexity from 21 to the 15 allowed.",
}

REPO_CFG = {
    "owner": "bstein",
    "repo": "ariadne",
    "gitea_base_url": "https://scm.example",
    "gitea_token": "t",
    "base_branch": "master",
}

INCIDENT = "sonar/ariadne/python:S3776/AZ1"

SUGGESTION = {
    "path": "ariadne/services/thing.py",
    "explanation": "Extract the validation branch into its own helper.",
    "code": "def _validate(row):\n    return bool(row)",
}


def _config(**overrides):
    values = {
        "hermes_ui_url": "https://agent.example",
        "hermes_sonar_ui_url": "https://quality.example",
        "hermes_gitea_base_url": "https://scm.example",
        "hermes_gitea_token": "t",
    }
    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))


def _response(**overrides):
    payload = {
        "incident_id": INCIDENT,
        "analysis": "The function branches five ways over the same row.",
        "code_suggestions": [dict(SUGGESTION)],
        "human_required": True,
        "reason": "needs a maintainer",
    }
    payload.update(overrides)
    return json.dumps(payload)


@pytest.fixture
def wiring(monkeypatch):
    state = {
        "existing": {"found": False},
        "contents": "def thing():\n    pass\n",
        "fetch_error": None,
        "run": SimpleNamespace(status="completed", output=_response(), run_id="run_x"),
        "created": {"issue_number": 9, "url": "https://scm.example/issues/9", "error": None},
        "filed_context": [],
    }
    monkeypatch.setattr(
        module.hermes_incident_issue, "find_open_incident_issue",
        lambda cfg, job, classification, incident: state["existing"],
    )
    monkeypatch.setattr(
        module.hermes_code_repair, "fetch_file",
        lambda cfg, path: (state["contents"], state["fetch_error"]),
    )
    monkeypatch.setattr(module.hermes_agent_client, "run_triage", lambda cfg, prompt: state["run"])

    def _create(cfg, context):
        state["filed_context"].append(context)
        return state["created"]

    monkeypatch.setattr(module.hermes_incident_issue, "create_incident_issue", _create)
    return state


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

    result = module.advise(storage, _config(), {}, REPO_CFG, "ariadne", dict(ISSUE))

    assert result == {"filed": True, "reason": "issue_filed", "url": "https://scm.example/issues/9"}
    context = wiring["filed_context"][0]
    assert context["incident_id"] == INCIDENT
    assert context["classification"] == "python:S3776"
    assert context["code_suggestions"][0].code.startswith("def _validate")
    assert context["reason"].startswith("The function branches")


def test_the_issue_links_to_both_the_finding_and_the_run(wiring) -> None:
    module.advise(_Storage(), _config(), {}, REPO_CFG, "ariadne", dict(ISSUE))

    context = wiring["filed_context"][0]
    assert context["build_url"] == (
        "https://quality.example/project/issues?resolved=false&id=ariadne&open=AZ1"
    )
    assert context["run_url"] == "https://agent.example/chat?resume=run_x"


def test_the_issue_says_why_nothing_was_patched(wiring) -> None:
    """Otherwise it reads as the automation having failed."""

    module.advise(_Storage(), _config(), {}, REPO_CFG, "ariadne", dict(ISSUE))

    assert "no automated patch was possible" in wiring["filed_context"][0]["authorize_reason"]


def test_a_rule_already_filed_is_not_filed_again(wiring) -> None:
    """One rule is one root cause; an issue per instance buries the repo."""

    wiring["existing"] = {"found": True, "url": "https://scm.example/issues/3"}

    result = module.advise(_Storage(), _config(), {}, REPO_CFG, "ariadne", dict(ISSUE))

    assert result["filed"] is False
    assert result["reason"] == "issue_already_open"
    assert wiring["filed_context"] == []


def test_an_unreadable_file_files_nothing(wiring) -> None:
    wiring["contents"] = None
    wiring["fetch_error"] = "http 404"

    result = module.advise(_Storage(), _config(), {}, REPO_CFG, "ariadne", dict(ISSUE))

    assert result["filed"] is False
    assert "file_fetch_failed" in result["reason"]


@pytest.mark.parametrize(
    ("status", "output"),
    [("failed", ""), ("completed", ""), ("timeout", "{}")],
)
def test_a_run_that_produced_nothing_files_nothing(wiring, status, output) -> None:
    wiring["run"] = SimpleNamespace(status=status, output=output, run_id="r")

    assert module.advise(_Storage(), _config(), {}, REPO_CFG, "ariadne", dict(ISSUE))["filed"] is False


def test_a_response_with_no_suggestions_files_nothing(wiring) -> None:
    wiring["run"] = SimpleNamespace(
        status="completed", output=_response(code_suggestions=[]), run_id="r"
    )

    result = module.advise(_Storage(), _config(), {}, REPO_CFG, "ariadne", dict(ISSUE))

    assert result["reason"] == "no_suggestions_returned"


def test_a_failed_creation_is_reported(wiring) -> None:
    wiring["created"] = {"issue_number": None, "url": None, "error": "http 500"}

    assert module.advise(_Storage(), _config(), {}, REPO_CFG, "ariadne", dict(ISSUE))["filed"] is False


def test_every_attempt_is_recorded(wiring) -> None:
    storage = _Storage()
    module.advise(storage, _config(), {}, REPO_CFG, "ariadne", dict(ISSUE))

    assert storage.events[0][0] == module.ADVICE_EVENT_TYPE
    assert storage.events[0][1]["incident_id"] == INCIDENT


def test_advice_never_raises(monkeypatch) -> None:
    """It runs after a pull request was already declined; it must not erase it."""

    monkeypatch.setattr(
        module.hermes_incident_issue, "find_open_incident_issue",
        lambda *a, **k: (_ for _ in ()).throw(RuntimeError("boom")),
    )

    result = module.advise(_Storage(), _config(), {}, REPO_CFG, "ariadne", dict(ISSUE))

    assert result["filed"] is False
    assert "advice_failed" in result["reason"]


@pytest.mark.parametrize(
    "raw",
    [
        "",
        "not json at all",
        '{"incident_id": "someone/else", "analysis": "a", "code_suggestions": [], "human_required": true, "reason": "r"}',
        '{"incident_id": "' + INCIDENT + '", "code_suggestions": "nope"}',
    ],
)
def test_an_unusable_response_yields_no_suggestions(raw) -> None:
    assert module.parse_advice(raw, INCIDENT) == ([], "")


def test_a_good_response_is_parsed() -> None:
    suggestions, analysis = module.parse_advice("noise " + _response() + " trailing", INCIDENT)

    assert len(suggestions) == 1
    assert suggestions[0].path == "ariadne/services/thing.py"
    assert analysis.startswith("The function branches")


def test_the_prompt_states_that_nothing_is_applied() -> None:
    prompt = module._prompt(INCIDENT, dict(ISSUE), "def thing(): pass")

    assert "not a build failure" in prompt
    assert "Nothing you return is applied" in prompt
    assert "python:S3776" in prompt
    assert "def thing(): pass" in prompt


def test_the_prompt_bounds_the_file_it_sends() -> None:
    prompt = module._prompt(INCIDENT, dict(ISSUE), "x" * 90000)

    assert len(prompt) < 45000