Gitea client: production review-policy client operations (#28) #31
@ -1,20 +1,62 @@
|
||||
"""Gitea API client (mockable seam, env token).
|
||||
|
||||
Shared by the factory front-end (issue #4) and the review gate (issue #5).
|
||||
Real HTTP only when a token is available; the token is read from
|
||||
``GITEA_TOKEN`` (or passed in) and is never printed or placed on a command
|
||||
line. Tests use an in-memory fake, so this module is never exercised over
|
||||
the network in CI.
|
||||
Shared by the factory front-end (issue #4), the review gate (issue #5), and
|
||||
the production review-policy client operations (issue #28: listing open PRs,
|
||||
label mutations, merge, issue close). Real HTTP only when a token is
|
||||
available; the token is read from ``GITEA_TOKEN`` (or passed in) and is
|
||||
never printed, placed on a command line, or echoed in errors. Tests use a
|
||||
local HTTP stub or an in-memory fake, so this module is never exercised
|
||||
against the live instance in CI.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from typing import Any, Optional
|
||||
|
||||
# Gitea list endpoints cap at 50 items per page (the API's ``limit`` bound).
|
||||
_PER_PAGE = 50
|
||||
# Safety valve so a misbehaving server that never returns a short page
|
||||
# cannot make ``_paginate`` loop forever.
|
||||
_MAX_PAGES = 100
|
||||
|
||||
|
||||
class GiteaClientError(Exception):
|
||||
"""Base error for the Gitea client.
|
||||
|
||||
Messages NEVER include the token or the ``Authorization`` header (the
|
||||
client redacts the token from any response text it copies into the
|
||||
message).
|
||||
"""
|
||||
|
||||
|
||||
class GiteaApiError(GiteaClientError):
|
||||
"""A non-2xx Gitea API response.
|
||||
|
||||
``message`` carries the method, API path, HTTP status, and a truncated
|
||||
response body — deliberately no request headers, so the authorization
|
||||
header can never leak.
|
||||
"""
|
||||
|
||||
def __init__(self, method: str, api_path: str, status: int, detail: str = ""):
|
||||
self.method = method
|
||||
self.api_path = api_path
|
||||
self.status = status
|
||||
self.detail = detail
|
||||
message = f"Gitea API {method} {api_path} returned HTTP {status}"
|
||||
if detail:
|
||||
message += f": {_truncate(detail, 300)}"
|
||||
super().__init__(message)
|
||||
|
||||
|
||||
def _truncate(text: str, limit: int = 300) -> str:
|
||||
text = (text or "").strip()
|
||||
return text if len(text) <= limit else text[:limit] + "…"
|
||||
|
||||
|
||||
class GiteaClient:
|
||||
"""Thin urllib client for the subset of the Gitea API the SaaS needs.
|
||||
@ -24,6 +66,11 @@ class GiteaClient:
|
||||
build ``/api/v1/repos/{owner}/{repo}/...``; org-repo creation uses
|
||||
``/api/v1/orgs/{owner}/repos`` and falls back to the authenticated
|
||||
user's repos when ``owner`` is a user account (orgs endpoint 404s).
|
||||
|
||||
Every non-2xx response raises :class:`GiteaApiError` (never silently
|
||||
ignored); the review-policy protocol (issue #28) is fail-closed: an
|
||||
unknown label or a failed post-condition verification raises
|
||||
:class:`GiteaClientError` instead of skipping the operation.
|
||||
"""
|
||||
|
||||
def __init__(self, base_url: Optional[str] = None, token: Optional[str] = None,
|
||||
@ -36,7 +83,12 @@ class GiteaClient:
|
||||
# -- low-level ---------------------------------------------------------
|
||||
|
||||
def _api(self, method: str, api_path: str, body: Optional[dict] = None) -> Any:
|
||||
"""Perform an API call under ``/api/v1/{api_path}`` and parse JSON."""
|
||||
"""Perform an API call under ``/api/v1/{api_path}`` and parse JSON.
|
||||
|
||||
Non-2xx responses raise :class:`GiteaApiError` with the method, API
|
||||
path, status, and a redacted response snippet — the authorization
|
||||
header (and therefore the token) never appears in the message.
|
||||
"""
|
||||
data = json.dumps(body).encode() if body is not None else None
|
||||
url = f"{self.base_url}/api/v1/{api_path}"
|
||||
req = urllib.request.Request(
|
||||
@ -44,14 +96,79 @@ class GiteaClient:
|
||||
headers={"Authorization": "token " + self.token,
|
||||
"Accept": "application/json",
|
||||
"Content-Type": "application/json"})
|
||||
with urllib.request.urlopen(req, timeout=30) as resp:
|
||||
raw = resp.read()
|
||||
return json.loads(raw) if raw else None
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=30) as resp:
|
||||
raw = resp.read()
|
||||
return json.loads(raw) if raw else None
|
||||
except urllib.error.HTTPError as exc:
|
||||
detail = ""
|
||||
try:
|
||||
raw_body = exc.read()
|
||||
except Exception: # noqa: BLE001 - the body may be unreadable; fail cleanly
|
||||
raw_body = b""
|
||||
if raw_body:
|
||||
text = raw_body.decode("utf-8", "replace")
|
||||
# Defensive: some servers echo the request (including its
|
||||
# headers) back in the error body — redact the token value
|
||||
# and any echoed Authorization header regardless.
|
||||
if self.token:
|
||||
text = text.replace(self.token, "***")
|
||||
# Scrub any echoed Authorization header completely (name and
|
||||
# value), not only the token — the header itself must not
|
||||
# appear in errors.
|
||||
text = re.sub(
|
||||
r"(?i)authorization\s*[:=]\s*(?:token\s+)?[^\s,}\]]+",
|
||||
"***", text)
|
||||
detail = _truncate(text, 300)
|
||||
raise GiteaApiError(method, api_path, exc.code, detail) from exc
|
||||
|
||||
def _request(self, method: str, path: str, body: Optional[dict] = None) -> Any:
|
||||
"""Repo-scoped call: ``path`` = ``<repo>/...`` under the owner."""
|
||||
return self._api(method, f"repos/{self.owner}/{path}", body)
|
||||
|
||||
def _paginate(self, api_path: str) -> list:
|
||||
"""Fetch every page of a Gitea list endpoint (``GET``).
|
||||
|
||||
Follows Gitea pagination (``page``/``limit`` query parameters) until
|
||||
a page returns fewer than ``_PER_PAGE`` items; a defensive page cap
|
||||
raises instead of looping forever on a pathologically misbehaving
|
||||
server.
|
||||
"""
|
||||
collected: list = []
|
||||
for page in range(1, _MAX_PAGES + 1):
|
||||
sep = "&" if "?" in api_path else "?"
|
||||
batch = self._api("GET", f"{api_path}{sep}page={page}&limit={_PER_PAGE}")
|
||||
if not isinstance(batch, list):
|
||||
raise GiteaClientError(
|
||||
f"expected a JSON list from {api_path!r}, got {type(batch).__name__}")
|
||||
collected.extend(batch)
|
||||
if len(batch) < _PER_PAGE:
|
||||
break
|
||||
else:
|
||||
raise GiteaClientError(
|
||||
f"pagination did not terminate after {_MAX_PAGES} pages of {api_path!r}")
|
||||
return collected
|
||||
|
||||
def _normalize_pr(self, pr: dict, repo: str = "") -> dict:
|
||||
"""Map a raw Gitea PR record to the candidate shape.
|
||||
|
||||
Every candidate supplies at least ``number``, ``body``, ``head_sha``,
|
||||
``base_ref``, ``labels`` (label NAMES), and ``repo`` — the shape
|
||||
``ReviewPolicy`` and the review gate consume.
|
||||
"""
|
||||
head = pr.get("head") or {}
|
||||
base = pr.get("base") or {}
|
||||
labels = [label.get("name") for label in (pr.get("labels") or [])
|
||||
if isinstance(label, dict) and label.get("name")]
|
||||
return {
|
||||
"number": pr.get("number"),
|
||||
"body": pr.get("body") or "",
|
||||
"head_sha": head.get("sha"),
|
||||
"base_ref": base.get("ref"),
|
||||
"labels": labels,
|
||||
"repo": repo,
|
||||
}
|
||||
|
||||
# -- API surface used by the factory front-end (issue #4) --------------
|
||||
|
||||
def repo_exists(self, name: str) -> bool:
|
||||
@ -62,8 +179,8 @@ class GiteaClient:
|
||||
"""
|
||||
try:
|
||||
self._request("GET", f"{name}")
|
||||
except urllib.error.HTTPError as exc:
|
||||
if exc.code == 404:
|
||||
except GiteaApiError as exc:
|
||||
if exc.status == 404:
|
||||
return False
|
||||
raise
|
||||
return True
|
||||
@ -86,8 +203,8 @@ class GiteaClient:
|
||||
payload: dict = {"name": name, "auto_init": False, "private": True}
|
||||
try:
|
||||
return self._api("POST", f"orgs/{self.owner}/repos", payload)
|
||||
except urllib.error.HTTPError as exc:
|
||||
if exc.code not in (404, 422):
|
||||
except GiteaApiError as exc:
|
||||
if exc.status not in (404, 422):
|
||||
raise
|
||||
# ``owner`` is a user account (or lacks org-create rights): fall back
|
||||
# to creating the repo under the authenticated token user.
|
||||
@ -138,13 +255,17 @@ class GiteaClient:
|
||||
return self._request("GET", f"{repo}/issues/{number}") or {}
|
||||
|
||||
def get_pr(self, repo: str, number: int) -> dict:
|
||||
"""Fetch a PR: {head_sha, changed_files: [filename, ...]}."""
|
||||
"""Fetch a PR in the normalized candidate shape.
|
||||
|
||||
``{number, body, head_sha, base_ref, labels: [names], repo,
|
||||
changed_files: [filename, ...]}`` — the review gate consumes
|
||||
``head_sha`` and ``changed_files``; policy consumes the rest.
|
||||
"""
|
||||
pr = self._request("GET", f"{repo}/pulls/{number}") or {}
|
||||
files = self._request("GET", f"{repo}/pulls/{number}/files") or []
|
||||
return {
|
||||
"head_sha": ((pr.get("head") or {}).get("sha")),
|
||||
"changed_files": [f.get("filename") for f in files if f.get("filename")],
|
||||
}
|
||||
normalized = self._normalize_pr(pr, repo)
|
||||
normalized["changed_files"] = [f.get("filename") for f in files if f.get("filename")]
|
||||
return normalized
|
||||
|
||||
def get_pr_diff(self, repo: str, number: int) -> str:
|
||||
"""Return the PR's diff text reconstructed from per-file patches."""
|
||||
@ -160,3 +281,129 @@ class GiteaClient:
|
||||
def post_comment(self, repo: str, number: int, body: str) -> dict:
|
||||
"""Post a comment on an issue or PR conversation."""
|
||||
return self._request("POST", f"{repo}/issues/{number}/comments", {"body": body}) or {}
|
||||
|
||||
# -- API surface used by the review policy (issue #28) ------------------
|
||||
#
|
||||
# Every operation in this section is fail-closed: a non-2xx response
|
||||
# raises GiteaApiError, an unknown label or a failed post-condition
|
||||
# verification raises GiteaClientError. Nothing is silently skipped.
|
||||
|
||||
def list_open_prs(self, repo: str) -> list:
|
||||
"""All open PRs in ``owner/repo``, normalized, pagination followed.
|
||||
|
||||
Requests ``?state=open`` (all and only open PRs) page by page and
|
||||
returns the normalized candidate shape (see ``_normalize_pr``).
|
||||
"""
|
||||
prs = self._paginate(f"repos/{self.owner}/{repo}/pulls?state=open")
|
||||
return [self._normalize_pr(pr, repo) for pr in prs]
|
||||
|
||||
def add_label(self, repo: str, issue_number: int, label: str) -> None:
|
||||
"""Add ``label`` to ``owner/repo`` issue ``issue_number``.
|
||||
|
||||
Resolves the repository label id, POSTs it (``{"labels": [id]}``),
|
||||
and verifies the resulting label state. Already-present labels are
|
||||
a no-op. An unknown label or a failed verification raises a clear
|
||||
:class:`GiteaClientError` — never silently skipped.
|
||||
"""
|
||||
name = (label or "").strip()
|
||||
if not name:
|
||||
raise GiteaClientError("add_label: label must be a non-empty string")
|
||||
label_id = self.label_id(repo, name)
|
||||
if label_id is None:
|
||||
raise GiteaClientError(
|
||||
f"add_label: unknown label {name!r} in repo {self.owner}/{repo}")
|
||||
if name in self._issue_label_names(repo, issue_number):
|
||||
return # already present: idempotent no-op
|
||||
self._request("POST", f"{repo}/issues/{issue_number}/labels", {"labels": [label_id]})
|
||||
if name not in self._issue_label_names(repo, issue_number):
|
||||
raise GiteaClientError(
|
||||
f"add_label: label {name!r} missing on issue {issue_number} after POST")
|
||||
|
||||
def remove_label(self, repo: str, issue_number: int, label: str) -> None:
|
||||
"""Remove ``label`` from ``owner/repo`` issue ``issue_number``.
|
||||
|
||||
Resolves the repository label id, DELETEs it, and verifies the
|
||||
resulting label state. Absent labels are a no-op. An unknown label
|
||||
or a failed verification raises a clear :class:`GiteaClientError`.
|
||||
"""
|
||||
name = (label or "").strip()
|
||||
if not name:
|
||||
raise GiteaClientError("remove_label: label must be a non-empty string")
|
||||
label_id = self.label_id(repo, name)
|
||||
if label_id is None:
|
||||
raise GiteaClientError(
|
||||
f"remove_label: unknown label {name!r} in repo {self.owner}/{repo}")
|
||||
if name not in self._issue_label_names(repo, issue_number):
|
||||
return # already absent: idempotent no-op
|
||||
self._api("DELETE",
|
||||
f"repos/{self.owner}/{repo}/issues/{issue_number}/labels/{label_id}")
|
||||
if name in self._issue_label_names(repo, issue_number):
|
||||
raise GiteaClientError(
|
||||
f"remove_label: label {name!r} still on issue {issue_number} after DELETE")
|
||||
|
||||
def merge_pr(self, repo: str, pr_number: int,
|
||||
expected_head_sha: Optional[str] = None) -> bool:
|
||||
"""Merge ``owner/repo`` PR ``pr_number``; True when it is merged.
|
||||
|
||||
Uses the Gitea merge endpoint and confirms the PR is merged after
|
||||
the request. The merge request carries the Gitea schema's
|
||||
mandatory ``Do`` mode and the reviewed ``head_commit_id``, so a
|
||||
branch that moved after review cannot be merged accidentally.
|
||||
|
||||
``expected_head_sha`` (optional, preserving the two-argument
|
||||
protocol) is the head the caller actually REVIEWED. When given,
|
||||
the pre-merge GET result is compared against it and a mismatch
|
||||
raises :class:`GiteaClientError` BEFORE any POST — closing the
|
||||
TOCTOU window between review and merge instead of only guarding
|
||||
the milliseconds between the in-method GET and POST. The
|
||||
reviewed SHA itself is sent as ``head_commit_id``. When omitted,
|
||||
the live PR head is read (and pinned) as before, which keeps the
|
||||
existing callers working while the policy call sites migrate to
|
||||
passing the reviewed head.
|
||||
|
||||
An already-merged PR short-circuits to True (no duplicate merge
|
||||
request). Non-2xx responses (e.g. 409 when the head is not
|
||||
mergeable) raise :class:`GiteaApiError`; a PR record without a
|
||||
head commit id raises :class:`GiteaClientError` rather than
|
||||
merging unpinned.
|
||||
"""
|
||||
current = self._request("GET", f"{repo}/pulls/{pr_number}") or {}
|
||||
if current.get("merged"):
|
||||
return True
|
||||
live_head = (current.get("head") or {}).get("sha")
|
||||
if expected_head_sha is not None and live_head != expected_head_sha:
|
||||
raise GiteaClientError(
|
||||
f"merge_pr: PR {pr_number} head is {live_head!r} but the "
|
||||
f"reviewed head is {expected_head_sha!r}; refusing to merge "
|
||||
"a branch that moved after review")
|
||||
head_commit_id = expected_head_sha if expected_head_sha is not None \
|
||||
else live_head
|
||||
if not head_commit_id:
|
||||
raise GiteaClientError(
|
||||
f"merge_pr: PR {pr_number} record has no head commit id; "
|
||||
"refusing to merge without the reviewed head")
|
||||
self._request("POST", f"{repo}/pulls/{pr_number}/merge",
|
||||
{"Do": "merge", "head_commit_id": head_commit_id})
|
||||
merged = self._request("GET", f"{repo}/pulls/{pr_number}") or {}
|
||||
return bool(merged.get("merged"))
|
||||
|
||||
def close_issue(self, repo: str, issue_number: int) -> None:
|
||||
"""Close ``owner/repo`` issue ``issue_number``.
|
||||
|
||||
Updates the issue state and confirms the issue reports ``closed``
|
||||
afterwards; a failed verification raises :class:`GiteaClientError`.
|
||||
"""
|
||||
self._request("PATCH", f"{repo}/issues/{issue_number}", {"state": "closed"})
|
||||
issue = self._request("GET", f"{repo}/issues/{issue_number}") or {}
|
||||
if issue.get("state") != "closed":
|
||||
raise GiteaClientError(
|
||||
f"close_issue: issue {issue_number} state is {issue.get('state')!r} "
|
||||
"after close request")
|
||||
|
||||
# -- shared helpers -----------------------------------------------------
|
||||
|
||||
def _issue_label_names(self, repo: str, issue_number: int) -> list:
|
||||
"""Label NAMES currently on ``owner/repo`` issue ``issue_number``."""
|
||||
issue = self._request("GET", f"{repo}/issues/{issue_number}") or {}
|
||||
return [label.get("name") for label in (issue.get("labels") or [])
|
||||
if isinstance(label, dict) and label.get("name")]
|
||||
|
||||
615
tests/test_gitea_client.py
Normal file
615
tests/test_gitea_client.py
Normal file
@ -0,0 +1,615 @@
|
||||
"""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_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}"
|
||||
Loading…
Reference in New Issue
Block a user