nisse/nisse-core/tests/test_gitea.py

import json

import httpx
import pytest

from nisse_core.gitea import (
    PR_PAGE_SIZE,
    Comment,
    GiteaClient,
    GiteaError,
    pull_request_reconciler,
)
from nisse_core.state import Intent

INTENT = Intent(
    id=1,
    kind="open_pr",
    payload={"branch": "milestone-1"},
    status="pending",
    created_at="t",
    resolved_at=None,
)


def client(handler) -> GiteaClient:
    return GiteaClient(
        "https://gitea.test/", "tok", transport=httpx.MockTransport(handler)
    )


def pr_json(number: int, head: str, state: str = "open") -> dict:
    return {
        "number": number,
        "state": state,
        "head": {"ref": head},
        "html_url": f"https://gitea.test/o/r/pulls/{number}",
    }


def test_open_pull_request_posts_expected_body() -> None:
    seen: dict = {}

    def handler(request: httpx.Request) -> httpx.Response:
        seen["url"] = str(request.url)
        seen["auth"] = request.headers.get("authorization")
        seen["body"] = json.loads(request.content)
        return httpx.Response(201, json=pr_json(7, "milestone-1"))

    pr = client(handler).open_pull_request(
        "o/r", head="milestone-1", base="main", title="T", body="B"
    )
    assert seen["url"] == "https://gitea.test/api/v1/repos/o/r/pulls"
    assert seen["auth"] == "token tok"
    assert seen["body"] == {
        "head": "milestone-1",
        "base": "main",
        "title": "T",
        "body": "B",
    }
    assert pr.number == 7
    assert pr.head == "milestone-1"


def test_find_pull_request_matches_head_branch() -> None:
    handler = lambda request: httpx.Response(200, json=[pr_json(3, "other")])
    assert client(handler).find_pull_request("o/r", "milestone-1") is None


def test_find_pull_request_also_finds_closed() -> None:
    # A crash after opening must not be answered with "nothing open, open
    # another" — the effect already happened.
    handler = lambda request: httpx.Response(
        200, json=[pr_json(4, "milestone-1", "closed")]
    )
    found = client(handler).find_pull_request("o/r", "milestone-1")
    assert found is not None
    assert found.number == 4


def test_find_pull_request_pages_past_the_first_page() -> None:
    # The match sits on page 2; sampling only the newest page would report
    # "absent" and licence a duplicate PR.
    def handler(request: httpx.Request) -> httpx.Response:
        page = int(request.url.params.get("page", 1))
        if page == 1:
            return httpx.Response(
                200, json=[pr_json(i, f"other-{i}") for i in range(PR_PAGE_SIZE)]
            )
        return httpx.Response(200, json=[pr_json(999, "milestone-1", "closed")])

    found = client(handler).find_pull_request("o/r", "milestone-1")
    assert found is not None
    assert found.number == 999


def test_short_page_ends_the_search_definitively() -> None:
    calls: list[int] = []

    def handler(request: httpx.Request) -> httpx.Response:
        calls.append(int(request.url.params.get("page", 1)))
        return httpx.Response(200, json=[pr_json(1, "other")])

    assert client(handler).find_pull_request("o/r", "milestone-1") is None
    assert calls == [1]  # one short page proves absence; no further paging


def test_unbounded_history_raises_rather_than_guessing() -> None:
    # Never answer "absent" from an incomplete search: the reconciler turns
    # this error into "unverifiable" and gates to the human.
    handler = lambda request: httpx.Response(
        200, json=[pr_json(i, f"other-{i}") for i in range(PR_PAGE_SIZE)]
    )
    with pytest.raises(GiteaError, match="without reaching the end"):
        client(handler).find_pull_request("o/r", "milestone-1")


def test_error_status_raises_gitea_error() -> None:
    handler = lambda request: httpx.Response(403, text="forbidden")
    with pytest.raises(GiteaError, match="403"):
        client(handler).whoami()


def test_list_and_post_comments() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        if request.method == "POST":
            return httpx.Response(
                201,
                json={
                    "id": 9,
                    "user": {"login": "don"},
                    "body": "hi",
                    "created_at": "t",
                },
            )
        return httpx.Response(
            200,
            json=[
                {
                    "id": 8,
                    "user": {"login": "don"},
                    "body": "approve rev-1",
                    "created_at": "t",
                }
            ],
        )

    c = client(handler)
    assert c.list_comments("o/r", 1) == [
        Comment(id=8, author="don", body="approve rev-1", created_at="t")
    ]
    assert c.post_comment("o/r", 1, "hi").id == 9


def test_list_comments_passes_since() -> None:
    seen: dict = {}

    def handler(request: httpx.Request) -> httpx.Response:
        seen["since"] = request.url.params.get("since")
        return httpx.Response(200, json=[])

    client(handler).list_comments("o/r", 1, since="2026-09-07T00:00:00Z")
    assert seen["since"] == "2026-09-07T00:00:00Z"


# --- reconciler (L8 external truth) --------------------------------------


def test_reconciler_reports_existing_pr_as_done() -> None:
    handler = lambda request: httpx.Response(200, json=[pr_json(5, "milestone-1")])
    assert pull_request_reconciler(client(handler), "o/r")(INTENT) is True


def test_reconciler_reports_missing_pr_as_not_done() -> None:
    handler = lambda request: httpx.Response(200, json=[])
    assert pull_request_reconciler(client(handler), "o/r")(INTENT) is False


