ariadne/tests/test_hermes_code_repair.py

from __future__ import annotations

import base64
import json

from ariadne.services import hermes_code_repair as module
from ariadne.services.hermes_code_patch import ProposedPatch


INCIDENT_ID = "hermes-code-demo/7"
BRANCH = "hermes-repair/7"


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

    def json(self):  # type: ignore[no-untyped-def]
        if self._payload is None:
            raise ValueError("no json body")
        return self._payload


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 put(self, url, **kwargs):  # type: ignore[no-untyped-def]
            return self._next("PUT", 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]
    base = {
        "gitea_base_url": "https://scm.example",
        "gitea_token": "secret-token",
        "owner": "bstein",
        "repo": "hermes-code-demo",
        "base_branch": "master",
        "timeout_seconds": 7.5,
    }
    base.update(overrides)
    return base


def _patch(**overrides) -> ProposedPatch:  # type: ignore[no-untyped-def]
    values = {
        "path": "src/discount.py",
        "original": "return price * 0.5",
        "replacement": "return price * 0.9",
        "rationale": "restore the intended discount",
    }
    values.update(overrides)
    return ProposedPatch(**values)


def _push(monkeypatch, responses):  # type: ignore[no-untyped-def]
    calls = _install_http(monkeypatch, responses)
    result = module.push_branch(_cfg(), INCIDENT_ID, 7, _patch(), "patched contents\n")
    return calls, result


def test_fetch_file_success(monkeypatch) -> None:
    calls = _install_http(monkeypatch, [FakeResponse(200, text="file body")])
    contents, error = module.fetch_file(_cfg(), "src/discount.py")
    assert (contents, error) == ("file body", None)
    method, url, kwargs = calls["requests"][0]
    assert method == "GET"
    assert url == "https://scm.example/api/v1/repos/bstein/hermes-code-demo/raw/src/discount.py"
    assert kwargs["params"] == {"ref": "master"}
    assert kwargs["headers"] == {"Authorization": "token secret-token"}
    assert calls["kwargs"] == {"timeout": 7.5}


def test_fetch_file_http_error(monkeypatch) -> None:
    _install_http(monkeypatch, [FakeResponse(404)])
    assert module.fetch_file(_cfg(), "src/discount.py") == (None, "file fetch http 404")


def test_fetch_file_request_exception(monkeypatch) -> None:
    _install_http(monkeypatch, [RuntimeError("connect refused")])
    contents, error = module.fetch_file(_cfg(), "src/discount.py")
    assert contents is None
    assert error == "file fetch failed: connect refused"


def test_fetch_file_without_base_url(monkeypatch) -> None:
    calls = _install_http(monkeypatch, [])
    assert module.fetch_file(_cfg(gitea_base_url=""), "x") == (None, "gitea base url is empty")
    assert calls["requests"] == []


def test_fetch_file_uses_default_timeout(monkeypatch) -> None:
    calls = _install_http(monkeypatch, [FakeResponse(200, text="ok")])
    module.fetch_file(_cfg(timeout_seconds="bad"), "src/discount.py")
    assert calls["kwargs"] == {"timeout": 15.0}


def _pull(number: int, head: str = BRANCH, base: str = "master") -> dict:
    return {
        "number": number,
        "html_url": f"https://scm.example/pulls/{number}",
        "head": {"ref": head},
        "base": {"ref": base},
    }


def _none_found(error=None) -> dict:  # type: ignore[no-untyped-def]
    return {"found": False, "pr_number": None, "url": None, "branch": None, "error": error}


def test_find_open_proposal_matches_repair_branch(monkeypatch) -> None:
    calls = _install_http(monkeypatch, [FakeResponse(200, [_pull(2)])])
    result = module.find_open_proposal(_cfg())
    assert result == {
        "found": True,
        "pr_number": 2,
        "url": "https://scm.example/pulls/2",
        "branch": BRANCH,
        "error": None,
    }
    method, url, kwargs = calls["requests"][0]
    assert (method, url) == ("GET", "https://scm.example/api/v1/repos/bstein/hermes-code-demo/pulls")
    assert kwargs["params"] == {"state": "open", "limit": 50}
    assert kwargs["headers"] == {"Authorization": "token secret-token"}


def test_find_open_proposal_ignores_other_branch_prefixes(monkeypatch) -> None:
    _install_http(monkeypatch, [FakeResponse(200, [_pull(3, head="feature/thing"), _pull(4, head="renovate/x")])])
    assert module.find_open_proposal(_cfg()) == _none_found()


def test_find_open_proposal_ignores_wrong_base_branch(monkeypatch) -> None:
    _install_http(monkeypatch, [FakeResponse(200, [_pull(3, base="develop")])])
    assert module.find_open_proposal(_cfg()) == _none_found()


