83 lines
2.5 KiB
Python
83 lines
2.5 KiB
Python
"""APIClient - Agent - Lightweight HTTP mock server."""
|
|
import threading
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
|
|
from app.core import storage
|
|
|
|
|
|
_server_instance: HTTPServer | None = None
|
|
_server_thread: threading.Thread | None = None
|
|
_server_port: int = 8888
|
|
|
|
|
|
class _MockHandler(BaseHTTPRequestHandler):
|
|
"""Queries the DB on every request so new/edited endpoints are served immediately."""
|
|
|
|
def log_message(self, fmt, *args):
|
|
pass # suppress default console logging
|
|
|
|
def _handle(self):
|
|
endpoints = storage.get_mock_endpoints()
|
|
matched = None
|
|
for ep in endpoints:
|
|
method_match = ep.method == "*" or ep.method == self.command
|
|
if method_match and ep.path == self.path:
|
|
matched = ep
|
|
break
|
|
|
|
if matched:
|
|
self.send_response(matched.status_code)
|
|
for k, v in matched.response_headers.items():
|
|
self.send_header(k, v)
|
|
body = matched.response_body.encode("utf-8")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
else:
|
|
body = b'{"error": "No mock endpoint matched"}'
|
|
self.send_response(404)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
do_GET = _handle
|
|
do_POST = _handle
|
|
do_PUT = _handle
|
|
do_PATCH = _handle
|
|
do_DELETE = _handle
|
|
do_HEAD = _handle
|
|
do_OPTIONS = _handle
|
|
|
|
|
|
def start(port: int = 8888) -> str:
|
|
global _server_instance, _server_thread, _server_port
|
|
if _server_instance:
|
|
return f"Already running on port {_server_port}"
|
|
_server_port = port
|
|
try:
|
|
_server_instance = HTTPServer(("localhost", port), _MockHandler)
|
|
except OSError as e:
|
|
return f"Failed to start: {e}"
|
|
_server_thread = threading.Thread(target=_server_instance.serve_forever, daemon=True)
|
|
_server_thread.start()
|
|
return f"Mock server running on http://localhost:{port}"
|
|
|
|
|
|
def stop() -> str:
|
|
global _server_instance, _server_thread
|
|
if _server_instance:
|
|
_server_instance.shutdown()
|
|
_server_instance = None
|
|
_server_thread = None
|
|
return "Mock server stopped"
|
|
return "Not running"
|
|
|
|
|
|
def is_running() -> bool:
|
|
return _server_instance is not None
|
|
|
|
|
|
def get_port() -> int:
|
|
return _server_port
|