from __future__ import annotations import asyncio import importlib.util import sys import types from pathlib import Path from types import SimpleNamespace import pytest ROOT = Path(__file__).resolve().parents[1] _ISOLATED_MODULE_NAMES = ( "headroom.proxy", "headroom.proxy.handlers", "httpx", "fastapi.responses", "tests.headroom_proxy_handlers_openai", "tests.headroom_proxy_handlers_streaming", ) @pytest.fixture(autouse=True) def restore_isolated_modules() -> None: saved_modules = {name: sys.modules.get(name) for name in _ISOLATED_MODULE_NAMES} try: yield finally: for name in _ISOLATED_MODULE_NAMES: sys.modules.pop(name, None) for name, module in saved_modules.items(): if module is not None: sys.modules[name] = module def _load_handler_module(monkeypatch: pytest.MonkeyPatch, module_name: str, relative_path: str): proxy_pkg = types.ModuleType("headroom.proxy") proxy_pkg.__path__ = [str(ROOT / "headroom" / "proxy")] monkeypatch.setitem(sys.modules, "headroom.proxy", proxy_pkg) handlers_pkg = types.ModuleType("headroom.proxy.handlers") handlers_pkg.__path__ = [str(ROOT / "headroom" / "proxy" / "handlers")] monkeypatch.setitem(sys.modules, "headroom.proxy.handlers", handlers_pkg) httpx_mod = types.ModuleType("httpx") httpx_mod.ConnectError = type("ConnectError", (Exception,), {}) httpx_mod.ConnectTimeout = type("ConnectTimeout", (Exception,), {}) httpx_mod.PoolTimeout = type("PoolTimeout", (Exception,), {}) httpx_mod.ReadTimeout = type("ReadTimeout", (Exception,), {}) monkeypatch.setitem(sys.modules, "httpx", httpx_mod) responses_mod = types.ModuleType("fastapi.responses") class Response: def __init__(self, content=None, status_code: int = 200, headers=None, media_type=None): self.content = content self.status_code = status_code self.headers = headers or {} self.media_type = media_type class StreamingResponse(Response): pass class JSONResponse(Response): pass responses_mod.Response = Response responses_mod.StreamingResponse = StreamingResponse responses_mod.JSONResponse = JSONResponse monkeypatch.setitem(sys.modules, "fastapi.responses", responses_mod) spec = importlib.util.spec_from_file_location(module_name, ROOT / relative_path) assert spec is not None and spec.loader is not None module = importlib.util.module_from_spec(spec) monkeypatch.setitem(sys.modules, module_name, module) spec.loader.exec_module(module) return module def test_openai_passthrough_applies_copilot_auth(monkeypatch: pytest.MonkeyPatch) -> None: openai_mod = _load_handler_module( monkeypatch, "tests.headroom_proxy_handlers_openai", "headroom/proxy/handlers/openai.py", ) seen: dict[str, object] = {} async def fake_apply(headers: dict[str, str], *, url: str) -> dict[str, str]: seen["headers"] = dict(headers) seen["url"] = url return {"Authorization": "Bearer upstream-token"} monkeypatch.setattr(openai_mod, "apply_copilot_api_auth", fake_apply) class Dummy(openai_mod.OpenAIHandlerMixin): def __init__(self) -> None: self.metrics = SimpleNamespace(record_request=self._record_request) self.http_client = SimpleNamespace(request=self._request) self.cost_tracker = None self._counter = 0 async def _record_request(self, **kwargs) -> None: # noqa: ANN003 return None async def _next_request_id(self) -> str: # The passthrough handler now allocates a request_id at end- # of-call because it records via ``_record_request_outcome``, # which requires one. Pre-refactor the dummy didn't need # this method because metrics.record_request was called # directly without a request_id. self._counter += 1 return f"req-{self._counter}" async def _record_request_outcome(self, outcome) -> None: # noqa: ANN001 from headroom.proxy.outcome import emit_request_outcome await emit_request_outcome(self, outcome) def _extract_tags(self, headers: dict) -> dict[str, str]: # Mirror of HeadroomProxy._extract_tags. The passthrough # handler now extracts tags at entry as part of the # outcome-tag invariant lock (PR #480). return { k.lower().replace("x-headroom-", ""): v for k, v in headers.items() if k.lower().startswith("x-headroom-") } async def _request(self, **kwargs): # noqa: ANN003 seen["request_kwargs"] = kwargs return SimpleNamespace(headers={}, content=b"{}", status_code=200) request = SimpleNamespace( url=SimpleNamespace(path="/v1/models", query=""), headers={ "authorization": "Bearer downstream", "host": "localhost", "accept-encoding": "gzip", }, method="GET", body=lambda: None, ) async def body() -> bytes: return b"" request.body = body handler = Dummy() response = asyncio.run( handler.handle_passthrough( request, "https://api.githubcopilot.com", "models", "openai", ) ) assert seen["url"] == "https://api.githubcopilot.com/models" assert seen["request_kwargs"]["headers"] == {"Authorization": "Bearer upstream-token"} assert response.status_code == 200 def test_streaming_response_applies_copilot_auth(monkeypatch: pytest.MonkeyPatch) -> None: streaming_mod = _load_handler_module( monkeypatch, "tests.headroom_proxy_handlers_streaming", "headroom/proxy/handlers/streaming.py", ) seen: dict[str, object] = {} async def fake_apply(headers: dict[str, str], *, url: str) -> dict[str, str]: seen["headers"] = dict(headers) seen["url"] = url return {"Authorization": "Bearer upstream-token"} monkeypatch.setattr(streaming_mod, "apply_copilot_api_auth", fake_apply) class Dummy(streaming_mod.StreamingMixin): def __init__(self) -> None: self.memory_handler = None self.config = SimpleNamespace( retry_max_attempts=1, retry_base_delay_ms=1, retry_max_delay_ms=1, ) self.http_client = SimpleNamespace( build_request=self._build_request, send=self._send, ) def _build_request(self, method: str, url: str, **kwargs): # noqa: ANN003 # PR-A3: streaming forwarder is byte-faithful; it now passes # ``content=`` instead of ``json=``. seen["request"] = { "method": method, "url": url, **kwargs, } return SimpleNamespace() async def _send(self, request, stream: bool): # noqa: ANN001, ANN003 return SimpleNamespace(headers={}, status_code=200) handler = Dummy() response = asyncio.run( handler._stream_response( url="https://api.githubcopilot.com/v1/responses", headers={"authorization": "Bearer downstream"}, body={"model": "gpt-4o"}, provider="openai", model="gpt-4o", request_id="req-test", original_tokens=0, optimized_tokens=0, tokens_saved=0, transforms_applied=[], tags={}, optimization_latency=0.0, ) ) assert seen["url"] == "https://api.githubcopilot.com/v1/responses" # PR-A3: byte-faithful forwarder always sets ``content-type`` explicitly. sent_headers = seen["request"]["headers"] assert sent_headers["Authorization"] == "Bearer upstream-token" assert sent_headers["content-type"] == "application/json" assert response.status_code == 200 def test_openai_chat_routes_copilot_requests_per_model(monkeypatch: pytest.MonkeyPatch) -> None: openai_mod = _load_handler_module( monkeypatch, "tests.headroom_proxy_handlers_openai", "headroom/proxy/handlers/openai.py", ) copilot_base = "https://api.githubcopilot.com" gpt54_mini_url = openai_mod.build_copilot_upstream_url( copilot_base, openai_mod._resolve_openai_handler_path( {}, handler_path=openai_mod._resolve_openai_chat_handler_path(copilot_base, "gpt-5.4-mini"), ), ) claude_url = openai_mod.build_copilot_upstream_url( copilot_base, openai_mod._resolve_openai_handler_path( {}, handler_path=openai_mod._resolve_openai_chat_handler_path( copilot_base, "claude-sonnet-5" ), ), ) openai_url = openai_mod.build_copilot_upstream_url( "https://api.openai.com", openai_mod._resolve_openai_handler_path( {}, handler_path=openai_mod._resolve_openai_chat_handler_path( "https://api.openai.com", "gpt-5.4-mini" ), ), ) assert gpt54_mini_url == "https://api.githubcopilot.com/responses" assert claude_url == "https://api.githubcopilot.com/chat/completions" assert openai_url == "https://api.openai.com/v1/chat/completions"