File size: 4,397 Bytes
1b6cc04 f412f5f 1b6cc04 f412f5f 1b6cc04 f412f5f 1b6cc04 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 | import json
import httpx
import pytest
from fastapi.testclient import TestClient
from free_claude_code.core.anthropic.stream_contracts import (
assert_anthropic_stream_contract,
parse_sse_text,
text_content,
)
from free_claude_code.providers.admission import ProviderAdmissionController
from free_claude_code.providers.base import ProviderConfig
from free_claude_code.providers.openai_codex.auth import (
OpenAIAccess,
OpenAIAuthManager,
)
from free_claude_code.providers.openai_codex.provider import OpenAICodexProvider
from tests.api.support import create_test_app
class _FakeOpenAIAuth(OpenAIAuthManager):
async def access(self, *, force_refresh: bool = False) -> OpenAIAccess:
return OpenAIAccess("access", "account", False)
async def recover_unauthorized(self, rejected_token: str) -> OpenAIAccess:
raise AssertionError("The compatibility request must not require auth recovery")
def _complete_stream(text: str) -> str:
events = [
(
"response.created",
{"type": "response.created", "response": {"id": "resp_1"}},
),
(
"response.output_text.delta",
{"type": "response.output_text.delta", "delta": text},
),
(
"response.completed",
{
"type": "response.completed",
"response": {
"id": "resp_1",
"usage": {"input_tokens": 3, "output_tokens": 2},
},
},
),
]
return "".join(
f"event: {event_type}\ndata: {json.dumps(payload)}\n\n"
for event_type, payload in events
)
@pytest.mark.asyncio
async def test_messages_accepts_current_claude_controls_for_openai_provider() -> None:
upstream_requests: list[httpx.Request] = []
def handler(request: httpx.Request) -> httpx.Response:
upstream_requests.append(request)
return httpx.Response(
200,
content=_complete_stream("hello").encode(),
request=request,
)
upstream_client = httpx.AsyncClient(
base_url="https://chatgpt.com/backend-api/codex/",
transport=httpx.MockTransport(handler),
)
provider = OpenAICodexProvider(
ProviderConfig(
api_key="",
base_url="https://chatgpt.com/backend-api/codex",
rate_limit=100,
rate_window=1,
max_concurrency=2,
),
auth=_FakeOpenAIAuth(),
admission=ProviderAdmissionController(
provider_name="openai",
rate_limit=100,
rate_window=1,
max_concurrency=2,
base_delay=0,
max_delay=0,
jitter=0,
),
client=upstream_client,
)
app = create_test_app(providers={"openai": provider})
try:
with TestClient(app) as client:
response = client.post(
"/v1/messages",
json={
"model": "openai/gpt-test",
"max_tokens": 1024,
"messages": [{"role": "user", "content": "hello"}],
"metadata": {"user_id": "example-user"},
"thinking": {"type": "adaptive", "display": "omitted"},
"context_management": {
"edits": [
{
"type": "clear_thinking_20251015",
"keep": "all",
}
]
},
"output_config": {"effort": "high"},
"stream": True,
},
)
finally:
await upstream_client.aclose()
assert response.status_code == 200
assert response.headers["content-type"].startswith("text/event-stream")
events = parse_sse_text(response.text)
assert_anthropic_stream_contract(events)
assert text_content(events) == "hello"
assert len(upstream_requests) == 1
payload = json.loads(upstream_requests[0].content)
assert payload["model"] == "gpt-test"
assert payload["reasoning"] == {"effort": "high", "summary": "auto"}
assert "max_output_tokens" not in payload
assert "metadata" not in payload
assert "context_management" not in payload
assert "output_config" not in payload
|