#!/usr/bin/env python3 """ Minimal HTTP stub for AI functional tests. Listens on 127.0.0.1: and serves canned JSON responses for any POST. Behaviour is selected via the URL path and the AI_STUB_MODE environment variable, set by the test runner before exec(). Modes: - "ok" : 200 with a valid OpenAI Responses-API body ({"output":[{"content":[{"text":"...","type":"output_text"}]}]}) - "openai" : 200 with a legacy OpenAI Chat Completions body ({"choices":[{"message":{"content":"..."}}]}) - "401" : 401 with an OpenAI-style error envelope - "500" : 500 with a body that triggers the parse-error fallback - "models" : 200 with a /v1/models list body The reply text is taken from AI_STUB_REPLY (default: "MOCK-REPLY"). The stub writes its actual listening port to stdout as a single line "PORT=" so the parent can read it back, then runs until SIGTERM. """ import json import os import sys from http.server import HTTPServer, BaseHTTPRequestHandler REPLY = os.environ.get("AI_STUB_REPLY", "MOCK-REPLY") MODE = os.environ.get("AI_STUB_MODE", "ok") def _dumps(body: object) -> bytes: """Produce compact JSON without whitespace — profanity's hand-rolled parser does strstr for `"key":"`, which fails when json.dumps inserts a space after the colon by default.""" return json.dumps(body, separators=(",", ":")).encode("utf-8") def _body_for(mode: str) -> tuple[int, bytes]: if mode == "ok": body = { "output": [ { "content": [ {"text": REPLY, "type": "output_text"} ] } ] } return 200, _dumps(body) if mode == "openai": body = { "choices": [ {"message": {"role": "assistant", "content": REPLY}} ] } return 200, _dumps(body) if mode == "401": body = { "error": { "message": REPLY, "type": "invalid_request_error", "code": "invalid_api_key", } } return 401, _dumps(body) if mode == "500": return 500, b"Internal Server Error" if mode == "models": body = { "object": "list", "data": [ {"id": "model-a", "object": "model"}, {"id": "model-b", "object": "model"}, {"id": "model-c", "object": "model"}, ], } return 200, _dumps(body) return 200, b"{}" class StubHandler(BaseHTTPRequestHandler): def _send(self): status, body = _body_for(MODE) 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_POST(self): length = int(self.headers.get("Content-Length", 0)) if length: self.rfile.read(length) self._send() def do_GET(self): self._send() def log_message(self, fmt, *args): return def main() -> None: port = int(sys.argv[1]) if len(sys.argv) > 1 else 0 srv = HTTPServer(("127.0.0.1", port), StubHandler) actual = srv.server_address[1] print(f"PORT={actual}", flush=True) try: srv.serve_forever() except KeyboardInterrupt: pass if __name__ == "__main__": main()