tomatocream
c6173b0ceb
Adds runtime deps (playwright, mitmproxy, anthropic, genson) and dev deps (pytest, pytest-asyncio); creates the tests/ scaffold with fixture_site.py, conftest.py, and a smoke test. Documents the mitmproxy aioquic/mitmproxy-rs stub workaround needed for free-threaded CPython 3.14. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
50 lines
1.9 KiB
Python
50 lines
1.9 KiB
Python
"""A tiny dependency-free JSON site for integration tests, served over HTTP."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import threading
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
|
|
class _Handler(BaseHTTPRequestHandler):
|
|
def log_message(self, *args: object) -> None: # silence test output
|
|
pass
|
|
|
|
def _send_json(self, status: int, payload: object) -> 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)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def do_GET(self) -> None:
|
|
if self.path == "/":
|
|
html = b"<html><body><script>fetch('/api/users')</script></body></html>"
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "text/html")
|
|
self.send_header("Content-Length", str(len(html)))
|
|
self.end_headers()
|
|
self.wfile.write(html)
|
|
elif self.path == "/api/users":
|
|
self._send_json(200, [{"id": 1, "name": "Ada"}])
|
|
elif self.path.startswith("/api/users/"):
|
|
self._send_json(200, {"id": int(self.path.rsplit("/", 1)[1]), "name": "Ada"})
|
|
else:
|
|
self._send_json(404, {"error": "not found"})
|
|
|
|
def do_POST(self) -> None:
|
|
length = int(self.headers.get("Content-Length", "0"))
|
|
raw = self.rfile.read(length) if length else b"{}"
|
|
self._send_json(201, {"received": json.loads(raw or b"{}")})
|
|
|
|
|
|
def start_fixture_site() -> tuple[ThreadingHTTPServer, str]:
|
|
"""Start the site on an ephemeral port; return (server, base_url)."""
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), _Handler)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
host, port = server.server_address
|
|
return server, f"http://{host}:{port}"
|