| 1 | from __future__ import annotations |
| 2 | |
| 3 | import json |
| 4 | import multiprocessing |
| 5 | import queue |
| 6 | import stat |
| 7 | import sys |
| 8 | import threading |
| 9 | import time |
| 10 | from pathlib import Path |
| 11 | |
| 12 | import pytest |
| 13 | import yaml |
| 14 | |
| 15 | sys.path.insert(0, str(Path(__file__).resolve().parents[1])) |
| 16 | from plugins._oauth.helpers import codex |
| 17 | from plugins._oauth.helpers import routes |
| 18 | from plugins._oauth.helpers.providers import codex as codex_provider |
| 19 | from plugins._oauth.extensions.python._functions.models.get_api_key.end import ( |
| 20 | _20_oauth_account_dummy_key as oauth_dummy_key, |
| 21 | ) |
| 22 | |
| 23 | |
| 24 | @pytest.fixture(autouse=True) |
| 25 | def use_temporary_auth_locks(tmp_path, monkeypatch): |
| 26 | def lock_path(path: Path) -> Path: |
| 27 | digest = codex.hashlib.sha256(codex._path_key(path).encode("utf-8")).hexdigest() |
| 28 | return tmp_path / "locks" / f"{digest}.lock" |
| 29 | |
| 30 | monkeypatch.setattr(codex, "_auth_lock_path", lock_path) |
| 31 | |
| 32 | |
| 33 | def test_generate_pkce_produces_urlsafe_verifier_and_challenge(): |
| 34 | pair = codex.generate_pkce() |
| 35 | |
| 36 | assert 43 <= len(pair.verifier) <= 128 |
| 37 | assert pair.verifier |
| 38 | assert pair.challenge |
| 39 | assert "=" not in pair.verifier |
| 40 | assert "=" not in pair.challenge |
| 41 | |
| 42 | |
| 43 | def test_build_authorize_url_uses_existing_a0_origin_callback(monkeypatch): |
| 44 | monkeypatch.setattr( |
| 45 | codex, |
| 46 | "codex_config", |
| 47 | lambda: { |
| 48 | "issuer": "https://auth.openai.com", |
| 49 | "client_id": "app_EMoamEEZ73f0CkXaXp7hrann", |
| 50 | "scopes": [ |
| 51 | "openid", |
| 52 | "profile", |
| 53 | "email", |
| 54 | "offline_access", |
| 55 | "api.connectors.read", |
| 56 | "api.connectors.invoke", |
| 57 | ], |
| 58 | "forced_workspace_id": "", |
| 59 | }, |
| 60 | ) |
| 61 | pair = codex.PkcePair(verifier="verifier", challenge="challenge") |
| 62 | auth_url = codex.build_authorize_url( |
| 63 | "http://localhost:50001/auth/callback", |
| 64 | "state", |
| 65 | pair, |
| 66 | ) |
| 67 | |
| 68 | assert auth_url.startswith("https://auth.openai.com/oauth/authorize?") |
| 69 | assert "redirect_uri=http%3A%2F%2Flocalhost%3A50001%2Fauth%2Fcallback" in auth_url |
| 70 | assert "code_challenge=challenge" in auth_url |
| 71 | assert "originator=codex_cli_rs" in auth_url |
| 72 | |
| 73 | |
| 74 | def test_chat_messages_to_response_body_extracts_instructions(): |
| 75 | body = codex.chat_messages_to_response_body( |
| 76 | { |
| 77 | "model": "gpt-5.2", |
| 78 | "messages": [ |
| 79 | {"role": "system", "content": "Be precise."}, |
| 80 | {"role": "user", "content": "Hello"}, |
| 81 | ], |
| 82 | "temperature": 0.2, |
| 83 | "reasoning_effort": "high", |
| 84 | } |
| 85 | ) |
| 86 | |
| 87 | assert body["model"] == "gpt-5.2" |
| 88 | assert body["instructions"] == "Be precise." |
| 89 | assert body["input"] == [{"role": "user", "content": "Hello"}] |
| 90 | assert body["temperature"] == 0.2 |
| 91 | assert body["reasoning"] == {"effort": "high"} |
| 92 | |
| 93 | |
| 94 | def test_chat_messages_to_response_body_uses_current_codex_default_model(): |
| 95 | body = codex.chat_messages_to_response_body( |
| 96 | { |
| 97 | "messages": [ |
| 98 | {"role": "user", "content": "Hello"}, |
| 99 | ], |
| 100 | } |
| 101 | ) |
| 102 | |
| 103 | assert body["model"] == "gpt-5.5" |
| 104 | |
| 105 | |
| 106 | def test_codex_provider_metadata_uses_upstream_models_only_by_default(monkeypatch): |
| 107 | monkeypatch.setattr( |
| 108 | codex_provider, |
| 109 | "_codex_config", |
| 110 | lambda: { |
| 111 | "models": [], |
| 112 | "proxy_base_path": "/oauth/codex", |
| 113 | "callback_path": "/auth/callback", |
| 114 | }, |
| 115 | ) |
| 116 | |
| 117 | metadata = codex_provider.CodexOAuthProvider().metadata() |
| 118 | |
| 119 | assert metadata.default_model == "gpt-5.5" |
| 120 | assert metadata.default_models == [] |
| 121 | |
| 122 | |
| 123 | def test_codex_fetch_models_omits_fake_client_version(monkeypatch): |
| 124 | calls = [] |
| 125 | |
| 126 | class FakeResponse: |
| 127 | ok = True |
| 128 | |
| 129 | def json(self): |
| 130 | return {"models": [{"slug": "upstream-model"}]} |
| 131 | |
| 132 | monkeypatch.setattr(codex, "codex_config", lambda: {"models": []}) |
| 133 | monkeypatch.setattr(codex, "resolve_codex_version", lambda: "") |
| 134 | monkeypatch.setattr(codex, "request_codex", lambda path, params=None: calls.append((path, params)) or FakeResponse()) |
| 135 | |
| 136 | assert codex.fetch_models() == ["upstream-model"] |
| 137 | assert calls == [("/models", None)] |
| 138 | |
| 139 | |
| 140 | def test_codex_fetch_model_catalog_preserves_model_metadata(monkeypatch): |
| 141 | class FakeResponse: |
| 142 | ok = True |
| 143 | |
| 144 | def json(self): |
| 145 | return { |
| 146 | "models": [ |
| 147 | { |
| 148 | "slug": "gpt-5.5", |
| 149 | "display_name": "GPT-5.5", |
| 150 | "description": "Frontier coding model.", |
| 151 | "default_reasoning_level": "medium", |
| 152 | "base_instructions": "too large for the settings UI", |
| 153 | } |
| 154 | ] |
| 155 | } |
| 156 | |
| 157 | monkeypatch.setattr(codex, "codex_config", lambda: {"models": []}) |
| 158 | monkeypatch.setattr(codex, "resolve_codex_version", lambda: "0.142.5") |
| 159 | monkeypatch.setattr(codex, "request_codex", lambda path, params=None: FakeResponse()) |
| 160 | |
| 161 | assert codex.fetch_model_catalog() == [ |
| 162 | { |
| 163 | "slug": "gpt-5.5", |
| 164 | "id": "gpt-5.5", |
| 165 | "display_name": "GPT-5.5", |
| 166 | "description": "Frontier coding model.", |
| 167 | "default_reasoning_level": "medium", |
| 168 | } |
| 169 | ] |
| 170 | |
| 171 | |
| 172 | def test_prepare_responses_body_adds_codex_client_metadata(monkeypatch): |
| 173 | monkeypatch.setattr( |
| 174 | codex, |
| 175 | "build_client_metadata", |
| 176 | lambda: { |
| 177 | "x-codex-installation-id": "install-1", |
| 178 | "session_id": "session-1", |
| 179 | "thread_id": "thread-1", |
| 180 | "x-codex-window-id": "agent-zero", |
| 181 | }, |
| 182 | ) |
| 183 | |
| 184 | body = codex.prepare_responses_body( |
| 185 | { |
| 186 | "model": "gpt-5.5", |
| 187 | "input": "hello", |
| 188 | "client_metadata": { |
| 189 | "caller": "plugin-test", |
| 190 | "x-codex-installation-id": "stale", |
| 191 | }, |
| 192 | "reasoning": {"effort": "medium"}, |
| 193 | "include": ["output_text"], |
| 194 | }, |
| 195 | force_stream=True, |
| 196 | ) |
| 197 | |
| 198 | assert body["client_metadata"] == { |
| 199 | "caller": "plugin-test", |
| 200 | "x-codex-installation-id": "install-1", |
| 201 | "session_id": "session-1", |
| 202 | "thread_id": "thread-1", |
| 203 | "x-codex-window-id": "agent-zero", |
| 204 | } |
| 205 | assert body["input"] == [{"role": "user", "content": "hello"}] |
| 206 | assert body["stream"] is True |
| 207 | assert body["reasoning"] == {"effort": "medium", "summary": "auto"} |
| 208 | assert body["include"] == ["output_text", "reasoning.encrypted_content"] |
| 209 | |
| 210 | |
| 211 | def test_prepare_responses_body_tightens_existing_response_tool_only(monkeypatch): |
| 212 | monkeypatch.setattr(codex, "build_client_metadata", lambda: {}) |
| 213 | response_tool = { |
| 214 | "type": "function", |
| 215 | "name": "response", |
| 216 | "description": "final answer", |
| 217 | "parameters": {"type": "object", "additionalProperties": True}, |
| 218 | } |
| 219 | other_tool = { |
| 220 | "type": "function", |
| 221 | "name": "search", |
| 222 | "parameters": {"type": "object", "additionalProperties": True}, |
| 223 | } |
| 224 | |
| 225 | body = codex.prepare_responses_body( |
| 226 | {"input": [], "tools": [response_tool, other_tool]}, |
| 227 | force_stream=True, |
| 228 | ) |
| 229 | |
| 230 | assert body["tools"] == [ |
| 231 | { |
| 232 | **response_tool, |
| 233 | "strict": True, |
| 234 | "parameters": { |
| 235 | "type": "object", |
| 236 | "properties": {"text": {"type": "string"}}, |
| 237 | "required": ["text"], |
| 238 | "additionalProperties": False, |
| 239 | }, |
| 240 | }, |
| 241 | other_tool, |
| 242 | ] |
| 243 | assert codex.prepare_responses_body( |
| 244 | {"input": [], "tools": [other_tool]}, |
| 245 | force_stream=True, |
| 246 | )["tools"] == [other_tool] |
| 247 | |
| 248 | |
| 249 | @pytest.mark.parametrize( |
| 250 | ("request_reasoning", "expected"), |
| 251 | [ |
| 252 | ({"reasoning_effort": "xhigh"}, {"effort": "xhigh", "summary": "auto"}), |
| 253 | ( |
| 254 | {"reasoning": {"effort": "medium"}, "reasoning_effort": "xhigh"}, |
| 255 | {"effort": "medium", "summary": "auto"}, |
| 256 | ), |
| 257 | ( |
| 258 | {"reasoning": {"effort": "medium", "summary": "detailed"}}, |
| 259 | {"effort": "medium", "summary": "detailed"}, |
| 260 | ), |
| 261 | ], |
| 262 | ) |
| 263 | def test_prepare_responses_body_normalizes_reasoning_effort( |
| 264 | monkeypatch, request_reasoning, expected |
| 265 | ): |
| 266 | monkeypatch.setattr(codex, "build_client_metadata", lambda: {}) |
| 267 | |
| 268 | body = codex.prepare_responses_body( |
| 269 | {"model": "gpt-5.5", "input": "hello", **request_reasoning}, |
| 270 | force_stream=True, |
| 271 | ) |
| 272 | |
| 273 | assert body["reasoning"] == expected |
| 274 | assert "reasoning_effort" not in body |
| 275 | |
| 276 | |
| 277 | def test_prepare_responses_body_applies_codex_response_defaults(monkeypatch): |
| 278 | monkeypatch.setattr(codex, "build_client_metadata", lambda: {}) |
| 279 | monkeypatch.setattr( |
| 280 | codex, |
| 281 | "codex_config", |
| 282 | lambda: { |
| 283 | "reasoning_effort": "high", |
| 284 | "reasoning_summary": "concise", |
| 285 | "text_verbosity": "low", |
| 286 | }, |
| 287 | ) |
| 288 | |
| 289 | defaults = codex.prepare_responses_body( |
| 290 | {"model": "gpt-5.5", "input": "hello"}, force_stream=True |
| 291 | ) |
| 292 | overrides = codex.prepare_responses_body( |
| 293 | { |
| 294 | "model": "gpt-5.5", |
| 295 | "input": "hello", |
| 296 | "reasoning": {"effort": "low", "summary": "detailed"}, |
| 297 | "text": {"verbosity": "high"}, |
| 298 | }, |
| 299 | force_stream=True, |
| 300 | ) |
| 301 | |
| 302 | assert defaults["reasoning"] == {"effort": "high", "summary": "concise"} |
| 303 | assert defaults["text"] == {"verbosity": "low"} |
| 304 | assert overrides["reasoning"] == {"effort": "low", "summary": "detailed"} |
| 305 | assert overrides["text"] == {"verbosity": "high"} |
| 306 | |
| 307 | |
| 308 | def test_codex_config_validates_response_defaults(): |
| 309 | from plugins._oauth.helpers.config import codex_config |
| 310 | |
| 311 | config = codex_config( |
| 312 | { |
| 313 | "codex": { |
| 314 | "reasoning_effort": "invalid", |
| 315 | "reasoning_summary": "DETAILED", |
| 316 | "text_verbosity": "low", |
| 317 | } |
| 318 | } |
| 319 | ) |
| 320 | |
| 321 | assert config["reasoning_effort"] == "high" |
| 322 | assert config["reasoning_summary"] == "detailed" |
| 323 | assert config["text_verbosity"] == "low" |
| 324 | |
| 325 | |
| 326 | def test_prepare_responses_body_sends_empty_continuation_input_as_list(monkeypatch): |
| 327 | monkeypatch.setattr(codex, "build_client_metadata", lambda: {}) |
| 328 | |
| 329 | body = codex.prepare_responses_body( |
| 330 | { |
| 331 | "model": "gpt-5.5", |
| 332 | "input": "", |
| 333 | "previous_response_id": "resp_1", |
| 334 | }, |
| 335 | force_stream=True, |
| 336 | ) |
| 337 | |
| 338 | assert body["input"] == [] |
| 339 | assert body["previous_response_id"] == "resp_1" |
| 340 | |
| 341 | |
| 342 | @pytest.mark.parametrize("client_version", ["0.142.5", ""]) |
| 343 | def test_request_codex_sends_current_codex_headers_from_body( |
| 344 | monkeypatch, client_version |
| 345 | ): |
| 346 | calls = [] |
| 347 | body = json.dumps( |
| 348 | { |
| 349 | "client_metadata": { |
| 350 | "x-codex-installation-id": "install-1", |
| 351 | "session_id": "session-1", |
| 352 | "thread_id": "thread-1", |
| 353 | "x-codex-window-id": "agent-zero", |
| 354 | } |
| 355 | } |
| 356 | ) |
| 357 | |
| 358 | class FakeResponse: |
| 359 | ok = True |
| 360 | |
| 361 | monkeypatch.setattr( |
| 362 | codex, |
| 363 | "codex_config", |
| 364 | lambda: { |
| 365 | "upstream_base_url": "https://chatgpt.example/backend-api/codex", |
| 366 | "request_timeout_seconds": 120, |
| 367 | }, |
| 368 | ) |
| 369 | monkeypatch.setattr( |
| 370 | codex, |
| 371 | "load_auth", |
| 372 | lambda: codex.EffectiveAuth( |
| 373 | access_token="access-token", |
| 374 | account_id="account-1", |
| 375 | ), |
| 376 | ) |
| 377 | monkeypatch.setattr(codex, "resolve_codex_version", lambda: client_version) |
| 378 | |
| 379 | def fake_request(method, target, headers, data, params, timeout, stream): |
| 380 | calls.append( |
| 381 | { |
| 382 | "method": method, |
| 383 | "target": target, |
| 384 | "headers": headers, |
| 385 | "data": data, |
| 386 | "params": params, |
| 387 | "timeout": timeout, |
| 388 | "stream": stream, |
| 389 | } |
| 390 | ) |
| 391 | return FakeResponse() |
| 392 | |
| 393 | monkeypatch.setattr(codex.requests, "request", fake_request) |
| 394 | |
| 395 | response = codex.request_codex( |
| 396 | "/responses", |
| 397 | method="POST", |
| 398 | headers={"Content-Type": "application/json"}, |
| 399 | body=body, |
| 400 | stream=True, |
| 401 | ) |
| 402 | |
| 403 | assert response.ok is True |
| 404 | call = calls[0] |
| 405 | assert call["target"] == "https://chatgpt.example/backend-api/codex/responses" |
| 406 | assert call["headers"]["Authorization"] == "Bearer access-token" |
| 407 | assert call["headers"]["chatgpt-account-id"] == "account-1" |
| 408 | assert call["headers"]["OpenAI-Beta"] == "responses=experimental" |
| 409 | assert call["headers"]["originator"] == "codex_cli_rs" |
| 410 | assert call["headers"]["x-codex-installation-id"] == "install-1" |
| 411 | assert call["headers"]["session-id"] == "session-1" |
| 412 | assert call["headers"]["thread-id"] == "thread-1" |
| 413 | assert call["headers"]["x-codex-window-id"] == "agent-zero" |
| 414 | if client_version: |
| 415 | assert call["headers"]["version"] == "0.142.5" |
| 416 | else: |
| 417 | assert "version" not in call["headers"] |
| 418 | |
| 419 | |
| 420 | def test_chat_messages_to_response_body_preserves_image_parts_for_responses(): |
| 421 | data_url = "data:image/png;base64,abcd" |
| 422 | |
| 423 | body = codex.chat_messages_to_response_body( |
| 424 | { |
| 425 | "model": "gpt-5.5", |
| 426 | "messages": [ |
| 427 | { |
| 428 | "role": "user", |
| 429 | "content": [ |
| 430 | {"type": "text", "text": "Inspect this screenshot."}, |
| 431 | {"type": "text", "text": "Fresh screen attached."}, |
| 432 | {"type": "image_url", "image_url": {"url": data_url}}, |
| 433 | ], |
| 434 | } |
| 435 | ], |
| 436 | } |
| 437 | ) |
| 438 | |
| 439 | assert body["input"] == [ |
| 440 | { |
| 441 | "role": "user", |
| 442 | "content": [ |
| 443 | {"type": "input_text", "text": "Inspect this screenshot."}, |
| 444 | {"type": "input_text", "text": "Fresh screen attached."}, |
| 445 | {"type": "input_image", "image_url": data_url, "detail": "auto"}, |
| 446 | ], |
| 447 | } |
| 448 | ] |
| 449 | |
| 450 | |
| 451 | def test_chat_messages_to_response_body_keeps_text_only_lists_as_text(): |
| 452 | body = codex.chat_messages_to_response_body( |
| 453 | { |
| 454 | "model": "gpt-5.5", |
| 455 | "messages": [ |
| 456 | { |
| 457 | "role": "user", |
| 458 | "content": [ |
| 459 | {"type": "text", "text": "first"}, |
| 460 | {"type": "text", "text": "second"}, |
| 461 | ], |
| 462 | } |
| 463 | ], |
| 464 | } |
| 465 | ) |
| 466 | |
| 467 | assert body["input"] == [{"role": "user", "content": "first\nsecond"}] |
| 468 | |
| 469 | |
| 470 | def test_response_text_reads_output_text_or_output_blocks(): |
| 471 | assert codex.response_text({"output_text": "direct"}) == "direct" |
| 472 | |
| 473 | assert ( |
| 474 | codex.response_text( |
| 475 | { |
| 476 | "output": [ |
| 477 | { |
| 478 | "content": [ |
| 479 | {"type": "output_text", "text": "a"}, |
| 480 | {"type": "output_text", "text": "b"}, |
| 481 | ] |
| 482 | } |
| 483 | ] |
| 484 | } |
| 485 | ) |
| 486 | == "ab" |
| 487 | ) |
| 488 | |
| 489 | |
| 490 | def test_parse_sse_block_joins_data_lines(): |
| 491 | event = codex.parse_sse_block( |
| 492 | 'event: response.completed\ndata: {"response":\ndata: {"id":"r"}}\n' |
| 493 | ) |
| 494 | |
| 495 | assert event["event"] == "response.completed" |
| 496 | assert json.loads(event["data"]) == {"response": {"id": "r"}} |
| 497 | |
| 498 | |
| 499 | def test_extract_sse_text_deltas_reads_chat_completion_chunks(): |
| 500 | deltas = codex.extract_sse_text_deltas( |
| 501 | { |
| 502 | "id": "chatcmpl_test", |
| 503 | "choices": [ |
| 504 | {"delta": {"role": "assistant"}}, |
| 505 | {"delta": {"content": "Hel"}}, |
| 506 | {"delta": {"content": "lo"}}, |
| 507 | ], |
| 508 | } |
| 509 | ) |
| 510 | |
| 511 | assert deltas == ["Hel", "lo"] |
| 512 | |
| 513 | |
| 514 | def test_extract_sse_text_deltas_ignores_final_done_text(): |
| 515 | assert ( |
| 516 | codex.extract_sse_text_deltas( |
| 517 | {"type": "response.output_text.done", "text": "Hello"}, |
| 518 | "response.output_text.done", |
| 519 | ) |
| 520 | == [] |
| 521 | ) |
| 522 | |
| 523 | |
| 524 | def test_collect_completed_response_restores_native_output_items(): |
| 525 | item = { |
| 526 | "id": "msg_1", |
| 527 | "type": "message", |
| 528 | "status": "completed", |
| 529 | "content": [ |
| 530 | { |
| 531 | "type": "output_text", |
| 532 | "annotations": [], |
| 533 | "logprobs": [], |
| 534 | "text": "Hello", |
| 535 | } |
| 536 | ], |
| 537 | "role": "assistant", |
| 538 | } |
| 539 | |
| 540 | class FakeResponse: |
| 541 | encoding = "utf-8" |
| 542 | |
| 543 | def iter_content(self, chunk_size=8192, decode_unicode=True): |
| 544 | del chunk_size, decode_unicode |
| 545 | yield ( |
| 546 | 'data: {"type":"response.output_item.done","output_index":0,' |
| 547 | f'"item":{json.dumps(item)}}}\n\n' |
| 548 | ).encode() |
| 549 | yield ( |
| 550 | b'data: {"type":"response.completed",' |
| 551 | b'"response":{"id":"resp_1","output":[]}}\n\n' |
| 552 | ) |
| 553 | |
| 554 | assert codex.collect_completed_response(FakeResponse()) == { |
| 555 | "id": "resp_1", |
| 556 | "output": [item], |
| 557 | } |
| 558 | |
| 559 | |
| 560 | def test_collect_completed_response_falls_back_to_text_deltas(): |
| 561 | class FakeResponse: |
| 562 | encoding = "utf-8" |
| 563 | |
| 564 | def iter_content(self, chunk_size=8192, decode_unicode=True): |
| 565 | del chunk_size, decode_unicode |
| 566 | yield b'data: {"choices":[{"delta":{"content":"Hel"}}]}\n\n' |
| 567 | yield b'data: {"choices":[{"delta":{"content":"lo"}}]}\n\n' |
| 568 | yield b'event: response.completed\ndata: {"response":{"output":[]}}\n\n' |
| 569 | yield b"data: [DONE]\n\n" |
| 570 | |
| 571 | assert codex.collect_completed_response(FakeResponse()) == {"output": [], "output_text": "Hello"} |
| 572 | |
| 573 | |
| 574 | def test_normalize_usage_payload_reads_codex_windows(): |
| 575 | usage = codex.normalize_usage_payload( |
| 576 | { |
| 577 | "plan_type": "plus", |
| 578 | "rate_limit": { |
| 579 | "primary_window": { |
| 580 | "used_percent": 39, |
| 581 | "reset_at": 1_738_300_000, |
| 582 | "limit_window_seconds": 18_000, |
| 583 | }, |
| 584 | "secondary_window": { |
| 585 | "used_percent": 15, |
| 586 | "reset_at": 1_738_900_000, |
| 587 | "limit_window_seconds": 604_800, |
| 588 | }, |
| 589 | }, |
| 590 | "credits": {"has_credits": True, "unlimited": False, "balance": 5.39}, |
| 591 | } |
| 592 | ) |
| 593 | |
| 594 | assert usage["available"] is True |
| 595 | assert usage["plan_type"] == "plus" |
| 596 | assert usage["primary"]["used_percent"] == 39 |
| 597 | assert usage["primary"]["remaining_percent"] == 61 |
| 598 | assert usage["primary"]["label"] == "5h" |
| 599 | assert usage["secondary"]["used_percent"] == 15 |
| 600 | assert usage["secondary"]["label"] == "7d" |
| 601 | assert usage["credits"]["balance"] == 5.39 |
| 602 | |
| 603 | |
| 604 | def test_normalize_usage_payload_accepts_zero_percent_headers(): |
| 605 | usage = codex.normalize_usage_payload( |
| 606 | {}, |
| 607 | { |
| 608 | "x-codex-primary-used-percent": "0", |
| 609 | "x-codex-primary-window-minutes": "300", |
| 610 | }, |
| 611 | ) |
| 612 | |
| 613 | assert usage["available"] is True |
| 614 | assert usage["primary"]["used_percent"] == 0 |
| 615 | assert usage["primary"]["remaining_percent"] == 100 |
| 616 | assert usage["primary"]["label"] == "5h" |
| 617 | |
| 618 | |
| 619 | def test_token_error_message_prefers_description(): |
| 620 | class FakeResponse: |
| 621 | status_code = 400 |
| 622 | text = '{"error":"invalid_grant","error_description":"refresh token was already used"}' |
| 623 | |
| 624 | @staticmethod |
| 625 | def json(): |
| 626 | return { |
| 627 | "error": "invalid_grant", |
| 628 | "error_description": "refresh token was already used", |
| 629 | } |
| 630 | |
| 631 | assert codex._token_error_message(FakeResponse()) == "refresh token was already used" |
| 632 | |
| 633 | |
| 634 | def test_refresh_tokens_sends_agent_zero_user_agent(monkeypatch): |
| 635 | requests: list[dict] = [] |
| 636 | |
| 637 | class FakeResponse: |
| 638 | ok = True |
| 639 | |
| 640 | @staticmethod |
| 641 | def json(): |
| 642 | return { |
| 643 | "access_token": "access-1", |
| 644 | "refresh_token": "refresh-1", |
| 645 | } |
| 646 | |
| 647 | def post(url, *, headers, json, timeout): |
| 648 | requests.append( |
| 649 | { |
| 650 | "url": url, |
| 651 | "headers": headers, |
| 652 | "json": json, |
| 653 | "timeout": timeout, |
| 654 | } |
| 655 | ) |
| 656 | return FakeResponse() |
| 657 | |
| 658 | monkeypatch.setattr( |
| 659 | codex, |
| 660 | "codex_config", |
| 661 | lambda: {"token_url": "https://auth.example/oauth/token", "client_id": "client"}, |
| 662 | ) |
| 663 | monkeypatch.setattr(codex, "resolve_agent_zero_user_agent", lambda: "agent-zero/v1.18") |
| 664 | monkeypatch.setattr(codex.requests, "post", post) |
| 665 | |
| 666 | assert codex.refresh_tokens("refresh-0") == { |
| 667 | "id_token": "", |
| 668 | "access_token": "access-1", |
| 669 | "refresh_token": "refresh-1", |
| 670 | } |
| 671 | assert requests == [ |
| 672 | { |
| 673 | "url": "https://auth.example/oauth/token", |
| 674 | "headers": { |
| 675 | "Content-Type": "application/json", |
| 676 | "User-Agent": "agent-zero/v1.18", |
| 677 | }, |
| 678 | "json": { |
| 679 | "client_id": "client", |
| 680 | "grant_type": "refresh_token", |
| 681 | "refresh_token": "refresh-0", |
| 682 | }, |
| 683 | "timeout": 30, |
| 684 | } |
| 685 | ] |
| 686 | |
| 687 | |
| 688 | def test_default_auth_file_ignores_codex_cli_credentials(tmp_path, monkeypatch): |
| 689 | shared_auth = tmp_path / ".codex" / "auth.json" |
| 690 | private_auth = tmp_path / "usr" / "plugins" / "_oauth" / "codex" / "auth.json" |
| 691 | shared_auth.parent.mkdir() |
| 692 | shared_auth.write_text(json.dumps({"tokens": {"refresh_token": "shared"}}), encoding="utf-8") |
| 693 | monkeypatch.setenv("HOME", str(tmp_path)) |
| 694 | monkeypatch.setattr(codex, "codex_config", lambda: {"auth_file_path": ""}) |
| 695 | monkeypatch.setattr(codex.files, "get_abs_path", lambda *parts: str(tmp_path.joinpath(*parts))) |
| 696 | |
| 697 | path, data = codex.read_auth_file() |
| 698 | |
| 699 | assert codex.resolve_auth_file_candidates() == [private_auth] |
| 700 | assert path == private_auth |
| 701 | assert data == {} |
| 702 | |
| 703 | |
| 704 | def test_explicit_codex_cli_auth_path_is_rejected(tmp_path, monkeypatch): |
| 705 | codex_home = tmp_path / "codex-home" |
| 706 | monkeypatch.setenv("CODEX_HOME", str(codex_home)) |
| 707 | monkeypatch.setattr(codex, "codex_config", lambda: {"auth_file_path": str(codex_home / "auth.json")}) |
| 708 | |
| 709 | with pytest.raises(RuntimeError, match="Agent Zero-owned auth file"): |
| 710 | codex.resolve_auth_write_path() |
| 711 | |
| 712 | |
| 713 | def test_explicit_codex_cli_auth_hard_link_is_rejected(tmp_path, monkeypatch): |
| 714 | shared_auth = tmp_path / ".codex" / "auth.json" |
| 715 | shared_auth.parent.mkdir() |
| 716 | shared_auth.write_text(json.dumps({"tokens": {"refresh_token": "shared"}}), encoding="utf-8") |
| 717 | alias = tmp_path / "agent-zero-auth.json" |
| 718 | alias.hardlink_to(shared_auth) |
| 719 | monkeypatch.setenv("HOME", str(tmp_path)) |
| 720 | monkeypatch.setattr(codex, "codex_config", lambda: {"auth_file_path": str(alias)}) |
| 721 | |
| 722 | with pytest.raises(RuntimeError, match="Agent Zero-owned auth file"): |
| 723 | codex.resolve_auth_write_path() |
| 724 | |
| 725 | |
| 726 | def test_explicit_private_auth_hard_link_is_rejected(tmp_path, monkeypatch): |
| 727 | private_auth = tmp_path / "private-auth.json" |
| 728 | private_auth.write_text(json.dumps({"tokens": {"refresh_token": "private"}}), encoding="utf-8") |
| 729 | alias = tmp_path / "agent-zero-auth.json" |
| 730 | alias.hardlink_to(private_auth) |
| 731 | monkeypatch.setattr(codex, "codex_config", lambda: {"auth_file_path": str(alias)}) |
| 732 | |
| 733 | with pytest.raises(RuntimeError, match="Agent Zero-owned auth file"): |
| 734 | codex.resolve_auth_write_path() |
| 735 | |
| 736 | |
| 737 | def test_write_auth_file_uses_atomic_replace_and_private_permissions(tmp_path, monkeypatch): |
| 738 | auth_path = tmp_path / "auth.json" |
| 739 | replacements: list[tuple[Path, Path]] = [] |
| 740 | replace = codex.os.replace |
| 741 | |
| 742 | def record_replace(source, destination): |
| 743 | replacements.append((Path(source), Path(destination))) |
| 744 | replace(source, destination) |
| 745 | |
| 746 | monkeypatch.setattr(codex.os, "replace", record_replace) |
| 747 | |
| 748 | codex.write_auth_file(auth_path, {"tokens": {"refresh_token": "refresh"}}) |
| 749 | |
| 750 | assert json.loads(auth_path.read_text(encoding="utf-8")) == { |
| 751 | "tokens": {"refresh_token": "refresh"} |
| 752 | } |
| 753 | assert stat.S_IMODE(auth_path.stat().st_mode) == 0o600 |
| 754 | assert len(replacements) == 1 |
| 755 | assert replacements[0][0] != auth_path |
| 756 | assert replacements[0][1] == auth_path |
| 757 | assert list(tmp_path.glob(".auth.json.*.tmp")) == [] |
| 758 | |
| 759 | |
| 760 | def test_write_auth_file_falls_back_for_file_bind_mount(tmp_path, monkeypatch): |
| 761 | auth_path = tmp_path / "auth.json" |
| 762 | auth_path.write_text(json.dumps({"tokens": {"refresh_token": "refresh-0"}}), encoding="utf-8") |
| 763 | |
| 764 | def reject_replace(source, destination): |
| 765 | raise OSError(codex.errno.EBUSY, "Device or resource busy", destination) |
| 766 | |
| 767 | monkeypatch.setattr(codex.os, "replace", reject_replace) |
| 768 | |
| 769 | codex.write_auth_file(auth_path, {"tokens": {"refresh_token": "refresh-1"}}) |
| 770 | |
| 771 | assert json.loads(auth_path.read_text(encoding="utf-8")) == { |
| 772 | "tokens": {"refresh_token": "refresh-1"} |
| 773 | } |
| 774 | assert stat.S_IMODE(auth_path.stat().st_mode) == 0o600 |
| 775 | assert list(tmp_path.glob(".auth.json.*.tmp")) == [] |
| 776 | |
| 777 | |
| 778 | def test_write_auth_file_falls_back_when_parent_rejects_temporary_files(tmp_path, monkeypatch): |
| 779 | auth_path = tmp_path / "auth.json" |
| 780 | auth_path.write_text(json.dumps({"tokens": {"refresh_token": "refresh-0"}}), encoding="utf-8") |
| 781 | open_file = codex.os.open |
| 782 | |
| 783 | def reject_temporary_file(path, flags, mode=0o777): |
| 784 | if str(path).endswith(".tmp"): |
| 785 | raise OSError(codex.errno.EACCES, "Permission denied", path) |
| 786 | return open_file(path, flags, mode) |
| 787 | |
| 788 | monkeypatch.setattr(codex.os, "open", reject_temporary_file) |
| 789 | |
| 790 | codex.write_auth_file(auth_path, {"tokens": {"refresh_token": "refresh-1"}}) |
| 791 | |
| 792 | assert json.loads(auth_path.read_text(encoding="utf-8")) == { |
| 793 | "tokens": {"refresh_token": "refresh-1"} |
| 794 | } |
| 795 | assert stat.S_IMODE(auth_path.stat().st_mode) == 0o600 |
| 796 | assert not auth_path.with_name(".auth.json.lock").exists() |
| 797 | |
| 798 | |
| 799 | def test_resolve_auth_write_path_preserves_custom_symlink_target(tmp_path, monkeypatch): |
| 800 | target = tmp_path / "mounted" / "auth.json" |
| 801 | target.parent.mkdir() |
| 802 | target.write_text(json.dumps({"tokens": {"refresh_token": "refresh-0"}}), encoding="utf-8") |
| 803 | symlink = tmp_path / "auth.json" |
| 804 | symlink.symlink_to(target) |
| 805 | monkeypatch.setattr(codex, "codex_config", lambda: {"auth_file_path": str(symlink)}) |
| 806 | |
| 807 | resolved_path = codex.resolve_auth_write_path() |
| 808 | codex.write_auth_file(resolved_path, {"tokens": {"refresh_token": "refresh-1"}}) |
| 809 | |
| 810 | assert resolved_path == target |
| 811 | assert symlink.is_symlink() |
| 812 | assert json.loads(target.read_text(encoding="utf-8")) == { |
| 813 | "tokens": {"refresh_token": "refresh-1"} |
| 814 | } |
| 815 | |
| 816 | |
| 817 | def test_lock_file_retries_windows_contention(tmp_path, monkeypatch): |
| 818 | class FakeMsvcrt: |
| 819 | LK_NBLCK = 1 |
| 820 | LK_UNLCK = 2 |
| 821 | |
| 822 | def __init__(self): |
| 823 | self.calls: list[int] = [] |
| 824 | |
| 825 | def locking(self, _descriptor: int, mode: int, _length: int) -> None: |
| 826 | self.calls.append(mode) |
| 827 | if mode == self.LK_NBLCK and self.calls.count(mode) < 3: |
| 828 | raise OSError(codex.errno.EACCES, "Permission denied") |
| 829 | |
| 830 | fake_msvcrt = FakeMsvcrt() |
| 831 | sleeps: list[float] = [] |
| 832 | monkeypatch.setattr(codex, "fcntl", None) |
| 833 | monkeypatch.setattr(codex, "msvcrt", fake_msvcrt) |
| 834 | monkeypatch.setattr(codex.time, "sleep", sleeps.append) |
| 835 | |
| 836 | with (tmp_path / "auth.lock").open("a+b") as handle: |
| 837 | codex._lock_file(handle) |
| 838 | codex._unlock_file(handle) |
| 839 | |
| 840 | assert fake_msvcrt.calls == [ |
| 841 | fake_msvcrt.LK_NBLCK, |
| 842 | fake_msvcrt.LK_NBLCK, |
| 843 | fake_msvcrt.LK_NBLCK, |
| 844 | fake_msvcrt.LK_UNLCK, |
| 845 | ] |
| 846 | assert sleeps == [codex.WINDOWS_LOCK_RETRY_SECONDS, codex.WINDOWS_LOCK_RETRY_SECONDS] |
| 847 | |
| 848 | |
| 849 | @pytest.mark.parametrize( |
| 850 | ("host", "expected"), |
| 851 | [ |
| 852 | ("localhost:5000", True), |
| 853 | ("127.0.0.1:5000", True), |
| 854 | ("[::1]:5000", True), |
| 855 | ("::1", True), |
| 856 | ("example.com:5000", False), |
| 857 | ], |
| 858 | ) |
| 859 | def test_proxy_local_host_detection_supports_loopback_ipv6(host, expected): |
| 860 | assert routes._host_is_local(host) is expected |
| 861 | |
| 862 | |
| 863 | def test_load_auth_serializes_refresh_across_threads(tmp_path, monkeypatch): |
| 864 | auth_path = tmp_path / "auth.json" |
| 865 | _write_refreshable_auth(auth_path) |
| 866 | monkeypatch.setattr(codex, "resolve_auth_write_path", lambda: auth_path) |
| 867 | refresh_started = threading.Event() |
| 868 | release_refresh = threading.Event() |
| 869 | calls: list[str] = [] |
| 870 | results: list[codex.EffectiveAuth] = [] |
| 871 | |
| 872 | def refresh_tokens(refresh_token: str) -> dict[str, str]: |
| 873 | calls.append(refresh_token) |
| 874 | refresh_started.set() |
| 875 | assert release_refresh.wait(timeout=2) |
| 876 | return _rotated_tokens() |
| 877 | |
| 878 | monkeypatch.setattr(codex, "refresh_tokens", refresh_tokens) |
| 879 | first = threading.Thread(target=lambda: results.append(codex.load_auth())) |
| 880 | second = threading.Thread(target=lambda: results.append(codex.load_auth())) |
| 881 | |
| 882 | first.start() |
| 883 | assert refresh_started.wait(timeout=2) |
| 884 | second.start() |
| 885 | time.sleep(0.1) |
| 886 | |
| 887 | assert calls == ["refresh-0"] |
| 888 | release_refresh.set() |
| 889 | first.join(timeout=2) |
| 890 | second.join(timeout=2) |
| 891 | |
| 892 | assert not first.is_alive() |
| 893 | assert not second.is_alive() |
| 894 | assert calls == ["refresh-0"] |
| 895 | assert [result.refresh_token for result in results] == ["refresh-1", "refresh-1"] |
| 896 | |
| 897 | |
| 898 | def test_load_auth_holds_lock_until_rotated_token_is_persisted(tmp_path, monkeypatch): |
| 899 | auth_path = tmp_path / "auth.json" |
| 900 | _write_refreshable_auth(auth_path) |
| 901 | monkeypatch.setattr(codex, "resolve_auth_write_path", lambda: auth_path) |
| 902 | persist_started = threading.Event() |
| 903 | release_persist = threading.Event() |
| 904 | calls: list[str] = [] |
| 905 | results: list[codex.EffectiveAuth] = [] |
| 906 | write_auth_file = codex._write_auth_file_unlocked |
| 907 | |
| 908 | def refresh_tokens(refresh_token: str) -> dict[str, str]: |
| 909 | calls.append(refresh_token) |
| 910 | return _rotated_tokens() |
| 911 | |
| 912 | def delay_write(path: Path, data: dict) -> None: |
| 913 | persist_started.set() |
| 914 | assert release_persist.wait(timeout=2) |
| 915 | write_auth_file(path, data) |
| 916 | |
| 917 | monkeypatch.setattr(codex, "refresh_tokens", refresh_tokens) |
| 918 | monkeypatch.setattr(codex, "_write_auth_file_unlocked", delay_write) |
| 919 | first = threading.Thread(target=lambda: results.append(codex.load_auth())) |
| 920 | second = threading.Thread(target=lambda: results.append(codex.load_auth())) |
| 921 | |
| 922 | first.start() |
| 923 | assert persist_started.wait(timeout=2) |
| 924 | second.start() |
| 925 | time.sleep(0.1) |
| 926 | |
| 927 | assert calls == ["refresh-0"] |
| 928 | release_persist.set() |
| 929 | first.join(timeout=2) |
| 930 | second.join(timeout=2) |
| 931 | |
| 932 | assert not first.is_alive() |
| 933 | assert not second.is_alive() |
| 934 | assert calls == ["refresh-0"] |
| 935 | assert [result.refresh_token for result in results] == ["refresh-1", "refresh-1"] |
| 936 | |
| 937 | |
| 938 | def test_load_auth_serializes_refresh_across_processes(tmp_path): |
| 939 | auth_path = tmp_path / "auth.json" |
| 940 | _write_refreshable_auth(auth_path) |
| 941 | context = multiprocessing.get_context("spawn") |
| 942 | refresh_started = context.Event() |
| 943 | release_refresh = context.Event() |
| 944 | calls = context.Queue() |
| 945 | results = context.Queue() |
| 946 | process_args = (str(auth_path), refresh_started, release_refresh, calls, results) |
| 947 | first = context.Process(target=_load_auth_in_process, args=process_args) |
| 948 | second = context.Process(target=_load_auth_in_process, args=process_args) |
| 949 | |
| 950 | first.start() |
| 951 | assert refresh_started.wait(timeout=2) |
| 952 | assert calls.get(timeout=2) == "refresh-0" |
| 953 | second.start() |
| 954 | with pytest.raises(queue.Empty): |
| 955 | calls.get(timeout=0.2) |
| 956 | |
| 957 | release_refresh.set() |
| 958 | first.join(timeout=3) |
| 959 | second.join(timeout=3) |
| 960 | |
| 961 | assert first.exitcode == 0 |
| 962 | assert second.exitcode == 0 |
| 963 | with pytest.raises(queue.Empty): |
| 964 | calls.get(timeout=0.2) |
| 965 | assert sorted([results.get(timeout=2), results.get(timeout=2)]) == ["refresh-1", "refresh-1"] |
| 966 | |
| 967 | |
| 968 | def test_disconnect_auth_only_mutates_agent_zero_private_auth_file(tmp_path, monkeypatch): |
| 969 | private_auth = tmp_path / "private-auth.json" |
| 970 | shared_auth = tmp_path / ".codex" / "auth.json" |
| 971 | private_auth.write_text( |
| 972 | json.dumps( |
| 973 | { |
| 974 | "auth_mode": "chatgpt", |
| 975 | "OPENAI_API_KEY": "sk-keep", |
| 976 | "tokens": { |
| 977 | "access_token": "access", |
| 978 | "refresh_token": "refresh", |
| 979 | "id_token": "id", |
| 980 | "account_id": "account", |
| 981 | }, |
| 982 | "last_refresh": "2026-01-01T00:00:00Z", |
| 983 | } |
| 984 | ), |
| 985 | encoding="utf-8", |
| 986 | ) |
| 987 | shared_auth.parent.mkdir() |
| 988 | shared_auth.write_text( |
| 989 | json.dumps( |
| 990 | { |
| 991 | "auth_mode": "chatgpt", |
| 992 | "OPENAI_API_KEY": None, |
| 993 | "tokens": {"access_token": "access", "account_id": "account"}, |
| 994 | "last_refresh": "2026-01-01T00:00:00Z", |
| 995 | } |
| 996 | ), |
| 997 | encoding="utf-8", |
| 998 | ) |
| 999 | shared_before = shared_auth.read_text(encoding="utf-8") |
| 1000 | monkeypatch.setattr(codex, "resolve_auth_write_path", lambda: private_auth) |
| 1001 | |
| 1002 | result = codex.disconnect_auth() |
| 1003 | |
| 1004 | assert result["disconnected"] is True |
| 1005 | assert result["preserved_auth_files"] == [str(private_auth)] |
| 1006 | preserved = json.loads(private_auth.read_text(encoding="utf-8")) |
| 1007 | assert preserved == {"OPENAI_API_KEY": "sk-keep"} |
| 1008 | assert shared_auth.read_text(encoding="utf-8") == shared_before |
| 1009 | |
| 1010 | |
| 1011 | def _write_refreshable_auth(path: Path) -> None: |
| 1012 | path.write_text( |
| 1013 | json.dumps( |
| 1014 | { |
| 1015 | "auth_mode": "chatgpt", |
| 1016 | "tokens": { |
| 1017 | "access_token": "", |
| 1018 | "refresh_token": "refresh-0", |
| 1019 | "id_token": "", |
| 1020 | "account_id": "account", |
| 1021 | }, |
| 1022 | "last_refresh": "", |
| 1023 | } |
| 1024 | ), |
| 1025 | encoding="utf-8", |
| 1026 | ) |
| 1027 | |
| 1028 | |
| 1029 | def _rotated_tokens() -> dict[str, str]: |
| 1030 | return { |
| 1031 | "access_token": "access-1", |
| 1032 | "refresh_token": "refresh-1", |
| 1033 | "id_token": "", |
| 1034 | } |
| 1035 | |
| 1036 | |
| 1037 | def _load_auth_in_process(auth_path: str, refresh_started, release_refresh, calls, results) -> None: |
| 1038 | codex.resolve_auth_write_path = lambda: Path(auth_path) |
| 1039 | codex._auth_lock_path = lambda _path: Path(auth_path).parent / ".auth.lock" |
| 1040 | |
| 1041 | def refresh_tokens(refresh_token: str) -> dict[str, str]: |
| 1042 | calls.put(refresh_token) |
| 1043 | refresh_started.set() |
| 1044 | assert release_refresh.wait(timeout=5) |
| 1045 | return _rotated_tokens() |
| 1046 | |
| 1047 | codex.refresh_tokens = refresh_tokens |
| 1048 | results.put(codex.load_auth().refresh_token) |
| 1049 | |
| 1050 | |
| 1051 | def test_provider_config_uses_container_local_agent_zero_origin(): |
| 1052 | provider_path = Path(__file__).resolve().parents[1] / "plugins/_oauth/conf/model_providers.yaml" |
| 1053 | provider_config = yaml.safe_load(provider_path.read_text(encoding="utf-8")) |
| 1054 | codex_provider = provider_config["chat"]["codex_oauth"] |
| 1055 | |
| 1056 | assert codex_provider["name"] == "Codex/ChatGPT Account" |
| 1057 | assert codex_provider["models_list"]["endpoint_url"] == "/models" |
| 1058 | assert codex_provider["kwargs"]["api_base"] == "http://127.0.0.1/oauth/codex/v1" |
| 1059 | assert "50001" not in json.dumps(codex_provider) |
| 1060 | |
| 1061 | |
| 1062 | def test_codex_provider_leaves_api_key_empty_until_connected(monkeypatch): |
| 1063 | monkeypatch.setattr(oauth_dummy_key, "oauth_provider_is_connected", lambda _provider_id: False) |
| 1064 | data = {"args": ("codex_oauth",), "kwargs": {}, "result": "None"} |
| 1065 | |
| 1066 | oauth_dummy_key.OAuthAccountDummyKey(agent=None).execute(data=data) |
| 1067 | |
| 1068 | assert data["result"] == "None" |
| 1069 | |
| 1070 | |
| 1071 | def test_codex_provider_reports_dummy_api_key_when_connected(monkeypatch): |
| 1072 | monkeypatch.setattr(oauth_dummy_key, "oauth_provider_is_connected", lambda _provider_id: True) |
| 1073 | data = {"args": ("codex_oauth",), "kwargs": {}, "result": "None"} |
| 1074 | |
| 1075 | oauth_dummy_key.OAuthAccountDummyKey(agent=None).execute(data=data) |
| 1076 | |
| 1077 | assert data["result"] == "oauth" |
| 1078 | |
| 1079 | |
| 1080 | def test_codex_provider_preserves_configured_api_key(): |
| 1081 | data = {"args": ("codex_oauth",), "kwargs": {}, "result": "configured"} |
| 1082 | |
| 1083 | oauth_dummy_key.OAuthAccountDummyKey(agent=None).execute(data=data) |
| 1084 | |
| 1085 | assert data["result"] == "configured" |