informed-decision/tests/test_http_transport.py

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),
                "/plain-refuse": (403, b"forbidden secret-sentinel"),
                "/large-refuse": (401, b"x" * 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"})


@pytest.mark.parametrize("path,status", [("/plain-refuse",403),("/large-refuse",401)])
def test_http_caller_refusal_is_not_relabelled_as_transport_outage(path,status):
    with upstream() as (origin,calls):
        assert JSONTransport(allow_internal_http=True).request("GET",origin+path)==(status,{})