def test_unreachable_gitea_is_unverifiable_not_a_guess() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        raise httpx.ConnectError("refused")

    # None gates to the human: a delayed decision is cheap, a duplicate PR
    # is not.
    assert pull_request_reconciler(client(handler), "o/r")(INTENT) is None


def test_intent_without_a_branch_is_unverifiable() -> None:
    handler = lambda request: httpx.Response(200, json=[])
    blank = Intent(
        id=2,
        kind="open_pr",
        payload={},
        status="pending",
        created_at="t",
        resolved_at=None,
    )
    assert pull_request_reconciler(client(handler), "o/r")(blank) is None


def test_closing_an_issue_and_a_pr() -> None:
    seen: list[tuple[str, str, dict]] = []

    def handler(request: httpx.Request) -> httpx.Response:
        body = json.loads(request.content) if request.content else {}
        seen.append((request.method, request.url.path, body))
        if request.method == "POST":
            return httpx.Response(
                201, json={"id": 5, "user": {"login": "bot"}, "body": body["body"]}
            )
        return httpx.Response(201, json={})

    gitea = client(handler)
    gitea.post_comment("o/r", 10, "abandoned by don")
    gitea.close_issue("o/r", 10)
    gitea.post_comment("o/r", 20, "abandoned by don")
    gitea.close_pull_request("o/r", 20)

    assert seen == [
        ("POST", "/api/v1/repos/o/r/issues/10/comments", {"body": "abandoned by don"}),
        ("PATCH", "/api/v1/repos/o/r/issues/10", {"state": "closed"}),
        ("POST", "/api/v1/repos/o/r/issues/20/comments", {"body": "abandoned by don"}),
        ("PATCH", "/api/v1/repos/o/r/pulls/20", {"state": "closed"}),
    ]


def test_assigned_at_is_the_last_assignment_across_pages() -> None:
    def assigned(login: str, at: str, removed: bool = False) -> dict:
        return {
            "type": "assignees",
            "assignee": {"login": login},
            "removed_assignee": removed,
            "created_at": at,
        }

    filler = {"type": "comment", "created_at": "t0"}
    pages = {
        1: [assigned("dev-agent", "t1")] + [filler] * (PR_PAGE_SIZE - 1),
        2: [
            assigned("dev-agent", "t2", removed=True),
            assigned("someone", "t3"),
            assigned("dev-agent", "t4"),
        ],
    }

    def handler(request: httpx.Request) -> httpx.Response:
        assert request.url.path == "/api/v1/repos/o/r/issues/6/timeline"
        return httpx.Response(200, json=pages[int(request.url.params["page"])])

    assert client(handler).assigned_at("o/r", 6, "dev-agent") == "t4"
    assert client(handler).assigned_at("o/r", 6, "nobody") is None


def test_approvals_are_the_standing_approving_reviews() -> None:
    reviews = [
        {"state": "APPROVED", "user": {"login": "don"}, "commit_id": "abc"},
        {
            "state": "APPROVED",
            "user": {"login": "don"},
            "commit_id": "old",
            "stale": True,
        },
        {
            "state": "APPROVED",
            "user": {"login": "x"},
            "commit_id": "abc",
            "dismissed": True,
        },
        {"state": "REQUEST_CHANGES", "user": {"login": "don"}, "commit_id": "abc"},
        {"state": "COMMENT", "user": {"login": "don"}, "commit_id": "abc"},
    ]
    handler = lambda request: httpx.Response(200, json=reviews)
    approvals = client(handler).approvals("o/r", 3)
    assert [(a.author, a.commit, a.stale) for a in approvals] == [
        ("don", "abc", False),
        ("don", "old", True),
    ]


def test_checks_are_the_combined_status_or_none() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        assert request.url.path == "/api/v1/repos/o/r/commits/abc/status"
        return httpx.Response(200, json=answer)

    answer: dict = {"state": "", "total_count": 0, "statuses": []}
    assert client(handler).checks("o/r", "abc") == "none"
    answer = {"state": "pending", "total_count": 1, "statuses": [{}]}
    assert client(handler).checks("o/r", "abc") == "pending"
    answer = {"state": "success", "total_count": 2, "statuses": [{}, {}]}
    assert client(handler).checks("o/r", "abc") == "success"


def test_a_pull_request_carries_its_head_commit_and_mergeability() -> None:
    data = {
        **pr_json(5, "m-1"),
        "head": {"ref": "m-1", "sha": "f00"},
        "mergeable": True,
    }
    pr = client(lambda request: httpx.Response(200, json=data)).pull_request("o/r", 5)
    assert (pr.head_sha, pr.mergeable) == ("f00", True)


def test_a_check_branch_is_made_through_the_api() -> None:
    seen: list = []

    def handler(request: httpx.Request) -> httpx.Response:
        seen.append((request.method, request.url.path, request.content))
        return httpx.Response(201, json={})

    gitea = client(handler)
    gitea.branch_with_file(
        "o/r", base="main", branch="chk", path="f", text="x", message="m"
    )
    gitea.delete_branch("o/r", "chk")
    body = json.loads(seen[0][2])
    assert (seen[0][0], seen[0][1]) == ("POST", "/api/v1/repos/o/r/contents/f")
    assert (body["branch"], body["new_branch"], body["content"]) == (
        "main",
        "chk",
        "eA==",
    )
    assert seen[1][:2] == ("DELETE", "/api/v1/repos/o/r/branches/chk")