DealSage template sync
83 lines
3.9 KiB
Python
Executable File
83 lines
3.9 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Stub of the DealSage endpoints the pipeline talks to, for local tests.
|
|
|
|
Records every request as JSON lines in the file named by STUB_LOG, and answers
|
|
like the real functions would:
|
|
POST /record-app-version allocate -> version 7 + upload token
|
|
promote -> promoted per STUB_POLICY (auto|manual)
|
|
fail -> ok
|
|
POST /record-app-check -> {"id": "..."}
|
|
POST /record-app-build start -> {"buildId": "bld-1"}; finish -> echo;
|
|
steps -> {"buildId": ..., "steps": <count>}
|
|
PUT /__upload/<app>/<v>/<path> -> 204
|
|
"""
|
|
import json, os, sys
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
|
|
LOG = os.environ.get("STUB_LOG", "/tmp/stub-requests.jsonl")
|
|
POLICY = os.environ.get("STUB_POLICY", "auto")
|
|
seq = {"n": 0}
|
|
|
|
class H(BaseHTTPRequestHandler):
|
|
def log_message(self, *a): # quiet
|
|
pass
|
|
|
|
def _record(self, body):
|
|
with open(LOG, "a") as f:
|
|
f.write(json.dumps({"method": self.command, "path": self.path,
|
|
"auth": self.headers.get("Authorization"), "body": body}) + "\n")
|
|
|
|
def _json(self, code, obj):
|
|
data = json.dumps(obj).encode()
|
|
self.send_response(code); self.send_header("content-type", "application/json")
|
|
self.send_header("content-length", str(len(data))); self.end_headers(); self.wfile.write(data)
|
|
|
|
def do_POST(self):
|
|
n = int(self.headers.get("content-length") or 0)
|
|
raw = self.rfile.read(n).decode() if n else ""
|
|
try: body = json.loads(raw) if raw else {}
|
|
except json.JSONDecodeError: body = {"_raw": raw}
|
|
self._record(body)
|
|
if self.headers.get("Authorization") != "Bearer test-publish-token":
|
|
return self._json(401, {"error": "bad token"})
|
|
if self.path.endswith("/record-app-version"):
|
|
a = body.get("action")
|
|
if a == "allocate":
|
|
return self._json(200, {"version": 7, "r2Prefix": f"apps/{body.get('appId')}/7/", "uploadToken": "upl-token"})
|
|
if a == "promote":
|
|
if POLICY == "auto":
|
|
return self._json(200, {"version": 7, "currentVersion": 7, "promoted": True, "channel": "live"})
|
|
return self._json(200, {"version": 7, "currentVersion": 6, "promoted": False, "channel": "branch/feature/x"})
|
|
if a == "fail":
|
|
return self._json(200, {"ok": True})
|
|
return self._json(400, {"error": "bad action"})
|
|
if self.path.endswith("/record-app-check"):
|
|
seq["n"] += 1
|
|
return self._json(200, {"id": f"chk-{seq['n']}"})
|
|
if self.path.endswith("/record-app-build"):
|
|
if body.get("action") == "start":
|
|
return self._json(200, {"buildId": "bld-1"})
|
|
if body.get("action") == "finish":
|
|
return self._json(200, {"buildId": body.get("buildId"), "outcome": body.get("outcome")})
|
|
if body.get("action") == "steps":
|
|
steps = body.get("steps")
|
|
if not isinstance(steps, list) or not steps:
|
|
return self._json(400, {"error": "steps must be a non-empty array"})
|
|
return self._json(200, {"buildId": body.get("buildId"), "steps": len(steps)})
|
|
return self._json(400, {"error": "bad action"})
|
|
self._json(404, {"error": "no route"})
|
|
|
|
def do_PUT(self):
|
|
n = int(self.headers.get("content-length") or 0)
|
|
self.rfile.read(n)
|
|
self._record({"bytes": n})
|
|
if not self.path.startswith("/__upload/"):
|
|
return self._json(404, {"error": "no route"})
|
|
if self.headers.get("Authorization") != "Bearer upl-token":
|
|
return self._json(401, {"error": "bad upload token"})
|
|
self.send_response(204); self.end_headers()
|
|
|
|
if __name__ == "__main__":
|
|
port = int(sys.argv[1]) if len(sys.argv) > 1 else 8765
|
|
HTTPServer(("127.0.0.1", port), H).serve_forever()
|