ariadne/tests/test_hermes_code_flow_repos.py

"""Multi-repository code-proposal tests for the Hermes code flow.

The legacy single-repo demo path lives in test_hermes_code_flow.py; this file
covers per-job repository resolution and evidence-driven candidate selection.
Shared fakes are imported from that module so both drive the same storage,
Gitea, and Hermes stand-ins.
"""

from __future__ import annotations

import json
from types import SimpleNamespace

from ariadne.services import hermes_code_flow as module
from tests.test_hermes_code_flow import (
    JOB,
    FakeStorage,
    _code_cfg,
    _event,
    _hermes_cfg,
    _install,
    _run,
)


SERVICE_JOB = "titan-api"
SERVICE_INCIDENT_ID = f"{SERVICE_JOB}/9"
SERVICE_SOURCE = "def balance(rows):\n    return sum(rows) - 1\n"
SERVICE_TEST = "def test_balance():\n    assert balance([1, 2]) == 3\n"
SERVICE_BUNDLE = {
    "incident_id": SERVICE_INCIDENT_ID,
    "jenkins": {
        "job": SERVICE_JOB,
        "console_failures": [
            {
                "marker": "FAILED ",
                "line_number": 40,
                "text": (
                    "FAILED tests/test_ledger.py::test_balance\n"
                    "src/ledger.py:2: AssertionError\n"
                    "src/ledger.py:2: in balance"
                ),
            }
        ],
        "console_tail": "ERROR: script returned exit code 1",
    },
    "log_evidence": {"records": []},
}


def _settings(**overrides) -> SimpleNamespace:  # type: ignore[no-untyped-def]
    values = {
        "hermes_code_job": JOB,
        "hermes_code_candidate_path": "src/discount.py",
        "hermes_code_allowed_prefixes": ["src/"],
        "hermes_code_allowed_suffixes": [".py"],
        "hermes_code_max_patch_bytes": 4000,
        "hermes_code_max_changed_lines": 20,
        "hermes_gitea_base_url": "https://scm.example",
        "hermes_gitea_token": "secret-token",
        "hermes_code_owner": "bstein",
        "hermes_code_repo": "hermes-code-demo",
        "hermes_code_base_branch": "master",
    }
    values.update(overrides)
    return SimpleNamespace(**values)


def _service_cfg(**overrides) -> dict:  # type: ignore[no-untyped-def]
    settings = _settings(
        hermes_code_repos=f"{SERVICE_JOB}=titan/api, other-job = titan/other",
        hermes_code_prefixes=f"{SERVICE_JOB}=src/|tests/",
        hermes_code_suffixes=f"{SERVICE_JOB}=.py",
        hermes_code_base_branches=f"{SERVICE_JOB}=main",
        **overrides,
    )
    return module.code_config(settings)


def _propose_service(monkeypatch, cfg=None, job=SERVICE_JOB, **kwargs):  # type: ignore[no-untyped-def]
    storage = FakeStorage()
    calls = _install(monkeypatch, **kwargs)
    result = module.propose_code_fix(
        storage,
        SERVICE_INCIDENT_ID,
        job,
        9,
        SERVICE_BUNDLE,
        _hermes_cfg(),
        _service_cfg() if cfg is None else cfg,
    )
    return storage, calls, result


def _service_fetch(**overrides) -> dict:  # type: ignore[no-untyped-def]
    base = {
        "src/ledger.py": (SERVICE_SOURCE, None),
        "tests/test_ledger.py": (SERVICE_TEST, None),
    }
    base.update(overrides)
    return base


def _service_output(path: str, original: str, replacement: str) -> str:
    return json.dumps(
        {
            "incident_id": SERVICE_INCIDENT_ID,
            "analysis": "The balance helper is off by one.",
            "patch": {
                "path": path,
                "original": original,
                "replacement": replacement,
                "rationale": "restore the intended balance",
            },
            "human_required": False,
            "reason": "small localized fix",
        }
    )


def test_code_config_defaults_when_multi_repo_settings_absent() -> None:
    cfg = module.code_config(_settings())
    assert cfg["repos"] == {}
    assert cfg["job_prefixes"] == {}
    assert cfg["job_suffixes"] == {}
    assert cfg["job_base_branches"] == {}
    assert cfg["max_candidates"] == 3
    assert cfg["max_context_chars"] == 60000


