-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathsimple-mock-api.py
More file actions
97 lines (80 loc) · 3.08 KB
/
simple-mock-api.py
File metadata and controls
97 lines (80 loc) · 3.08 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
"""Lightweight HTTP mock API server driven by a JSON config file.
Config format (mock-api.json):
{
"routes": [
{"method": "GET", "path": "/health", "status": 200, "body": {"ok": true}},
{"method": "POST", "path": "/echo", "status": 201, "body": {"created": true}, "delay": 0.3}
]
}
"""
import json
import time
import argparse
from http.server import HTTPServer, BaseHTTPRequestHandler
from pathlib import Path
from urllib.parse import urlparse
def load_config(path: str) -> dict:
return json.loads(Path(path).read_text(encoding="utf-8"))
def make_handler(routes: dict, verbose: bool) -> type:
class Handler(BaseHTTPRequestHandler):
def log_message(self, fmt: str, *args) -> None: # type: ignore[override]
if verbose:
super().log_message(fmt, *args)
def handle_request(self) -> None:
parsed = urlparse(self.path)
route = routes.get(f"{self.command} {parsed.path}") or routes.get(
f"ANY {parsed.path}"
)
if route is None:
self.send_response(404)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps({"error": "not found"}).encode())
return
delay = route.get("delay", 0)
if delay:
time.sleep(delay)
status = route.get("status", 200)
body = route.get("body", {})
headers = route.get("headers", {})
self.send_response(status)
self.send_header(
"Content-Type", headers.get("Content-Type", "application/json")
)
for k, v in headers.items():
if k != "Content-Type":
self.send_header(k, v)
self.end_headers()
if isinstance(body, (dict, list)):
self.wfile.write(json.dumps(body).encode())
else:
self.wfile.write(str(body).encode())
do_GET = do_POST = do_PUT = do_DELETE = do_PATCH = handle_request
return Handler
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Lightweight HTTP mock API server.")
parser.add_argument(
"config",
nargs="?",
default="mock-api.json",
help="JSON config file (default: mock-api.json)",
)
parser.add_argument("-p", "--port", type=int, default=8000)
parser.add_argument("--host", default="127.0.0.1")
parser.add_argument(
"-q", "--quiet", action="store_true", help="Suppress request logs"
)
args = parser.parse_args()
cfg = load_config(args.config)
routes: dict = {}
for route in cfg.get("routes", []):
method = route.get("method", "ANY").upper()
routes[f"{method} {route['path']}"] = route
server = HTTPServer((args.host, args.port), make_handler(routes, not args.quiet))
print(
f"Mock API on http://{args.host}:{args.port} ({len(routes)} routes) Ctrl+C to stop"
)
try:
server.serve_forever()
except KeyboardInterrupt:
print("\nStopped.")