63 lines
2.3 KiB
Python
63 lines
2.3 KiB
Python
|
from contextlib import contextmanager
|
||
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||
|
|
import threading
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from informed_decision.http_transport import JSONTransport, TransportError
|
||
|
|
|
||
|
|
|
||
|
|
@contextmanager
|
||
|
|
def upstream():
|
||
|
|
calls = []
|
||
|
|
class Handler(BaseHTTPRequestHandler):
|
||
|
|
def do_GET(self):
|
||
|
|
calls.append((self.path, self.headers.get("Authorization")))
|
||
|
|
if self.path == "/redirect":
|
||
|
|
self.send_response(302)
|
||
|
|
self.send_header("Location", "/must-not-receive-token")
|
||
|
|
self.end_headers()
|
||
|
|
return
|
||
|
|
status, body = {
|
||
|
|
"/ok": (200, b'{"ok":true}'), "/refuse": (403, b'{"error":"forbidden"}'),
|
||
|
|
"/broken": (200, b"broken"), "/array": (200, b"[]"),
|
||
|
|
"/large": (200, b" " * 262145),
|
||
|
|
}.get(self.path, (500, b"{}"))
|
||
|
|
self.send_response(status)
|
||
|
|
self.end_headers()
|
||
|
|
self.wfile.write(body)
|
||
|
|
|
||
|
|
def log_message(self, *args): pass
|
||
|
|
|
||
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
||
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
||
|
|
thread.start()
|
||
|
|
try:
|
||
|
|
yield f"http://127.0.0.1:{server.server_port}", calls
|
||
|
|
finally:
|
||
|
|
server.shutdown()
|
||
|
|
server.server_close()
|
||
|
|
thread.join()
|
||
|
|
|
||
|
|
|
||
|
|
def test_redirect_never_forwards_bearer():
|
||
|
|
with upstream() as (origin, calls):
|
||
|
|
with pytest.raises(TransportError, match="redirect"):
|
||
|
|
JSONTransport(allow_internal_http=True).request("GET", origin + "/redirect",
|
||
|
|
headers={"Authorization": "Bearer synthetic-sentinel"})
|
||
|
|
assert calls == [("/redirect", "Bearer synthetic-sentinel")]
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize("path", ["/broken", "/array", "/large"])
|
||
|
|
def test_invalid_or_oversize_response_refused(path):
|
||
|
|
with upstream() as (origin, calls):
|
||
|
|
with pytest.raises(TransportError):
|
||
|
|
JSONTransport(allow_internal_http=True).request("GET", origin + path)
|
||
|
|
|
||
|
|
|
||
|
|
def test_json_success_and_typed_refusal_preserved():
|
||
|
|
with upstream() as (origin, calls):
|
||
|
|
client = JSONTransport(allow_internal_http=True)
|
||
|
|
assert client.request("GET", origin + "/ok") == (200, {"ok": True})
|
||
|
|
assert client.request("GET", origin + "/refuse") == (403, {"error": "forbidden"})
|