import json from pathlib import Path import httpx import pytest from nisse_dev.gitea import ( PR_PAGE_SIZE, Comment, GiteaClient, GiteaError, pull_request_reconciler, ) from nisse_dev.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 (A3 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_comments_first_then_closes_issue_and_pr() -> None: from nisse_dev.approvals import PlanBoard from nisse_dev.git import GitRepo from nisse_dev.publish import MilestonePublisher 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) PlanBoard(gitea, "o/r").close(10, "abandoned by don") publisher = MilestonePublisher( None, # type: ignore[arg-type] -- closing touches neither store nor git gitea, "o/r", GitRepo(Path(".")), base_url="https://gitea.test", account="bot", token="tok", ) publisher.close_pull_request(20, "abandoned by don") 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"}), ]