70 lines
1.9 KiB
Python
Executable File
70 lines
1.9 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Mock webhook for showcase/e2e.
|
|
|
|
Decides response status by substring in incident.incident_key from request body:
|
|
- contains 'retry' -> 503 (retryable failure)
|
|
- contains 'fail' -> 400 (permanent failure)
|
|
- otherwise -> 200 (success)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
|
|
def _classify(body: bytes) -> int:
|
|
try:
|
|
data = json.loads(body or b"{}")
|
|
except Exception:
|
|
return 200
|
|
incident = data.get("incident") or {}
|
|
key = str(incident.get("incident_key") or "")
|
|
if "retry" in key:
|
|
return 503
|
|
if "fail" in key:
|
|
return 400
|
|
return 200
|
|
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def do_POST(self) -> None: # noqa: N802
|
|
length = int(self.headers.get("Content-Length") or 0)
|
|
body = self.rfile.read(length) if length else b""
|
|
code = _classify(body)
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", "text/plain")
|
|
self.send_header("Content-Length", "0")
|
|
self.end_headers()
|
|
try:
|
|
data = json.loads(body or b"{}")
|
|
inc = (data.get("incident") or {})
|
|
print(
|
|
f"webhook {code} event_type={data.get('event_type')!r} "
|
|
f"incident_key={inc.get('incident_key')!r}",
|
|
flush=True,
|
|
)
|
|
except Exception:
|
|
print(f"webhook {code} (unparsable body, {length}B)", flush=True)
|
|
|
|
def log_message(self, *_args: object) -> None:
|
|
return
|
|
|
|
|
|
def main() -> int:
|
|
addr = ("0.0.0.0", 8080)
|
|
srv = ThreadingHTTPServer(addr, Handler)
|
|
print(f"mock_webhook listening on {addr[0]}:{addr[1]}", flush=True)
|
|
try:
|
|
srv.serve_forever()
|
|
except KeyboardInterrupt:
|
|
pass
|
|
finally:
|
|
srv.server_close()
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|