def test_find_open_proposal_returns_lowest_numbered_match(monkeypatch) -> None:
    payload = [
        _pull(9, head="hermes-repair/9"),
        _pull(2, head="hermes-repair/4"),
        _pull(1, head="other/1"),
        _pull(5, head="hermes-repair/5"),
    ]
    _install_http(monkeypatch, [FakeResponse(200, payload)])
    result = module.find_open_proposal(_cfg())
    assert (result["found"], result["pr_number"], result["branch"]) == (True, 2, "hermes-repair/4")


def test_find_open_proposal_skips_malformed_entries(monkeypatch) -> None:
    payload = ["not a dict", {"number": 1}, {"number": True, "head": {"ref": BRANCH}, "base": {"ref": "master"}}]
    _install_http(monkeypatch, [FakeResponse(200, payload)])
    assert module.find_open_proposal(_cfg()) == _none_found()


def test_find_open_proposal_empty_list(monkeypatch) -> None:
    _install_http(monkeypatch, [FakeResponse(200, [])])
    assert module.find_open_proposal(_cfg()) == _none_found()


def test_find_open_proposal_http_error_fails_open(monkeypatch) -> None:
    _install_http(monkeypatch, [FakeResponse(503)])
    assert module.find_open_proposal(_cfg()) == _none_found("open proposal lookup http 503")


def test_find_open_proposal_malformed_json_fails_open(monkeypatch) -> None:
    _install_http(monkeypatch, [FakeResponse(200)])
    result = module.find_open_proposal(_cfg())
    assert result["found"] is False
    assert result["error"] == "open proposal parse failed: no json body"


def test_find_open_proposal_non_list_payload_fails_open(monkeypatch) -> None:
    _install_http(monkeypatch, [FakeResponse(200, {"message": "nope"})])
    assert module.find_open_proposal(_cfg()) == _none_found("open proposal payload is not a list")


def test_find_open_proposal_request_exception_fails_open(monkeypatch) -> None:
    _install_http(monkeypatch, [RuntimeError("connection reset")])
    assert module.find_open_proposal(_cfg()) == _none_found("open proposal lookup failed: connection reset")


def test_find_open_proposal_without_base_url(monkeypatch) -> None:
    calls = _install_http(monkeypatch, [])
    assert module.find_open_proposal(_cfg(gitea_base_url="")) == _none_found("gitea base url is empty")
    assert calls["requests"] == []


def test_push_branch_success_with_new_branch_payload(monkeypatch) -> None:
    calls, result = _push(monkeypatch, [FakeResponse(200, {"sha": "abc123"}), FakeResponse(201, {})])
    assert result == {"branch": BRANCH, "committed": True, "error": None}
    get_method, get_url, get_kwargs = calls["requests"][0]
    assert (get_method, get_kwargs["params"]) == ("GET", {"ref": "master"})
    assert get_url == "https://scm.example/api/v1/repos/bstein/hermes-code-demo/contents/src/discount.py"
    put_method, put_url, put_kwargs = calls["requests"][1]
    assert (put_method, put_url) == ("PUT", get_url)
    body = put_kwargs["json"]
    assert body["branch"] == "master"
    assert body["new_branch"] == BRANCH
    assert body["sha"] == "abc123"
    assert body["message"] == f"fix(hermes): restore the intended discount (incident {INCIDENT_ID})"
    assert base64.b64decode(body["content"]).decode() == "patched contents\n"
    assert body["author"] == {"name": "Hermes Agent", "email": "hermes@bstein.dev"}
    assert body["committer"] == {"name": "Hermes Agent", "email": "hermes@bstein.dev"}


def test_push_branch_falls_back_to_branch_only_payload(monkeypatch) -> None:
    calls, result = _push(
        monkeypatch,
        [FakeResponse(200, {"sha": "abc123"}), FakeResponse(422, {}), FakeResponse(201, {})],
    )
    assert result == {"branch": BRANCH, "committed": True, "error": None}
    fallback_body = calls["requests"][2][2]["json"]
    assert fallback_body["branch"] == BRANCH
    assert "new_branch" not in fallback_body


def test_push_branch_reports_both_failed_payload_shapes(monkeypatch) -> None:
    _, result = _push(
        monkeypatch,
        [FakeResponse(200, {"sha": "abc123"}), FakeResponse(404, {}), FakeResponse(422, {})],
    )
    assert result["committed"] is False
    assert result["error"] == "commit http 404 then fallback http 422"


def test_push_branch_non_retryable_commit_error(monkeypatch) -> None:
    calls, result = _push(monkeypatch, [FakeResponse(200, {"sha": "abc123"}), FakeResponse(500, {})])
    assert result["error"] == "commit http 500"
    assert len(calls["requests"]) == 2