def test_code_config_parses_per_job_maps() -> None:
    cfg = _service_cfg(hermes_code_max_candidates=5, hermes_code_max_context_chars=1234)
    assert cfg["repos"] == {
        SERVICE_JOB: {"owner": "titan", "repo": "api"},
        "other-job": {"owner": "titan", "repo": "other"},
    }
    assert cfg["job_prefixes"] == {SERVICE_JOB: ["src/", "tests/"]}
    assert cfg["job_suffixes"] == {SERVICE_JOB: [".py"]}
    assert cfg["job_base_branches"] == {SERVICE_JOB: "main"}
    assert cfg["max_candidates"] == 5
    assert cfg["max_context_chars"] == 1234


def test_code_config_ignores_malformed_pairs() -> None:
    cfg = module.code_config(
        _settings(
            hermes_code_repos="broken, =owner/repo, job=, job2=owneronly, job3=titan/api",
            hermes_code_prefixes="job3=,job4=src/",
            hermes_code_max_candidates="not-a-number",
        )
    )
    assert cfg["repos"] == {"job3": {"owner": "titan", "repo": "api"}}
    assert cfg["job_prefixes"] == {"job4": ["src/"]}
    assert cfg["max_candidates"] == 3


def test_resolve_repo_config_uses_per_job_mapping() -> None:
    resolved = module.resolve_repo_config(SERVICE_JOB, _service_cfg())
    assert resolved is not None
    assert resolved["owner"] == "titan"
    assert resolved["repo"] == "api"
    assert resolved["base_branch"] == "main"
    assert resolved["allowed_path_prefixes"] == ["src/", "tests/"]
    assert resolved["allowed_suffixes"] == [".py"]
    assert resolved["candidate_path"] == ""
    assert resolved["gitea_token"] == "secret-token"
    assert resolved["max_patch_bytes"] == 4000


def test_resolve_repo_config_falls_back_to_single_repo_settings() -> None:
    resolved = module.resolve_repo_config(JOB, _service_cfg())
    assert resolved is not None
    assert (resolved["owner"], resolved["repo"]) == ("bstein", "hermes-code-demo")
    assert resolved["base_branch"] == "master"
    assert resolved["allowed_path_prefixes"] == ["src/"]
    assert resolved["candidate_path"] == "src/discount.py"


def test_resolve_repo_config_is_unchanged_for_legacy_only_settings() -> None:
    assert module.resolve_repo_config(JOB, _code_cfg()) == _code_cfg()


def test_resolve_repo_config_returns_none_for_unmapped_job() -> None:
    assert module.resolve_repo_config("unmapped-job", _service_cfg()) is None
    assert module.resolve_repo_config("", _service_cfg()) is None


def test_unmapped_job_requires_human_without_http(monkeypatch) -> None:
    storage, calls, result = _propose_service(monkeypatch, job="unmapped-job")
    assert result == {"status": "human_required", "reason": "no_repo_mapping"}
    assert calls["lookups"] == []
    assert calls["fetches"] == []
    assert calls["runs"] == []
    detail = _event(storage)
    assert detail["reject_reason"] == "no_repo_mapping"
    assert detail["repo"] is None
    assert detail["candidates"] == []
    assert detail["chosen_path"] is None


def test_no_candidate_files_makes_no_model_call(monkeypatch) -> None:
    bundle = {"incident_id": SERVICE_INCIDENT_ID, "jenkins": {"console_tail": "no file here"}}
    storage = FakeStorage()
    calls = _install(monkeypatch)
    result = module.propose_code_fix(
        storage, SERVICE_INCIDENT_ID, SERVICE_JOB, 9, bundle, _hermes_cfg(), _service_cfg()
    )
    assert result == {"status": "human_required", "reason": "no_candidate_files"}
    assert len(calls["lookups"]) == 1
    assert calls["fetches"] == []
    assert calls["runs"] == []
    assert _event(storage)["reject_reason"] == "no_candidate_files"


