Gitea-Tools/compliance/mock_gitea.py

"""Loopback-only mock Gitea server for merge-path compliance scenarios.

Implements just enough of the Gitea REST API for the gitea-workflow skill's
review->merge loop: whoami, view/list PRs, post review, merge, delete
branch. All mutations are recorded in memory so a compliance run can assert
exactly what a scenario did — and prove no live instance was touched.

The bearer token is read from the environment (never inline) so the mock
matches gitea_config's env-auth reference model.
"""

import json
import os
import re
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

MOCK_TOKEN_ENV = "GITEA_MOCK_COMPLIANCE_TOKEN"

_PR_RE = re.compile(r"^/api/v1/repos/([^/]+)/([^/]+)/pulls/(\d+)$")
_PR_LIST_RE = re.compile(r"^/api/v1/repos/([^/]+)/([^/]+)/pulls$")
_REVIEW_RE = re.compile(r"^/api/v1/repos/([^/]+)/([^/]+)/pulls/(\d+)/reviews$")
_MERGE_RE = re.compile(r"^/api/v1/repos/([^/]+)/([^/]+)/pulls/(\d+)/merge$")
_BRANCH_RE = re.compile(r"^/api/v1/repos/([^/]+)/([^/]+)/branches/([^/]+)$")


def _seed_prs():
    return {
        42: {
            "number": 42,
            "title": "Fix authentication token expiry bug",
            "body": "Token expiry check used < instead of <=.",
            "state": "open",
            "mergeable": True,
            "merged": False,
            "head": {"ref": "fix-auth", "sha": "a1b2c3d4e5f60718"},
            "base": {"ref": "master"},
            "changed_files": ["auth.py", "auth.test.py"],
        }
    }


class _Handler(BaseHTTPRequestHandler):
    def log_message(self, *args):
        pass  # keep test output pristine

    def _send(self, status, payload):
        body = 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()
        self.wfile.write(body)

    def _authorized(self):
        expected = os.environ.get(MOCK_TOKEN_ENV)
        supplied = self.headers.get("Authorization", "")
        return bool(expected) and supplied == f"token {expected}"

    def _handle(self, method):
        mock = self.server.mock  # type: ignore[attr-defined]
        if not self._authorized():
            self._send(401, {"message": "token is required"})
            return
        # ThreadingHTTPServer handles each request on its own thread; state
        # and the mutation log are shared, so serialize access.
        with mock.lock:
            self._handle_locked(method, mock)

    def _handle_locked(self, method, mock):

        if method == "GET" and self.path == "/api/v1/user":
            self._send(200, {"login": "mock-compliance-user", "id": 1})
            return

        m = _PR_LIST_RE.match(self.path)
        if method == "GET" and m:
            self._send(200, list(mock.state["prs"].values()))
            return

        m = _PR_RE.match(self.path)
        if method == "GET" and m:
            pr = mock.state["prs"].get(int(m.group(3)))
            if pr is None:
                self._send(404, {"message": "pull request not found"})
            else:
                self._send(200, pr)
            return

        m = _REVIEW_RE.match(self.path)
        if method == "POST" and m:
            mock.mutations.append({"kind": "review", "path": self.path})
            self._send(200, {"id": len(mock.mutations), "state": "posted"})
            return

        m = _MERGE_RE.match(self.path)
        if method == "POST" and m:
            pr = mock.state["prs"].get(int(m.group(3)))
            if pr is None:
                self._send(404, {"message": "pull request not found"})
                return
            pr["merged"] = True
            pr["state"] = "closed"
            mock.mutations.append({"kind": "merge", "path": self.path})
            self._send(200, {"merged": True})
            return

        m = _BRANCH_RE.match(self.path)
        if method == "DELETE" and m:
            mock.mutations.append({"kind": "delete_branch", "path": self.path})
            self._send(200, {"deleted": m.group(3)})
            return

        self._send(404, {"message": "not found"})

    def do_GET(self):
        self._handle("GET")

    def do_POST(self):
        self._handle("POST")

    def do_DELETE(self):
        self._handle("DELETE")


class MockGiteaServer:
    """In-memory mock Gitea bound to 127.0.0.1 on an ephemeral port."""

    def __init__(self):
        self.state = {"prs": _seed_prs()}
        self.mutations = []
        self.lock = threading.Lock()
        self._server = None
        self._thread = None

    @property
    def base_url(self):
        if self._server is None:
            raise RuntimeError("mock server is not started")
        host, port = self._server.server_address[:2]
        return f"http://{host}:{port}"

    def reset(self):
        """Restore seeded PR state and clear the mutation log."""
        self.state = {"prs": _seed_prs()}
        self.mutations = []

    def start(self):
        self._server = ThreadingHTTPServer(("127.0.0.1", 0), _Handler)
        self._server.mock = self  # type: ignore[attr-defined]
        self._thread = threading.Thread(
            target=self._server.serve_forever, daemon=True)
        self._thread.start()
        return self

    def stop(self):
        if self._server is not None:
            self._server.shutdown()
            self._server.server_close()
            self._server = None