stackchain-dashboard/tests/e2e/fake_gitea.py
timmy c86a63c54b
All checks were successful
CI / lint (pull_request) Successful in 1m54s
CI / build-release (pull_request) Successful in 5s
CI / browser-journey (pull_request) Successful in 56s
CI / release-candidate (pull_request) Has been skipped
test: gate offline release journey deterministically
2026-08-15 13:46:28 +00:00

87 lines
3.1 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 urllib.parse import 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"}
class FakeGiteaServer(ThreadingHTTPServer):
def __init__(self, address: tuple[str, int]):
super().__init__(address, FakeGiteaHandler)
self.created_issues: list[dict] = []
self.issue_creation_enabled = False
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
path = urlsplit(self.path).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.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
length = int(self.headers.get("Content-Length", "0"))
payload = json.loads(self.rfile.read(length) or b"{}")
if path != "/api/v1/repos/acme/mobile/issues":
self._json(404, {"message": "not found"})
return
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",
},
)