harden dashboard API contracts and proxy errors
This commit is contained in:
@@ -839,10 +839,11 @@ class ApiHandler(BaseHTTPRequestHandler):
|
||||
return self.send_json(200,{"business_id":bid,"status":"unknown","reason":"not_configured","provider_configured":False,"items":[{"domain":d,"status":"unknown","reason":"not_configured"} for d in domains]})
|
||||
|
||||
def list_jobs(self, db, org, query):
|
||||
try: limit=max(1,min(int(query.get("page_size",[50])[0]),JOB_PAGE_SIZE)); offset=max(0,int(query.get("offset",[0])[0]))
|
||||
try:
|
||||
limit=max(1,min(int(query.get("page_size",[50])[0]),JOB_PAGE_SIZE)); page=max(1,int(query.get("page",[1])[0])); offset=max(0,int(query.get("offset",[0])[0]))+(page-1)*limit
|
||||
except (ValueError, TypeError): return self.send_json(400,{"error":"invalid_pagination"})
|
||||
rows=db.execute("SELECT * FROM jobs WHERE organization_id=? ORDER BY id DESC LIMIT ? OFFSET ?",(org,limit+1,offset)).fetchall(); more=len(rows)>limit
|
||||
return self.send_json(200,{"organization_id":org,"items":[job_json(r) for r in rows[:limit]],"limit":limit,"offset":offset,"has_more":more})
|
||||
return self.send_json(200,{"organization_id":org,"items":[job_json(r) for r in rows[:limit]],"limit":limit,"offset":offset,"page":page,"page_size":limit,"has_more":more,"has_next":more,"next_page":page+1 if more else None})
|
||||
|
||||
def _discovery_run_json(self, row):
|
||||
item = row_json(row)
|
||||
@@ -1038,7 +1039,7 @@ class ApiHandler(BaseHTTPRequestHandler):
|
||||
if stage: where.append("EXISTS (SELECT 1 FROM pipeline_entries p WHERE p.business_id=b.id AND p.organization_id=b.organization_id AND p.stage=?)"); params.append(stage)
|
||||
offset=(number("cursor",0) or 0)+(page-1)*size
|
||||
rows=db.execute("SELECT b.* FROM businesses b WHERE "+" AND ".join(where)+" ORDER BY b.score DESC,b.id LIMIT ? OFFSET ?",params+[size+1,offset]).fetchall(); more=len(rows)>size; rows=rows[:size]
|
||||
return self.send_json(200,{"organization_id":org,"items":[row_json(r) for r in rows],"page":page,"page_size":size,"next_cursor":str(offset+size) if more else None})
|
||||
return self.send_json(200,{"organization_id":org,"items":[row_json(r) for r in rows],"page":page,"page_size":size,"has_next":more,"next_page":page+1 if more else None,"next_cursor":str(offset+size) if more else None})
|
||||
def bulk_review(self, payload, db, user):
|
||||
ids = payload.get("ids", payload.get("business_ids")); action = str(payload.get("action", "")).strip().lower()
|
||||
if not isinstance(ids, list) or not ids or len(ids) > 100 or any(not isinstance(i, int) or i < 1 for i in ids) or len(set(ids)) != len(ids): return self.send_json(400, {"error": "invalid_bulk_ids"})
|
||||
|
||||
@@ -0,0 +1,95 @@
|
||||
import json
|
||||
import os
|
||||
import threading
|
||||
import unittest
|
||||
from http.client import HTTPConnection
|
||||
from tempfile import TemporaryDirectory
|
||||
|
||||
from app.main import create_server
|
||||
|
||||
|
||||
class DashboardBootstrapContractTests(unittest.TestCase):
|
||||
"""Exercise the exact independent GETs issued by the dashboard bootstrap."""
|
||||
|
||||
BOOTSTRAP_GETS = (
|
||||
("/api/v1/businesses?page=1&page_size=10", ("items", "page", "page_size", "next_cursor")),
|
||||
("/api/v1/dashboard/summary", ("businesses", "counts", "clickable_filters")),
|
||||
("/api/v1/jobs", ("items", "has_more", "limit", "offset")),
|
||||
("/api/v1/sources", ("items",)),
|
||||
("/api/v1/sources/adapters", ("items",)),
|
||||
("/api/v1/discovery-runs?page_size=50", ("items",)),
|
||||
("/api/v1/scoring/summary", ("businesses", "bands")),
|
||||
("/api/v1/score-rules", ("items",)),
|
||||
("/api/v1/saved-filters", ("items",)),
|
||||
("/api/v1/review-queue?page=1&page_size=100", ("items", "has_more", "limit", "offset")),
|
||||
("/api/v1/pipeline-entries", ("items",)),
|
||||
("/api/v1/reports/pipeline", ("items",)),
|
||||
("/api/v1/reports/outcomes", ("items",)),
|
||||
("/api/v1/reports/activity", ("items",)),
|
||||
("/api/v1/suppressions", ("items",)),
|
||||
("/api/v1/outreach/provider-config", ("enabled", "policy")),
|
||||
("/api/v1/ai/provider-config", ("status",)),
|
||||
)
|
||||
|
||||
def setUp(self):
|
||||
self.tmp = TemporaryDirectory()
|
||||
os.environ["BOOTSTRAP_ADMIN_EMAIL"] = "owner@example.test"
|
||||
os.environ["BOOTSTRAP_ADMIN_PASSWORD"] = "development-password"
|
||||
self.server = create_server("127.0.0.1", 0, self.tmp.name + "/dashboard.db")
|
||||
self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
|
||||
self.thread.start()
|
||||
self.conn = HTTPConnection("127.0.0.1", self.server.server_port, timeout=3)
|
||||
self.cookie = None
|
||||
status, body = self.request("POST", "/api/v1/auth/login", {"email": "owner@example.test", "password": "development-password"})
|
||||
self.assertEqual(status, 200, body)
|
||||
|
||||
def tearDown(self):
|
||||
self.server.shutdown()
|
||||
self.server.server_close()
|
||||
self.thread.join(timeout=2)
|
||||
self.tmp.cleanup()
|
||||
|
||||
def request(self, method, path, payload=None, authenticated=True):
|
||||
body = json.dumps(payload).encode() if payload is not None else None
|
||||
headers = {"Content-Type": "application/json"} if body else {}
|
||||
if authenticated and self.cookie:
|
||||
headers["Cookie"] = self.cookie
|
||||
self.conn.request(method, path, body, headers)
|
||||
response = self.conn.getresponse()
|
||||
set_cookie = response.getheader("Set-Cookie")
|
||||
if set_cookie and "session=" in set_cookie:
|
||||
self.cookie = set_cookie.split(";", 1)[0]
|
||||
raw = response.read()
|
||||
return response.status, json.loads(raw or b"{}")
|
||||
|
||||
def test_every_bootstrap_get_is_authenticated_json_and_shape_compatible(self):
|
||||
failures = []
|
||||
for path, required in self.BOOTSTRAP_GETS:
|
||||
status, body = self.request("GET", path)
|
||||
if status != 200:
|
||||
failures.append(f"{path}: HTTP {status} {body}")
|
||||
elif not all(key in body for key in required):
|
||||
failures.append(f"{path}: missing {sorted(set(required) - set(body))} in {body}")
|
||||
self.assertEqual(failures, [])
|
||||
status, body = self.request("GET", "/api/v1/businesses?page=1&page_size=10")
|
||||
self.assertEqual(status, 200)
|
||||
self.assertIn("has_next", body)
|
||||
self.assertIn("next_page", body)
|
||||
status, body = self.request("GET", "/api/v1/jobs?page=1&page_size=10")
|
||||
self.assertEqual(status, 200)
|
||||
self.assertIn("page", body)
|
||||
self.assertIn("page_size", body)
|
||||
|
||||
status, body = self.request("GET", "/api/v1/businesses?page_size=0")
|
||||
self.assertEqual(status, 400)
|
||||
self.assertEqual(body["error"], "invalid_pagination")
|
||||
status, body = self.request("GET", "/api/v1/health/live", authenticated=False)
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(body["status"], "ok")
|
||||
status, body = self.request("GET", "/api/v1/health/ready", authenticated=False)
|
||||
self.assertEqual(status, 200)
|
||||
self.assertTrue(body["ready"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
+18
-6
@@ -3,6 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import http.client
|
||||
import json
|
||||
import os
|
||||
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
|
||||
from urllib.parse import urlsplit
|
||||
@@ -15,20 +16,29 @@ UPSTREAM_PORT = int(os.environ.get("API_UPSTREAM_PORT", "8000"))
|
||||
class ProxyStaticHandler(SimpleHTTPRequestHandler):
|
||||
proxy_api = True
|
||||
|
||||
def _send_json_error(self, status: int, code: str) -> None:
|
||||
payload = json.dumps({"error": code}, separators=(",", ":")).encode()
|
||||
self.send_response(status)
|
||||
self.send_header("Content-Type", "application/json; charset=utf-8")
|
||||
self.send_header("Cache-Control", "no-store, private")
|
||||
self.send_header("Content-Length", str(len(payload)))
|
||||
self.end_headers()
|
||||
self.wfile.write(payload)
|
||||
|
||||
def _proxy_request(self) -> None:
|
||||
parsed = urlsplit(self.path)
|
||||
if parsed.path == "/api" or parsed.path.startswith("/api/"):
|
||||
target = self.path
|
||||
else:
|
||||
self.send_error(404)
|
||||
self._send_json_error(404, "not_found")
|
||||
return
|
||||
try:
|
||||
length = int(self.headers.get("Content-Length", "0"))
|
||||
except ValueError:
|
||||
self.send_error(400, "invalid content length")
|
||||
self._send_json_error(400, "invalid_content_length")
|
||||
return
|
||||
if length > MAX_PROXY_BODY:
|
||||
self.send_error(413, "request body too large")
|
||||
self._send_json_error(413, "request_body_too_large")
|
||||
return
|
||||
body = self.rfile.read(length) if length else None
|
||||
headers = {
|
||||
@@ -37,13 +47,14 @@ class ProxyStaticHandler(SimpleHTTPRequestHandler):
|
||||
if key.lower() in {"accept", "content-type", "cookie", "user-agent", "x-request-id"}
|
||||
}
|
||||
headers["Host"] = f"{UPSTREAM_HOST}:{UPSTREAM_PORT}"
|
||||
connection = http.client.HTTPConnection(UPSTREAM_HOST, UPSTREAM_PORT, timeout=15)
|
||||
connection = None
|
||||
try:
|
||||
connection = http.client.HTTPConnection(UPSTREAM_HOST, UPSTREAM_PORT, timeout=15)
|
||||
connection.request(self.command, target, body=body, headers=headers)
|
||||
response = connection.getresponse()
|
||||
payload = response.read(MAX_PROXY_BODY + 1)
|
||||
if len(payload) > MAX_PROXY_BODY:
|
||||
self.send_error(502, "upstream response too large")
|
||||
self._send_json_error(502, "upstream_response_too_large")
|
||||
return
|
||||
self.send_response(response.status, response.reason)
|
||||
for key, value in response.getheaders():
|
||||
@@ -52,8 +63,9 @@ class ProxyStaticHandler(SimpleHTTPRequestHandler):
|
||||
self.end_headers()
|
||||
self.wfile.write(payload)
|
||||
except (OSError, http.client.HTTPException) as exc:
|
||||
self.send_error(502, f"api upstream unavailable: {exc}")
|
||||
self._send_json_error(502, "api_upstream_unavailable")
|
||||
finally:
|
||||
if connection is not None:
|
||||
connection.close()
|
||||
|
||||
def do_GET(self):
|
||||
|
||||
@@ -1,10 +1,14 @@
|
||||
import http.client
|
||||
import json
|
||||
import os
|
||||
import threading
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
from server import ProxyStaticHandler
|
||||
from http.server import ThreadingHTTPServer
|
||||
|
||||
RealHTTPConnection = http.client.HTTPConnection
|
||||
|
||||
|
||||
class FakeUpstream:
|
||||
@@ -30,6 +34,22 @@ class ProxyServerTests(unittest.TestCase):
|
||||
with patch('server.http.client.HTTPConnection', FakeUpstream):
|
||||
self.assertTrue(server.proxy_api)
|
||||
|
||||
def test_upstream_failure_is_structured_json(self):
|
||||
httpd = ThreadingHTTPServer(("127.0.0.1", 0), ProxyStaticHandler)
|
||||
thread = threading.Thread(target=httpd.serve_forever, daemon=True)
|
||||
thread.start()
|
||||
try:
|
||||
with patch('server.http.client.HTTPConnection', side_effect=OSError('connection refused')):
|
||||
conn = RealHTTPConnection('127.0.0.1', httpd.server_port, timeout=3)
|
||||
conn.request('GET', '/api/v1/dashboard/summary')
|
||||
response = conn.getresponse()
|
||||
body = response.read()
|
||||
self.assertEqual(response.status, 502)
|
||||
self.assertEqual(response.getheader('Content-Type'), 'application/json; charset=utf-8')
|
||||
self.assertEqual(json.loads(body), {'error': 'api_upstream_unavailable'})
|
||||
finally:
|
||||
httpd.shutdown(); httpd.server_close(); thread.join(timeout=2)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user