def test_service_repo_offers_ranked_candidates(monkeypatch) -> None:
    storage, calls, result = _propose_service(
        monkeypatch,
        fetch=_service_fetch(),
        run=_run(output=_service_output("src/ledger.py", "sum(rows) - 1", "sum(rows)")),
    )
    assert result["status"] == "pr_opened"
    assert [path for _cfg, path in calls["fetches"]] == ["src/ledger.py", "tests/test_ledger.py"]
    push_cfg, _incident, build_number, patch, contents = calls["pushes"][0]
    assert (push_cfg["owner"], push_cfg["repo"], push_cfg["base_branch"]) == ("titan", "api", "main")
    assert (build_number, patch.path) == (9, "src/ledger.py")
    assert contents == "def balance(rows):\n    return sum(rows)\n"
    detail = _event(storage)
    assert detail["repo"] == "titan/api"
    assert detail["candidates"] == ["src/ledger.py", "tests/test_ledger.py"]
    assert detail["chosen_path"] == "src/ledger.py"


def test_prompt_lists_every_offered_candidate(monkeypatch) -> None:
    _, calls, _ = _propose_service(
        monkeypatch,
        fetch=_service_fetch(),
        run=_run(output=_service_output("src/ledger.py", "sum(rows) - 1", "sum(rows)")),
    )
    prompt = calls["runs"][0][1]
    assert "The repository is titan/api branch main." in prompt
    assert "- src/ledger.py\n- tests/test_ledger.py" in prompt
    assert "`patch.path` MUST be exactly one of these candidate paths" in prompt
    assert "Current content of the candidate file src/ledger.py:" in prompt
    assert "Current content of the candidate file tests/test_ledger.py:" in prompt
    assert SERVICE_SOURCE in prompt
    assert prompt.rstrip().endswith(SERVICE_TEST.rstrip())


def test_failed_candidate_fetch_is_skipped(monkeypatch) -> None:
    storage, calls, result = _propose_service(
        monkeypatch,
        fetch=_service_fetch(**{"tests/test_ledger.py": (None, "file fetch http 404")}),
        run=_run(output=_service_output("src/ledger.py", "sum(rows) - 1", "sum(rows)")),
    )
    assert result["status"] == "pr_opened"
    assert len(calls["fetches"]) == 2
    prompt = calls["runs"][0][1]
    assert "tests/test_ledger.py" not in prompt.split("Failing test evidence bundle:")[0]
    assert _event(storage)["candidates"] == ["src/ledger.py"]


def test_all_candidate_fetches_failing_requires_human(monkeypatch) -> None:
    storage, calls, result = _propose_service(
        monkeypatch, fetch={"src/ledger.py": (None, "file fetch http 500")}
    )
    assert result == {"status": "human_required", "reason": "candidate_fetch_failed: file fetch http 500"}
    assert calls["runs"] == []
    assert _event(storage)["candidates"] == []


def test_context_budget_skips_oversized_candidate(monkeypatch) -> None:
    cfg = _service_cfg(hermes_code_max_context_chars=len(SERVICE_SOURCE) + 1)
    storage, calls, result = _propose_service(
        monkeypatch,
        cfg=cfg,
        fetch=_service_fetch(),
        run=_run(output=_service_output("src/ledger.py", "sum(rows) - 1", "sum(rows)")),
    )
    assert result["status"] == "pr_opened"
    assert len(calls["fetches"]) == 2
    assert SERVICE_TEST not in calls["runs"][0][1]
    assert _event(storage)["candidates"] == ["src/ledger.py"]


def test_patch_path_not_offered_requires_human(monkeypatch) -> None:
    storage, calls, result = _propose_service(
        monkeypatch,
        fetch=_service_fetch(),
        run=_run(output=_service_output("src/other.py", "sum(rows) - 1", "sum(rows)")),
    )
    assert result["status"] == "human_required"
    assert result["reason"].startswith("patch_path_not_offered: got 'src/other.py'")
    assert calls["pushes"] == []
    assert _event(storage)["chosen_path"] is None


def test_validation_resolves_against_the_chosen_file(monkeypatch) -> None:
    storage, calls, result = _propose_service(
        monkeypatch,
        fetch=_service_fetch(),
        run=_run(output=_service_output("tests/test_ledger.py", "== 3", "== 2")),
    )
    assert result["status"] == "pr_opened"
    assert result["path"] == "tests/test_ledger.py"
    _cfg, _incident, _build, patch, contents = calls["pushes"][0]
    assert patch.path == "tests/test_ledger.py"
    assert contents == "def test_balance():\n    assert balance([1, 2]) == 2\n"
    assert _event(storage)["chosen_path"] == "tests/test_ledger.py"