195 lines
7.8 KiB
Python
195 lines
7.8 KiB
Python
"""Deterministic Gitea double used by the release-artifact browser journey."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from threading import Event
|
|
from urllib.parse import parse_qs, urlsplit
|
|
|
|
|
|
REPOSITORY = {
|
|
"id": 7,
|
|
"name": "mobile",
|
|
"full_name": "acme/mobile",
|
|
"description": "Mobile filing acceptance repository",
|
|
"html_url": "http://127.0.0.1/acme/mobile",
|
|
"owner": {"login": "acme"},
|
|
}
|
|
USER = {"id": 1, "login": "timmy", "full_name": "Timmy"}
|
|
AVAILABLE_ISSUES = [
|
|
{
|
|
"id": 101,
|
|
"number": 41,
|
|
"title": "Ship mobile capture",
|
|
"body": "Complete the capture flow on a phone.",
|
|
"state": "open",
|
|
"html_url": "http://127.0.0.1/acme/mobile/issues/41",
|
|
"repository": REPOSITORY,
|
|
"user": USER,
|
|
"assignees": [],
|
|
"labels": [{"name": "P1"}],
|
|
"updated_at": "2026-08-15T12:00:00Z",
|
|
},
|
|
{
|
|
"id": 102,
|
|
"number": 42,
|
|
"title": "Polish desktop filters",
|
|
"body": "Keep the desktop filter controls clear.",
|
|
"state": "open",
|
|
"html_url": "http://127.0.0.1/acme/mobile/issues/42",
|
|
"repository": REPOSITORY,
|
|
"user": USER,
|
|
"assignees": [],
|
|
"labels": [],
|
|
"updated_at": "2026-08-15T11:00:00Z",
|
|
},
|
|
]
|
|
|
|
|
|
class FakeGiteaServer(ThreadingHTTPServer):
|
|
def __init__(self, address: tuple[str, int]):
|
|
super().__init__(address, FakeGiteaHandler)
|
|
self.created_issues: list[dict] = []
|
|
self.issue_creation_enabled = False
|
|
self.issue_creation_ready = Event()
|
|
self.assigned_issue_numbers = [issue["number"] for issue in AVAILABLE_ISSUES]
|
|
self.comments: list[tuple[int, str]] = []
|
|
self.requests: list[tuple[str, str]] = []
|
|
|
|
|
|
class FakeGiteaHandler(BaseHTTPRequestHandler):
|
|
server: FakeGiteaServer
|
|
|
|
def log_message(self, _format: str, *_args: object) -> None:
|
|
return
|
|
|
|
def _json(self, status: int, payload: object, **headers: str) -> None:
|
|
body = json.dumps(payload).encode()
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
for name, value in headers.items():
|
|
self.send_header(name, value)
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def do_GET(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API
|
|
parsed = urlsplit(self.path)
|
|
path = parsed.path
|
|
query = parse_qs(parsed.query)
|
|
self.server.requests.append(("GET", self.path))
|
|
if path == "/api/v1/user":
|
|
self._json(200, USER)
|
|
elif path == "/api/v1/user/repos":
|
|
self._json(200, [REPOSITORY], **{"X-Total-Count": "1"})
|
|
elif path in {"/api/v1/repos/acme/mobile", "/api/v1/repos/acme/mobile/"}:
|
|
self._json(200, REPOSITORY)
|
|
elif path == "/api/v1/repos/search":
|
|
self._json(200, {"data": [REPOSITORY], "ok": True})
|
|
elif path == "/api/v1/repos/issues/search":
|
|
is_assigned_scan = query.get("assigned") == ["true"] and query.get("type") == ["issues"]
|
|
is_available_scan = not any(
|
|
name in query for name in ("assigned", "created", "review_requested", "q")
|
|
) and query.get("type") == ["issues"]
|
|
assigned = [
|
|
{**issue, "assignees": [USER]}
|
|
for issue in AVAILABLE_ISSUES
|
|
if issue["number"] in self.server.assigned_issue_numbers
|
|
]
|
|
available = [
|
|
issue for issue in AVAILABLE_ISSUES
|
|
if issue["number"] not in self.server.assigned_issue_numbers
|
|
]
|
|
issues = assigned if is_assigned_scan else available if is_available_scan else []
|
|
self._json(200, issues, **{"X-Total-Count": str(len(issues))})
|
|
elif path.startswith("/api/v1/repos/acme/mobile/issues/") and path.endswith("/comments"):
|
|
self._json(200, [], **{"X-Total-Count": "0"})
|
|
elif path.startswith("/api/v1/repos/acme/mobile/issues/") and path.endswith("/dependencies"):
|
|
self._json(200, [])
|
|
elif path.startswith("/api/v1/repos/acme/mobile/issues/"):
|
|
try:
|
|
number = int(path.rsplit("/", 1)[-1])
|
|
except ValueError:
|
|
number = 0
|
|
issue = next((item for item in AVAILABLE_ISSUES if item["number"] == number), None)
|
|
if issue and number in self.server.assigned_issue_numbers:
|
|
issue = {**issue, "assignees": [USER]}
|
|
self._json(200, issue) if issue else self._json(404, {"message": "not found"})
|
|
elif path.startswith("/api/v1/"):
|
|
self._json(200, [], **{"X-Total-Count": "0"})
|
|
else:
|
|
self._json(404, {"message": "not found"})
|
|
|
|
def do_POST(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API
|
|
path = urlsplit(self.path).path
|
|
self.server.requests.append(("POST", self.path))
|
|
length = int(self.headers.get("Content-Length", "0"))
|
|
payload = json.loads(self.rfile.read(length) or b"{}")
|
|
if path.startswith("/api/v1/repos/acme/mobile/issues/") and path.endswith("/comments"):
|
|
try:
|
|
number = int(path.split("/")[-2])
|
|
except ValueError:
|
|
self._json(404, {"message": "not found"})
|
|
return
|
|
body = str(payload.get("body", ""))
|
|
self.server.comments.append((number, body))
|
|
self._json(
|
|
201,
|
|
{
|
|
"id": len(self.server.comments),
|
|
"body": body,
|
|
"user": USER,
|
|
"created_at": "2026-08-16T20:00:00Z",
|
|
"updated_at": "2026-08-16T20:00:00Z",
|
|
"html_url": f"http://127.0.0.1/acme/mobile/issues/{number}#issuecomment-1",
|
|
},
|
|
)
|
|
return
|
|
if path != "/api/v1/repos/acme/mobile/issues":
|
|
self._json(404, {"message": "not found"})
|
|
return
|
|
if not self.server.issue_creation_enabled:
|
|
self.server.issue_creation_ready.wait(timeout=45)
|
|
if not self.server.issue_creation_enabled:
|
|
self._json(503, {"message": "release journey is still offline"})
|
|
return
|
|
self.server.created_issues.append(payload)
|
|
assignee = payload.get("assignee")
|
|
self._json(
|
|
201,
|
|
{
|
|
"id": 101,
|
|
"number": 41,
|
|
"title": payload.get("title", ""),
|
|
"body": payload.get("body", ""),
|
|
"state": "open",
|
|
"html_url": "http://127.0.0.1/acme/mobile/issues/41",
|
|
"assignees": [{"login": assignee}] if assignee else [],
|
|
"labels": [],
|
|
"milestone": None,
|
|
"due_date": payload.get("due_date"),
|
|
"updated_at": "2026-08-15T12:00:00Z",
|
|
},
|
|
)
|
|
|
|
def do_PATCH(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API
|
|
path = urlsplit(self.path).path
|
|
length = int(self.headers.get("Content-Length", "0"))
|
|
payload = json.loads(self.rfile.read(length) or b"{}")
|
|
if not path.startswith("/api/v1/repos/acme/mobile/issues/"):
|
|
self._json(404, {"message": "not found"})
|
|
return
|
|
try:
|
|
number = int(path.rsplit("/", 1)[-1])
|
|
except ValueError:
|
|
self._json(404, {"message": "not found"})
|
|
return
|
|
issue = next((item for item in AVAILABLE_ISSUES if item["number"] == number), None)
|
|
if issue is None:
|
|
self._json(404, {"message": "not found"})
|
|
return
|
|
assignee = payload.get("assignee")
|
|
self.server.assigned_issue_numbers.append(number)
|
|
self._json(200, {**issue, "assignees": [{"login": assignee}] if assignee else []})
|