def test_push_branch_sha_read_error(monkeypatch) -> None:
    _, result = _push(monkeypatch, [FakeResponse(404)])
    assert result == {"branch": BRANCH, "committed": False, "error": "file sha http 404"}


def test_push_branch_sha_missing(monkeypatch) -> None:
    _, result = _push(monkeypatch, [FakeResponse(200, {})])
    assert result["error"] == "file sha missing from contents response"


def test_push_branch_request_exception(monkeypatch) -> None:
    _, result = _push(monkeypatch, [RuntimeError("boom")])
    assert result == {"branch": BRANCH, "committed": False, "error": "branch push failed: boom"}


def test_push_branch_refuses_protected_branch_names(monkeypatch) -> None:
    calls = _install_http(monkeypatch, [])
    for name in ("master", "main", " "):
        monkeypatch.setattr(module, "_branch_name", lambda build_number, name=name: name)
        result = module.push_branch(_cfg(), INCIDENT_ID, 7, _patch(), "contents")
        assert result["committed"] is False
        assert result["error"] == f"refusing branch {name!r}"
    assert calls["requests"] == []


def test_push_branch_without_base_url(monkeypatch) -> None:
    calls = _install_http(monkeypatch, [])
    result = module.push_branch(_cfg(gitea_base_url=""), INCIDENT_ID, 7, _patch(), "contents")
    assert result == {"branch": BRANCH, "committed": False, "error": "gitea base url is empty"}
    assert calls["requests"] == []


def test_open_pull_request_created(monkeypatch) -> None:
    calls = _install_http(
        monkeypatch,
        [FakeResponse(201, {"number": 5, "html_url": "https://scm.example/pulls/5"})],
    )
    result = module.open_pull_request(_cfg(), INCIDENT_ID, 7, BRANCH, _patch(), "root cause analysis")
    assert result == {"pr_number": 5, "url": "https://scm.example/pulls/5", "error": None}
    method, url, kwargs = calls["requests"][0]
    assert (method, url) == ("POST", "https://scm.example/api/v1/repos/bstein/hermes-code-demo/pulls")
    payload = kwargs["json"]
    assert payload["head"] == BRANCH
    assert payload["base"] == "master"
    assert payload["title"] == f"fix(hermes): repair {INCIDENT_ID}"
    body = payload["body"]
    assert INCIDENT_ID in body
    assert "root cause analysis" in body
    assert "restore the intended discount" in body
    assert "`src/discount.py`" in body
    assert "Proposed by Hermes; validated and pushed by Ariadne; requires human review — no automatic merge." in body
    assert "secret-token" not in json.dumps(payload)


def test_open_pull_request_conflict_returns_existing(monkeypatch) -> None:
    _install_http(monkeypatch, [FakeResponse(409, {"number": 9, "html_url": "https://scm.example/pulls/9"})])
    result = module.open_pull_request(_cfg(), INCIDENT_ID, 7, BRANCH, _patch(), "analysis")
    assert result == {"pr_number": 9, "url": "https://scm.example/pulls/9", "error": None}


def test_open_pull_request_conflict_without_payload(monkeypatch) -> None:
    _install_http(monkeypatch, [FakeResponse(409)])
    result = module.open_pull_request(_cfg(), INCIDENT_ID, 7, BRANCH, _patch(), "analysis")
    assert result == {"pr_number": None, "url": None, "error": "pull request already exists"}


def test_open_pull_request_http_error(monkeypatch) -> None:
    _install_http(monkeypatch, [FakeResponse(500)])
    result = module.open_pull_request(_cfg(), INCIDENT_ID, 7, BRANCH, _patch(), "analysis")
    assert result == {"pr_number": None, "url": None, "error": "pull request http 500"}


def test_open_pull_request_created_with_bad_payload(monkeypatch) -> None:
    _install_http(monkeypatch, [FakeResponse(201, {"number": True})])
    result = module.open_pull_request(_cfg(), INCIDENT_ID, 7, BRANCH, _patch(), "analysis")
    assert result == {"pr_number": None, "url": None, "error": None}


def test_open_pull_request_request_exception(monkeypatch) -> None:
    _install_http(monkeypatch, [RuntimeError("down")])
    result = module.open_pull_request(_cfg(), INCIDENT_ID, 7, BRANCH, _patch(), "analysis")
    assert result == {"pr_number": None, "url": None, "error": "pull request failed: down"}


def test_open_pull_request_without_base_url(monkeypatch) -> None:
    calls = _install_http(monkeypatch, [])
    result = module.open_pull_request(_cfg(gitea_base_url=" "), INCIDENT_ID, 7, BRANCH, _patch(), "a")
    assert result == {"pr_number": None, "url": None, "error": "gitea base url is empty"}
    assert calls["requests"] == []