wehub-resource-sync
424 lines
15 KiB
Python
424 lines
15 KiB
Python
"""Contract tests for the openai_codex (ChatGPT OAuth) provider."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import base64
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from opensquilla.provider.codex_auth import (
|
|
CodexAuthError,
|
|
load_codex_credentials,
|
|
refresh_codex_credentials,
|
|
)
|
|
from opensquilla.provider.openai_codex import OpenAICodexProvider
|
|
from opensquilla.provider.registry import get_provider_spec
|
|
from opensquilla.provider.selector import build_provider
|
|
from opensquilla.provider.types import (
|
|
ChatConfig,
|
|
DoneEvent,
|
|
ErrorEvent,
|
|
Message,
|
|
ReasoningDeltaEvent,
|
|
TextDeltaEvent,
|
|
ToolDefinition,
|
|
ToolInputSchema,
|
|
ToolUseDeltaEvent,
|
|
ToolUseEndEvent,
|
|
ToolUseStartEvent,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _write_auth(path: Path, *, access="tok-access", refresh="tok-refresh", account="acct-1"):
|
|
payload = {
|
|
"auth_mode": "chatgpt",
|
|
"OPENAI_API_KEY": None,
|
|
"tokens": {
|
|
"id_token": "",
|
|
"access_token": access,
|
|
"refresh_token": refresh,
|
|
"account_id": account,
|
|
},
|
|
"last_refresh": "2026-06-29T21:53:00Z",
|
|
}
|
|
path.write_text(json.dumps(payload))
|
|
return path
|
|
|
|
|
|
def _sse(events: list[dict[str, Any]]) -> bytes:
|
|
return b"".join(f"data: {json.dumps(ev)}\n\n".encode() for ev in events)
|
|
|
|
|
|
def _happy_sse() -> bytes:
|
|
return _sse(
|
|
[
|
|
{"type": "response.created", "response": {"id": "resp-1"}},
|
|
{
|
|
"type": "response.output_item.added",
|
|
"item": {
|
|
"type": "function_call",
|
|
"id": "fc-1",
|
|
"call_id": "call-1",
|
|
"name": "search",
|
|
},
|
|
},
|
|
{"type": "response.function_call_arguments.delta", "item_id": "fc-1", "delta": '{"que'},
|
|
{
|
|
"type": "response.function_call_arguments.delta",
|
|
"item_id": "fc-1",
|
|
"delta": 'ry": "x"}',
|
|
},
|
|
{
|
|
"type": "response.output_item.done",
|
|
"item": {
|
|
"type": "function_call",
|
|
"id": "fc-1",
|
|
"call_id": "call-1",
|
|
"name": "search",
|
|
"arguments": '{"query": "x"}',
|
|
},
|
|
},
|
|
{"type": "response.output_text.delta", "delta": "done"},
|
|
{
|
|
"type": "response.completed",
|
|
"response": {
|
|
"id": "resp-1",
|
|
"model": "gpt-5.2-codex",
|
|
"usage": {
|
|
"input_tokens": 40,
|
|
"output_tokens": 9,
|
|
"input_tokens_details": {"cached_tokens": 12},
|
|
"output_tokens_details": {"reasoning_tokens": 3},
|
|
},
|
|
},
|
|
},
|
|
]
|
|
)
|
|
|
|
|
|
def _patch_codex_transport(monkeypatch, handler) -> None:
|
|
transport = httpx.MockTransport(handler)
|
|
real = httpx.AsyncClient
|
|
|
|
def patched(*args: Any, **kwargs: Any) -> httpx.AsyncClient:
|
|
kwargs["transport"] = transport
|
|
return real(*args, **kwargs)
|
|
|
|
monkeypatch.setattr("opensquilla.provider.openai_codex.httpx.AsyncClient", patched)
|
|
|
|
|
|
def _collect(provider: OpenAICodexProvider, *, tools=None, cfg=None):
|
|
async def _run():
|
|
return [
|
|
ev
|
|
async for ev in provider.chat(
|
|
[Message(role="user", content="hi")], tools=tools, config=cfg or ChatConfig()
|
|
)
|
|
]
|
|
|
|
return asyncio.run(_run())
|
|
|
|
|
|
_SEARCH_TOOL = ToolDefinition(
|
|
name="search",
|
|
description="Search things.",
|
|
input_schema=ToolInputSchema(properties={"query": {"type": "string"}}),
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Auth
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_load_credentials_from_auth_file(tmp_path: Path) -> None:
|
|
auth = _write_auth(tmp_path / "auth.json")
|
|
creds = load_codex_credentials(auth)
|
|
assert creds.access_token == "tok-access"
|
|
assert creds.refresh_token == "tok-refresh"
|
|
assert creds.account_id == "acct-1"
|
|
|
|
|
|
def test_missing_auth_file_points_at_codex_login(tmp_path: Path) -> None:
|
|
with pytest.raises(CodexAuthError, match="codex login"):
|
|
load_codex_credentials(tmp_path / "missing.json")
|
|
|
|
|
|
def test_account_id_falls_back_to_jwt_claim(tmp_path: Path) -> None:
|
|
claims = {"https://api.openai.com/auth": {"chatgpt_account_id": "acct-jwt"}}
|
|
payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).decode().rstrip("=")
|
|
jwt = f"h.{payload}.s"
|
|
auth = tmp_path / "auth.json"
|
|
auth.write_text(
|
|
json.dumps({"tokens": {"access_token": "tok", "id_token": jwt, "account_id": None}})
|
|
)
|
|
assert load_codex_credentials(auth).account_id == "acct-jwt"
|
|
|
|
|
|
def test_refresh_persists_new_tokens(tmp_path: Path, monkeypatch) -> None:
|
|
auth = _write_auth(tmp_path / "auth.json")
|
|
creds = load_codex_credentials(auth)
|
|
captured: dict[str, Any] = {}
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
captured["url"] = str(request.url)
|
|
captured["body"] = json.loads(request.content)
|
|
return httpx.Response(
|
|
200,
|
|
json={"access_token": "tok-new", "refresh_token": "refresh-new", "id_token": ""},
|
|
)
|
|
|
|
transport = httpx.MockTransport(handler)
|
|
real = httpx.AsyncClient
|
|
monkeypatch.setattr(
|
|
"opensquilla.provider.codex_auth.httpx.AsyncClient",
|
|
lambda *a, **kw: real(*a, **{**kw, "transport": transport}),
|
|
)
|
|
|
|
updated = asyncio.run(refresh_codex_credentials(creds, path=auth))
|
|
assert updated.access_token == "tok-new"
|
|
assert captured["body"]["grant_type"] == "refresh_token"
|
|
assert captured["body"]["client_id"].startswith("app_")
|
|
on_disk = json.loads(auth.read_text())
|
|
assert on_disk["tokens"]["access_token"] == "tok-new"
|
|
assert on_disk["tokens"]["refresh_token"] == "refresh-new"
|
|
assert on_disk["last_refresh"].endswith("Z")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Provider wire behavior
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_stream_maps_responses_events(tmp_path: Path, monkeypatch) -> None:
|
|
auth = _write_auth(tmp_path / "auth.json")
|
|
captured: dict[str, Any] = {}
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
captured["url"] = str(request.url)
|
|
captured["headers"] = dict(request.headers)
|
|
captured["payload"] = json.loads(request.content)
|
|
return httpx.Response(
|
|
200, headers={"content-type": "text/event-stream"}, content=_happy_sse()
|
|
)
|
|
|
|
_patch_codex_transport(monkeypatch, handler)
|
|
provider = OpenAICodexProvider(
|
|
model="gpt-5.2-codex",
|
|
base_url="https://chatgpt.com", # normalization adds /backend-api
|
|
auth_path=str(auth),
|
|
)
|
|
events = _collect(provider, tools=[_SEARCH_TOOL], cfg=ChatConfig(system="be brief"))
|
|
|
|
assert captured["url"] == "https://chatgpt.com/backend-api/codex/responses"
|
|
assert captured["headers"]["authorization"] == "Bearer tok-access"
|
|
assert captured["headers"]["chatgpt-account-id"] == "acct-1"
|
|
assert captured["headers"]["originator"] == "codex_cli_rs"
|
|
payload = captured["payload"]
|
|
assert payload["instructions"] == "be brief"
|
|
assert payload["store"] is False and payload["stream"] is True
|
|
assert payload["include"] == ["reasoning.encrypted_content"]
|
|
# Protocol fact (verified live): the ChatGPT codex backend rejects
|
|
# max_output_tokens with HTTP 400 "Unsupported parameter" — it must
|
|
# never be sent; subscription turns carry no client-set output cap.
|
|
assert "max_output_tokens" not in payload
|
|
(tool,) = payload["tools"]
|
|
assert tool["type"] == "function" and tool["name"] == "search"
|
|
assert "function" not in tool # flat Responses shape, not chat-completions nesting
|
|
|
|
starts = [e for e in events if isinstance(e, ToolUseStartEvent)]
|
|
deltas = [e for e in events if isinstance(e, ToolUseDeltaEvent)]
|
|
ends = [e for e in events if isinstance(e, ToolUseEndEvent)]
|
|
texts = [e for e in events if isinstance(e, TextDeltaEvent)]
|
|
(done,) = [e for e in events if isinstance(e, DoneEvent)]
|
|
assert [s.tool_use_id for s in starts] == ["call-1"]
|
|
assert [d.json_fragment for d in deltas] == ['{"que', 'ry": "x"}']
|
|
assert ends[0].arguments == {"query": "x"}
|
|
assert [t.text for t in texts] == ["done"]
|
|
assert done.stop_reason == "tool_use"
|
|
assert (done.input_tokens, done.output_tokens) == (40, 9)
|
|
assert done.cached_tokens == 12 and done.reasoning_tokens == 3
|
|
assert done.model == "gpt-5.2-codex"
|
|
|
|
|
|
def test_burst_function_call_without_deltas(tmp_path: Path, monkeypatch) -> None:
|
|
auth = _write_auth(tmp_path / "auth.json")
|
|
body = _sse(
|
|
[
|
|
{
|
|
"type": "response.output_item.done",
|
|
"item": {
|
|
"type": "function_call",
|
|
"id": "fc-9",
|
|
"call_id": "call-9",
|
|
"name": "search",
|
|
"arguments": '{"query": "y"}',
|
|
},
|
|
},
|
|
{"type": "response.completed", "response": {"id": "r", "usage": {}}},
|
|
]
|
|
)
|
|
_patch_codex_transport(
|
|
monkeypatch,
|
|
lambda request: httpx.Response(
|
|
200, headers={"content-type": "text/event-stream"}, content=body
|
|
),
|
|
)
|
|
provider = OpenAICodexProvider(auth_path=str(auth))
|
|
events = _collect(provider, tools=[_SEARCH_TOOL])
|
|
ends = [e for e in events if isinstance(e, ToolUseEndEvent)]
|
|
assert len(ends) == 1 and ends[0].arguments == {"query": "y"}
|
|
assert any(isinstance(e, ToolUseStartEvent) for e in events)
|
|
|
|
|
|
def test_reasoning_deltas_reach_done_event(tmp_path: Path, monkeypatch) -> None:
|
|
auth = _write_auth(tmp_path / "auth.json")
|
|
body = _sse(
|
|
[
|
|
{
|
|
"type": "response.reasoning_summary_text.delta",
|
|
"delta": "think ",
|
|
"summary_index": 0,
|
|
},
|
|
{"type": "response.reasoning_text.delta", "delta": "hard", "content_index": 0},
|
|
{"type": "response.output_text.delta", "delta": "hi"},
|
|
{"type": "response.completed", "response": {"id": "r", "usage": {}}},
|
|
]
|
|
)
|
|
_patch_codex_transport(
|
|
monkeypatch,
|
|
lambda request: httpx.Response(
|
|
200, headers={"content-type": "text/event-stream"}, content=body
|
|
),
|
|
)
|
|
provider = OpenAICodexProvider(auth_path=str(auth))
|
|
events = _collect(provider)
|
|
assert [e.text for e in events if isinstance(e, ReasoningDeltaEvent)] == ["think ", "hard"]
|
|
(done,) = [e for e in events if isinstance(e, DoneEvent)]
|
|
assert done.reasoning_content == "think hard"
|
|
|
|
|
|
def test_401_refreshes_and_retries_once(tmp_path: Path, monkeypatch) -> None:
|
|
auth = _write_auth(tmp_path / "auth.json")
|
|
calls = {"stream": 0, "refresh": 0}
|
|
|
|
# One handler for both endpoints: the provider and the auth module share
|
|
# the global httpx module, so separate transports would fight over it.
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
if "auth.openai.com" in str(request.url):
|
|
calls["refresh"] += 1
|
|
return httpx.Response(200, json={"access_token": "tok-new"})
|
|
calls["stream"] += 1
|
|
if request.headers.get("authorization") == "Bearer tok-access":
|
|
return httpx.Response(401, json={"error": {"message": "expired"}})
|
|
return httpx.Response(
|
|
200,
|
|
headers={"content-type": "text/event-stream"},
|
|
content=_sse(
|
|
[
|
|
{"type": "response.output_text.delta", "delta": "ok"},
|
|
{"type": "response.completed", "response": {"id": "r", "usage": {}}},
|
|
]
|
|
),
|
|
)
|
|
|
|
_patch_codex_transport(monkeypatch, handler)
|
|
transport = httpx.MockTransport(handler)
|
|
real = httpx.AsyncClient
|
|
monkeypatch.setattr(
|
|
"opensquilla.provider.codex_auth.httpx.AsyncClient",
|
|
lambda *a, **kw: real(*a, **{**kw, "transport": transport}),
|
|
)
|
|
|
|
provider = OpenAICodexProvider(auth_path=str(auth))
|
|
events = _collect(provider)
|
|
assert calls["stream"] == 2
|
|
assert calls["refresh"] == 1
|
|
assert any(isinstance(e, TextDeltaEvent) and e.text == "ok" for e in events)
|
|
assert any(isinstance(e, DoneEvent) for e in events)
|
|
assert json.loads(auth.read_text())["tokens"]["access_token"] == "tok-new"
|
|
|
|
|
|
def test_response_failed_yields_error_event(tmp_path: Path, monkeypatch) -> None:
|
|
auth = _write_auth(tmp_path / "auth.json")
|
|
body = _sse(
|
|
[
|
|
{
|
|
"type": "response.failed",
|
|
"response": {"error": {"code": "usage_limit_reached", "message": "limit"}},
|
|
}
|
|
]
|
|
)
|
|
_patch_codex_transport(
|
|
monkeypatch,
|
|
lambda request: httpx.Response(
|
|
200, headers={"content-type": "text/event-stream"}, content=body
|
|
),
|
|
)
|
|
provider = OpenAICodexProvider(auth_path=str(auth))
|
|
events = _collect(provider)
|
|
(error,) = [e for e in events if isinstance(e, ErrorEvent)]
|
|
assert error.code == "usage_limit_reached"
|
|
assert not any(isinstance(e, DoneEvent) for e in events)
|
|
|
|
|
|
def test_truncated_stream_still_terminates_with_done(tmp_path: Path, monkeypatch) -> None:
|
|
auth = _write_auth(tmp_path / "auth.json")
|
|
body = _sse(
|
|
[
|
|
{
|
|
"type": "response.output_item.added",
|
|
"item": {
|
|
"type": "function_call",
|
|
"id": "fc-1",
|
|
"call_id": "call-1",
|
|
"name": "search",
|
|
},
|
|
},
|
|
{"type": "response.function_call_arguments.delta", "item_id": "fc-1", "delta": "{}"},
|
|
# stream drops before output_item.done / response.completed
|
|
]
|
|
)
|
|
_patch_codex_transport(
|
|
monkeypatch,
|
|
lambda request: httpx.Response(
|
|
200, headers={"content-type": "text/event-stream"}, content=body
|
|
),
|
|
)
|
|
provider = OpenAICodexProvider(auth_path=str(auth))
|
|
events = _collect(provider, tools=[_SEARCH_TOOL])
|
|
assert any(isinstance(e, ToolUseEndEvent) for e in events)
|
|
assert any(isinstance(e, DoneEvent) for e in events)
|
|
|
|
|
|
def test_missing_credentials_is_auth_error_event(tmp_path: Path) -> None:
|
|
provider = OpenAICodexProvider(auth_path=str(tmp_path / "missing.json"))
|
|
events = _collect(provider)
|
|
(error,) = [e for e in events if isinstance(e, ErrorEvent)]
|
|
assert error.code == "401"
|
|
assert "codex login" in error.message
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Registry / selector wiring
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_registry_and_selector_wire_openai_codex() -> None:
|
|
spec = get_provider_spec("openai_codex")
|
|
assert spec.backend == "openai_codex"
|
|
assert spec.runtime_supported is True
|
|
assert spec.requires_api_key() is False # OAuth, not an API key field
|
|
assert spec.requires_base_url() is False
|
|
provider = build_provider("openai_codex", "gpt-5.2-codex")
|
|
assert isinstance(provider, OpenAICodexProvider)
|