d8faabf2e5
Review gate: replaces the arbitrary repo_dir test target with a review_workspace seam that builds a clean temporary git worktree from the configured clone, fetches the exact base/head SHAs, and constructs a real synthetic merge of the reviewed head onto the reviewed base. Verdicts pin repo, PR, head SHA, and base SHA; tests + git diff --check run inside the workspace with an explicit base..HEAD range; a merge conflict yields a deterministic BOUNCE with conflicted-file evidence and the suite is never run against a fallback checkout; no workspace configured fails closed to NEEDS_GUIDANCE. Worktrees/refs are cleaned up on success, conflict, timeout, and exceptions; the configured clone and worker issue worktrees are never mutated. Policy: before applying any verdict the PR is re-read and the verdict is refused (outcome=stale/PAIR_MOVED, zero side effects) if the head OR base moved after evaluation — the new pair stays eligible for a fresh review. Client: normalized PR shape gains base_sha (REVIEW_CLONE_DIR is the new config knob). Hermetic tests build local temp repos proving clean merge, conflict, moved head/base, no shared-checkout mutation, correct command cwd/range, and cleanup on failure (no network).
617 lines
25 KiB
Python
617 lines
25 KiB
Python
"""Hermetic tests for the production Gitea client (issue #28).
|
|
|
|
A local threaded HTTP stub on ``127.0.0.1`` stands in for the Gitea
|
|
instance: requests are scripted per route and every request is recorded
|
|
(method, path, query, JSON body, Authorization header) so tests can assert
|
|
exactly what the client sends — methods, paths, request bodies, pagination,
|
|
normalized results, and representative failure responses. The live Gitea
|
|
instance is never contacted.
|
|
"""
|
|
|
|
import inspect
|
|
import json
|
|
import re
|
|
import sys
|
|
import threading
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from urllib.parse import urlsplit
|
|
|
|
import pytest
|
|
|
|
# PEP 420 namespace package: make the project root importable regardless of
|
|
# how pytest was invoked.
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from gitea_saas.gitea_client import ( # noqa: E402
|
|
GiteaApiError,
|
|
GiteaClient,
|
|
GiteaClientError,
|
|
_PER_PAGE,
|
|
)
|
|
|
|
TEST_TOKEN = "test-token-123-abc-secret"
|
|
TEST_OWNER = "pbwraith"
|
|
|
|
|
|
def gitea_pr(number, head_sha="sha-head", base_ref="main",
|
|
labels=("in-review",), body="Closes #1\nSome PR"):
|
|
"""A raw Gitea PR record as the list/get endpoints return it."""
|
|
return {
|
|
"number": number,
|
|
"title": f"PR {number}",
|
|
"state": "open",
|
|
"body": body,
|
|
"head": {"ref": f"issue-{number}", "sha": head_sha},
|
|
"base": {"ref": base_ref, "sha": "sha-base"},
|
|
"labels": [{"id": i, "name": name} for i, name in enumerate(labels, 1)],
|
|
}
|
|
|
|
|
|
class StubServer:
|
|
"""Minimal scripted Gitea HTTP stub (threaded, localhost only)."""
|
|
|
|
def __init__(self, dispatcher):
|
|
self.requests = [] # [{method, path, query, body, auth}]
|
|
self.dispatcher = dispatcher # (method, path, query, body) -> (status, payload|None)
|
|
self.httpd = ThreadingHTTPServer(("127.0.0.1", 0), self._handler())
|
|
self.port = self.httpd.server_address[1]
|
|
self.thread = threading.Thread(target=self.httpd.serve_forever, daemon=True)
|
|
self.thread.start()
|
|
|
|
def _handler(self):
|
|
server = self
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def log_message(self, format, *args): # noqa: A002 - keep test output clean
|
|
pass
|
|
|
|
def _handle(self, method):
|
|
parsed = urlsplit(self.path)
|
|
length = int(self.headers.get("Content-Length") or 0)
|
|
raw_body = self.rfile.read(length) if length else b""
|
|
record = {
|
|
"method": method,
|
|
"path": parsed.path,
|
|
"query": parsed.query,
|
|
"body": raw_body.decode("utf-8", "replace") if raw_body else None,
|
|
"auth": self.headers.get("Authorization"),
|
|
}
|
|
server.requests.append(record)
|
|
status, payload = server.dispatcher(method, parsed.path,
|
|
parsed.query, record["body"])
|
|
body = b""
|
|
if payload is not None:
|
|
body = payload if isinstance(payload, bytes) else json.dumps(payload).encode()
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
if body:
|
|
self.wfile.write(body)
|
|
|
|
def do_GET(self):
|
|
self._handle("GET")
|
|
|
|
def do_POST(self):
|
|
self._handle("POST")
|
|
|
|
def do_PATCH(self):
|
|
self._handle("PATCH")
|
|
|
|
def do_DELETE(self):
|
|
self._handle("DELETE")
|
|
|
|
return Handler
|
|
|
|
def close(self):
|
|
self.httpd.shutdown()
|
|
self.httpd.server_close()
|
|
|
|
|
|
@pytest.fixture
|
|
def make_client():
|
|
"""Build a GiteaClient wired to a fresh scripted stub; teardown shuts it down."""
|
|
stubs = []
|
|
|
|
def build(dispatcher, token=TEST_TOKEN, owner=TEST_OWNER):
|
|
stub = StubServer(dispatcher)
|
|
stubs.append(stub)
|
|
client = GiteaClient(base_url=f"http://127.0.0.1:{stub.port}",
|
|
token=token, owner=owner)
|
|
return client, stub
|
|
|
|
yield build
|
|
for stub in stubs:
|
|
stub.close()
|
|
|
|
|
|
def requests_for(stub, method=None, path_suffix=None):
|
|
out = []
|
|
for r in stub.requests:
|
|
if method is not None and r["method"] != method:
|
|
continue
|
|
if path_suffix is not None and not r["path"].endswith(path_suffix):
|
|
continue
|
|
out.append(r)
|
|
return out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# list_open_prs: pagination + normalization
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_list_open_prs_follows_pagination_and_normalizes(make_client):
|
|
full_page = [gitea_pr(n, labels=("in-review", "p1") if n == 1 else ("in-review",))
|
|
for n in range(1, _PER_PAGE + 1)]
|
|
|
|
def dispatcher(method, path, query, body):
|
|
if path.endswith("/api/v1/repos/pbwraith/acme/pulls") and query.startswith("state=open&page=1"):
|
|
return 200, full_page
|
|
if path.endswith("/api/v1/repos/pbwraith/acme/pulls") and query.startswith("state=open&page=2"):
|
|
return 200, [gitea_pr(_PER_PAGE + 1)]
|
|
return 404, {"error": "unexpected route"}
|
|
|
|
client, stub = make_client(dispatcher)
|
|
result = client.list_open_prs("acme")
|
|
|
|
assert [p["number"] for p in result] == list(range(1, _PER_PAGE + 2))
|
|
assert result[0]["labels"] == ["in-review", "p1"]
|
|
for pr in result:
|
|
assert pr["head_sha"] == "sha-head"
|
|
assert pr["base_ref"] == "main"
|
|
assert pr["repo"] == "acme"
|
|
assert pr["body"] == "Closes #1\nSome PR"
|
|
assert isinstance(pr["labels"], list)
|
|
|
|
# Only open PRs were requested, at limit=50, and pagination stopped at
|
|
# the first short page (page=2 returned 1 < 50 items -> no page=3).
|
|
pulls = requests_for(stub, method="GET", path_suffix="/pulls")
|
|
assert pulls, "expected GET /pulls requests"
|
|
assert all("state=open" in r["query"] for r in pulls)
|
|
assert all("limit=50" in r["query"] for r in pulls)
|
|
assert any("page=1" in r["query"] for r in pulls)
|
|
assert any("page=2" in r["query"] for r in pulls)
|
|
assert not any("page=3" in r["query"] for r in pulls)
|
|
|
|
|
|
def test_list_open_prs_empty_repo(make_client):
|
|
def dispatcher(method, path, query, body):
|
|
return 200, []
|
|
|
|
client, _ = make_client(dispatcher)
|
|
assert client.list_open_prs("empty-repo") == []
|
|
|
|
|
|
def test_pagination_that_never_terminates_raises(make_client):
|
|
def dispatcher(method, path, query, body):
|
|
# A full page on every request: without the defensive page cap the
|
|
# client would loop forever.
|
|
return 200, [gitea_pr(n) for n in range(1, _PER_PAGE + 1)]
|
|
|
|
client, _ = make_client(dispatcher)
|
|
with pytest.raises(GiteaClientError, match="pagination did not terminate"):
|
|
client.list_open_prs("pathological")
|
|
|
|
|
|
def test_get_pr_normalized_shape_with_changed_files(make_client):
|
|
def dispatcher(method, path, query, body):
|
|
if method == "GET" and path.endswith("/pulls/4"):
|
|
return 200, gitea_pr(4, labels=("in-review", "p1"))
|
|
if method == "GET" and path.endswith("/pulls/4/files"):
|
|
return 200, [{"filename": "gitea_saas/gitea_client.py"},
|
|
{"filename": "tests/test_gitea_client.py"}]
|
|
return 404, {"error": "unexpected route"}
|
|
|
|
client, _ = make_client(dispatcher)
|
|
pr = client.get_pr("acme", 4)
|
|
|
|
assert pr["number"] == 4
|
|
assert pr["body"] == "Closes #1\nSome PR"
|
|
assert pr["head_sha"] == "sha-head"
|
|
assert pr["base_sha"] == "sha-base"
|
|
assert pr["base_ref"] == "main"
|
|
assert pr["labels"] == ["in-review", "p1"]
|
|
assert pr["repo"] == "acme"
|
|
assert pr["changed_files"] == ["gitea_saas/gitea_client.py",
|
|
"tests/test_gitea_client.py"]
|
|
|
|
|
|
def test_list_open_prs_authorization_header_sent(make_client):
|
|
def dispatcher(method, path, query, body):
|
|
return 200, []
|
|
|
|
client, stub = make_client(dispatcher)
|
|
client.list_open_prs("acme")
|
|
assert requests_for(stub, method="GET", path_suffix="/pulls")[0]["auth"] == \
|
|
f"token {TEST_TOKEN}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# add_label / remove_label
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_add_label_posts_label_id_and_verifies(make_client):
|
|
calls = {"issue_gets": 0}
|
|
|
|
def dispatcher(method, path, query, body):
|
|
if method == "GET" and path.endswith("/labels"):
|
|
return 200, [{"id": 7, "name": "bounced"}]
|
|
if method == "GET" and path.endswith("/issues/5"):
|
|
calls["issue_gets"] += 1
|
|
if calls["issue_gets"] == 1: # before
|
|
return 200, {"number": 5, "labels": [{"id": 8, "name": "in-review"}]}
|
|
return 200, {"number": 5, "labels": [{"id": 8, "name": "in-review"},
|
|
{"id": 7, "name": "bounced"}]}
|
|
if method == "POST" and path.endswith("/issues/5/labels"):
|
|
return 200, [{"id": 8, "name": "in-review"}, {"id": 7, "name": "bounced"}]
|
|
return 404, {"error": "unexpected route"}
|
|
|
|
client, stub = make_client(dispatcher)
|
|
client.add_label("acme", 5, "bounced")
|
|
|
|
posts = requests_for(stub, method="POST", path_suffix="/issues/5/labels")
|
|
assert len(posts) == 1
|
|
assert json.loads(posts[0]["body"]) == {"labels": [7]} # label ID, not name
|
|
assert calls["issue_gets"] == 2 # verification re-GET happened
|
|
|
|
|
|
def test_add_label_already_present_is_noop(make_client):
|
|
def dispatcher(method, path, query, body):
|
|
if method == "GET" and path.endswith("/labels"):
|
|
return 200, [{"id": 7, "name": "bounced"}]
|
|
if method == "GET" and path.endswith("/issues/5"):
|
|
return 200, {"number": 5, "labels": [{"id": 7, "name": "bounced"}]}
|
|
return 404, {"error": "unexpected route"}
|
|
|
|
client, stub = make_client(dispatcher)
|
|
client.add_label("acme", 5, "bounced")
|
|
assert not requests_for(stub, method="POST", path_suffix="/labels")
|
|
|
|
|
|
def test_add_label_unknown_label_raises_and_skips_post(make_client):
|
|
def dispatcher(method, path, query, body):
|
|
if method == "GET" and path.endswith("/labels"):
|
|
return 200, [{"id": 7, "name": "bounced"}]
|
|
return 404, {"error": "unexpected route"}
|
|
|
|
client, stub = make_client(dispatcher)
|
|
with pytest.raises(GiteaClientError, match="unknown label 'nope'"):
|
|
client.add_label("acme", 5, "nope")
|
|
assert not requests_for(stub, method="POST", path_suffix="/labels")
|
|
|
|
|
|
def test_add_label_verification_failure_raises(make_client):
|
|
def dispatcher(method, path, query, body):
|
|
if method == "GET" and path.endswith("/labels"):
|
|
return 200, [{"id": 7, "name": "bounced"}]
|
|
if method == "GET" and path.endswith("/issues/5"):
|
|
return 200, {"number": 5, "labels": []} # POST is never reflected
|
|
if method == "POST" and path.endswith("/issues/5/labels"):
|
|
return 200, [{"id": 7, "name": "bounced"}]
|
|
return 404, {"error": "unexpected route"}
|
|
|
|
client, stub = make_client(dispatcher)
|
|
with pytest.raises(GiteaClientError, match="missing on issue 5 after POST"):
|
|
client.add_label("acme", 5, "bounced")
|
|
|
|
|
|
def test_remove_label_deletes_label_id_and_verifies(make_client):
|
|
calls = {"issue_gets": 0}
|
|
|
|
def dispatcher(method, path, query, body):
|
|
if method == "GET" and path.endswith("/labels"):
|
|
return 200, [{"id": 7, "name": "bounced"}, {"id": 8, "name": "in-review"}]
|
|
if method == "GET" and path.endswith("/issues/5"):
|
|
calls["issue_gets"] += 1
|
|
if calls["issue_gets"] == 1: # before
|
|
return 200, {"number": 5, "labels": [{"id": 7, "name": "bounced"},
|
|
{"id": 8, "name": "in-review"}]}
|
|
return 200, {"number": 5, "labels": [{"id": 8, "name": "in-review"}]}
|
|
if method == "DELETE" and path.endswith("/issues/5/labels/7"):
|
|
return 204, None
|
|
return 404, {"error": "unexpected route"}
|
|
|
|
client, stub = make_client(dispatcher)
|
|
client.remove_label("acme", 5, "bounced")
|
|
|
|
deletes = requests_for(stub, method="DELETE", path_suffix="/issues/5/labels/7")
|
|
assert len(deletes) == 1 # resolved the label name to id 7
|
|
assert calls["issue_gets"] == 2 # verification re-GET happened
|
|
|
|
|
|
def test_remove_label_already_absent_is_noop(make_client):
|
|
def dispatcher(method, path, query, body):
|
|
if method == "GET" and path.endswith("/labels"):
|
|
return 200, [{"id": 7, "name": "bounced"}]
|
|
if method == "GET" and path.endswith("/issues/5"):
|
|
return 200, {"number": 5, "labels": []}
|
|
return 404, {"error": "unexpected route"}
|
|
|
|
client, stub = make_client(dispatcher)
|
|
client.remove_label("acme", 5, "bounced")
|
|
assert not [r for r in stub.requests if r["method"] == "DELETE"]
|
|
|
|
|
|
def test_remove_label_unknown_label_raises(make_client):
|
|
def dispatcher(method, path, query, body):
|
|
if method == "GET" and path.endswith("/labels"):
|
|
return 200, [{"id": 7, "name": "bounced"}]
|
|
return 404, {"error": "unexpected route"}
|
|
|
|
client, stub = make_client(dispatcher)
|
|
with pytest.raises(GiteaClientError, match="unknown label 'nope'"):
|
|
client.remove_label("acme", 5, "nope")
|
|
assert not [r for r in stub.requests if r["method"] == "DELETE"]
|
|
|
|
|
|
def test_remove_label_verification_failure_raises(make_client):
|
|
def dispatcher(method, path, query, body):
|
|
if method == "GET" and path.endswith("/labels"):
|
|
return 200, [{"id": 7, "name": "bounced"}]
|
|
if method == "GET" and path.endswith("/issues/5"):
|
|
return 200, {"number": 5, "labels": [{"id": 7, "name": "bounced"}]}
|
|
if method == "DELETE" and path.endswith("/issues/5/labels/7"):
|
|
return 204, None # DELETE succeeds but the label stays
|
|
return 404, {"error": "unexpected route"}
|
|
|
|
client, stub = make_client(dispatcher)
|
|
with pytest.raises(GiteaClientError, match="still on issue 5 after DELETE"):
|
|
client.remove_label("acme", 5, "bounced")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# merge_pr
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_merge_pr_posts_merge_and_confirms(make_client):
|
|
state = {"gets": 0}
|
|
|
|
def dispatcher(method, path, query, body):
|
|
if method == "GET" and path.endswith("/pulls/7"):
|
|
state["gets"] += 1
|
|
merged = state["gets"] >= 2 # first GET = pre-check, later = confirm
|
|
return 200, {**gitea_pr(7, head_sha="sha-head"), "merged": merged}
|
|
if method == "POST" and path.endswith("/pulls/7/merge"):
|
|
return 200, {"merged": True, "message": "Pull Request was successfully merged"}
|
|
return 404, {"error": "unexpected route"}
|
|
|
|
client, stub = make_client(dispatcher)
|
|
assert client.merge_pr("acme", 7) is True
|
|
|
|
posts = requests_for(stub, method="POST", path_suffix="/pulls/7/merge")
|
|
assert len(posts) == 1
|
|
# The real request must carry a supported merge mode plus the reviewed
|
|
# head commit id — the Gitea schema rejects `{}` and an unpinned merge.
|
|
assert json.loads(posts[0]["body"]) == {"Do": "merge", "head_commit_id": "sha-head"}
|
|
|
|
|
|
def test_merge_pr_sends_reviewed_head_commit_id_regression(make_client):
|
|
# Regression (review bounce): the merge payload must pin the exact head
|
|
# commit id read from the PR, so a branch that moved after review cannot
|
|
# be merged accidentally; `Do` must be a schema-supported merge mode.
|
|
reviewed_sha = "abcd-1234-reviewed-head"
|
|
state = {"gets": 0}
|
|
|
|
def dispatcher(method, path, query, body):
|
|
if method == "GET" and path.endswith("/pulls/7"):
|
|
state["gets"] += 1
|
|
merged = state["gets"] >= 2 # first GET = pre-check, later = confirm
|
|
return 200, {**gitea_pr(7, head_sha=reviewed_sha), "merged": merged}
|
|
if method == "POST" and path.endswith("/pulls/7/merge"):
|
|
return 200, {"merged": True, "message": "merged"}
|
|
return 404, {"error": "unexpected route"}
|
|
|
|
client, stub = make_client(dispatcher)
|
|
assert client.merge_pr("acme", 7) is True
|
|
|
|
posts = requests_for(stub, method="POST", path_suffix="/pulls/7/merge")
|
|
assert len(posts) == 1
|
|
payload = json.loads(posts[0]["body"])
|
|
assert payload == {"Do": "merge", "head_commit_id": reviewed_sha}
|
|
assert payload["Do"] in {"merge", "rebase", "rebase-merge", "squash",
|
|
"fast-forward-only"}
|
|
|
|
|
|
def test_merge_pr_raises_when_pr_has_no_head_commit_id(make_client):
|
|
# Without a head commit id the merge cannot be pinned to the reviewed
|
|
# head: fail closed instead of sending an unpinned (or empty) request.
|
|
def dispatcher(method, path, query, body):
|
|
if method == "GET" and path.endswith("/pulls/7"):
|
|
pr = gitea_pr(7)
|
|
pr["head"] = {"ref": "issue-7"} # sha missing
|
|
return 200, pr
|
|
return 404, {"error": "unexpected route"}
|
|
|
|
client, stub = make_client(dispatcher)
|
|
with pytest.raises(GiteaClientError, match="no head commit id"):
|
|
client.merge_pr("acme", 7)
|
|
assert not requests_for(stub, method="POST", path_suffix="/merge")
|
|
|
|
|
|
def test_merge_pr_refuses_moved_head_before_post(make_client):
|
|
# TOCTOU regression (governor review): the branch moved from the
|
|
# reviewed SHA A to SHA B between review and merge. Passing the
|
|
# reviewed SHA must fail closed BEFORE any merge POST — no unstably
|
|
# merged, unreviewed SHA B.
|
|
reviewed_sha = "reviewed-sha-AAAA"
|
|
moved_sha = "moved-sha-BBBB"
|
|
state = {"gets": 0}
|
|
|
|
def dispatcher(method, path, query, body):
|
|
if method == "GET" and path.endswith("/pulls/7"):
|
|
state["gets"] += 1
|
|
merged = state["gets"] >= 2 # first GET = pre-check, later = confirm
|
|
return 200, {**gitea_pr(7, head_sha=moved_sha), "merged": merged}
|
|
return 404, {"error": "unexpected route"}
|
|
|
|
client, stub = make_client(dispatcher)
|
|
with pytest.raises(GiteaClientError, match="moved after review"):
|
|
client.merge_pr("acme", 7, expected_head_sha=reviewed_sha)
|
|
|
|
# Fail closed: the merge endpoint was never called, so unreviewed
|
|
# SHA B cannot be merged.
|
|
assert not requests_for(stub, method="POST", path_suffix="/merge")
|
|
assert "reviewed-sha-AAAA" not in [
|
|
json.loads(r["body"]).get("head_commit_id")
|
|
for r in stub.requests if r["method"] == "POST"]
|
|
|
|
|
|
def test_merge_pr_pins_expected_head_in_payload(make_client):
|
|
# The reviewed SHA passed by the caller is what gets sent as
|
|
# head_commit_id (not whatever the pre-merge GET happens to report).
|
|
reviewed_sha = "reviewed-sha-CCCC"
|
|
state = {"gets": 0}
|
|
|
|
def dispatcher(method, path, query, body):
|
|
if method == "GET" and path.endswith("/pulls/7"):
|
|
state["gets"] += 1
|
|
merged = state["gets"] >= 2 # first GET = pre-check, later = confirm
|
|
return 200, {**gitea_pr(7, head_sha=reviewed_sha), "merged": merged}
|
|
if method == "POST" and path.endswith("/pulls/7/merge"):
|
|
return 200, {"merged": True, "message": "merged"}
|
|
return 404, {"error": "unexpected route"}
|
|
|
|
client, stub = make_client(dispatcher)
|
|
assert client.merge_pr("acme", 7, expected_head_sha=reviewed_sha) is True
|
|
|
|
posts = requests_for(stub, method="POST", path_suffix="/pulls/7/merge")
|
|
assert len(posts) == 1
|
|
assert json.loads(posts[0]["body"]) == {
|
|
"Do": "merge", "head_commit_id": reviewed_sha}
|
|
|
|
|
|
def test_merge_pr_already_merged_short_circuits(make_client):
|
|
def dispatcher(method, path, query, body):
|
|
if method == "GET" and path.endswith("/pulls/7"):
|
|
return 200, {**gitea_pr(7), "merged": True}
|
|
return 404, {"error": "unexpected route"}
|
|
|
|
client, stub = make_client(dispatcher)
|
|
assert client.merge_pr("acme", 7) is True
|
|
assert not requests_for(stub, method="POST", path_suffix="/merge")
|
|
|
|
|
|
def test_merge_pr_returns_false_when_post_condition_unmet(make_client):
|
|
def dispatcher(method, path, query, body):
|
|
if method == "GET" and path.endswith("/pulls/7"):
|
|
return 200, {**gitea_pr(7), "merged": False}
|
|
if method == "POST" and path.endswith("/pulls/7/merge"):
|
|
return 200, {"merged": False, "message": "not merged yet"}
|
|
return 404, {"error": "unexpected route"}
|
|
|
|
client, stub = make_client(dispatcher)
|
|
assert client.merge_pr("acme", 7) is False
|
|
assert len(requests_for(stub, method="POST", path_suffix="/merge")) == 1
|
|
|
|
|
|
def test_merge_pr_conflict_raises(make_client):
|
|
def dispatcher(method, path, query, body):
|
|
if method == "GET" and path.endswith("/pulls/7"):
|
|
return 200, {**gitea_pr(7), "merged": False}
|
|
if method == "POST" and path.endswith("/pulls/7/merge"):
|
|
return 409, {"message": "conflict: merge is blocked"}
|
|
return 404, {"error": "unexpected route"}
|
|
|
|
client, stub = make_client(dispatcher)
|
|
with pytest.raises(GiteaApiError) as excinfo:
|
|
client.merge_pr("acme", 7)
|
|
assert excinfo.value.status == 409
|
|
assert "merge is blocked" in str(excinfo.value)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# close_issue
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_close_issue_patches_state_and_confirms(make_client):
|
|
def dispatcher(method, path, query, body):
|
|
if method == "GET" and path.endswith("/issues/9"):
|
|
return 200, {"number": 9, "state": "closed"}
|
|
if method == "PATCH" and path.endswith("/issues/9"):
|
|
return 200, {"number": 9, "state": "closed"}
|
|
return 404, {"error": "unexpected route"}
|
|
|
|
client, stub = make_client(dispatcher)
|
|
client.close_issue("acme", 9)
|
|
|
|
patches = requests_for(stub, method="PATCH", path_suffix="/issues/9")
|
|
assert len(patches) == 1
|
|
assert json.loads(patches[0]["body"]) == {"state": "closed"}
|
|
|
|
|
|
def test_close_issue_verification_failure_raises(make_client):
|
|
def dispatcher(method, path, query, body):
|
|
if method == "GET" and path.endswith("/issues/9"):
|
|
return 200, {"number": 9, "state": "open"} # never actually closed
|
|
if method == "PATCH" and path.endswith("/issues/9"):
|
|
return 200, {"number": 9, "state": "closed"}
|
|
return 404, {"error": "unexpected route"}
|
|
|
|
client, stub = make_client(dispatcher)
|
|
with pytest.raises(GiteaClientError, match="state is 'open' after close request"):
|
|
client.close_issue("acme", 9)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Non-2xx handling: useful error, token never leaks
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_non_2xx_raises_useful_error_without_leaking_token(make_client):
|
|
def dispatcher(method, path, query, body):
|
|
# Simulate a hostile server that echoes the request back, including
|
|
# the Authorization header — the client must redact the token.
|
|
return 500, {"error": "boom", "echo": f"Authorization: token {TEST_TOKEN}"}
|
|
|
|
client, _ = make_client(dispatcher)
|
|
with pytest.raises(GiteaApiError) as excinfo:
|
|
client.list_labels("acme")
|
|
|
|
message = str(excinfo.value)
|
|
assert excinfo.value.status == 500
|
|
assert excinfo.value.method == "GET"
|
|
assert "boom" in message
|
|
assert TEST_TOKEN not in message
|
|
assert "Authorization" not in message
|
|
assert "token" not in message.lower() or TEST_TOKEN not in message
|
|
|
|
|
|
def test_non_2xx_message_carries_path_and_status(make_client):
|
|
def dispatcher(method, path, query, body):
|
|
return 422, {"message": "invalid pull request"}
|
|
|
|
client, _ = make_client(dispatcher)
|
|
with pytest.raises(GiteaApiError) as excinfo:
|
|
client.get_pr("acme", 3)
|
|
assert "HTTP 422" in str(excinfo.value)
|
|
assert "pulls/3" in str(excinfo.value)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Contract: the production client exposes every operation ReviewPolicy needs
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_production_client_exposes_review_policy_contract():
|
|
"""Issue #28 acceptance: the client exposes every seam the policy calls."""
|
|
from gitea_saas.review_policy import ReviewPolicy
|
|
from gitea_saas.reviewer import Reviewer
|
|
|
|
def gitea_ops(cls):
|
|
# ``self.gitea.<op>(`` in ReviewPolicy / ``self.repo_client.<op>(``
|
|
# in Reviewer — every attribute access on the seam, derived from the
|
|
# source so the contract stays honest as the policy evolves.
|
|
source = inspect.getsource(cls)
|
|
prefix = "self.gitea." if cls is ReviewPolicy else "self.repo_client."
|
|
ops = set(re.findall(re.escape(prefix) + r"(\w+)\(", source))
|
|
# ``merge_pr`` is also invoked through the shared debit helper; keep
|
|
# the set stable by including explicitly required operations.
|
|
return ops
|
|
|
|
required = gitea_ops(ReviewPolicy) | gitea_ops(Reviewer)
|
|
# Explicit core of the issue's scope (ReviewPolicy's seam list):
|
|
required |= {"list_open_prs", "get_pr", "get_issue", "post_comment",
|
|
"add_label", "remove_label", "merge_pr", "close_issue"}
|
|
|
|
missing = sorted(op for op in required if not callable(getattr(GiteaClient, op, None)))
|
|
assert not missing, f"production client is missing ReviewPolicy operations: {missing}"
|