Add approval mode contract tests

Cover the exact public ApprovalMode values and ensure unsupported modes fail before sync or async high-level APIs can issue client requests.

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Ahmed Ibrahim
2026-05-10 11:58:13 +03:00
parent ffe6e44a03
commit 800aa1d6ba
2 changed files with 97 additions and 1 deletions

View File

@@ -327,6 +327,50 @@ def test_sync_api_maps_approval_modes_for_started_work() -> None:
]
def test_sync_api_rejects_unknown_approval_mode_before_rpc() -> None:
"""Unknown approval modes should fail before building any client request."""
calls: list[str] = []
class FailOnRpcClient:
def thread_start(self, _params: object) -> SimpleNamespace:
calls.append("thread_start")
return SimpleNamespace(thread=SimpleNamespace(id="thread-started"))
def turn_start(
self,
_thread_id: str,
_wire_input: object,
*,
params: object | None = None, # noqa: ARG002
) -> SimpleNamespace:
calls.append("turn_start")
return SimpleNamespace(turn=SimpleNamespace(id="turn-1"))
client = FailOnRpcClient()
codex = object.__new__(Codex)
codex._client = client
errors: list[str] = []
for call in (
lambda: codex.thread_start(approval_mode="allow_all"), # type: ignore[arg-type]
lambda: Thread(client, "thread-1").turn( # type: ignore[arg-type]
TextInput("hello"),
approval_mode="allow_all",
),
):
with pytest.raises(ValueError) as exc_info:
call()
errors.append(str(exc_info.value))
assert (errors, calls) == (
[
"approval_mode must be one of: deny_all, auto_review",
"approval_mode must be one of: deny_all, auto_review",
],
[],
)
def test_async_api_maps_approval_modes_for_started_work() -> None:
"""Async start methods should serialize only supported approval modes."""
@@ -398,6 +442,55 @@ def test_async_api_maps_approval_modes_for_started_work() -> None:
asyncio.run(scenario())
def test_async_api_rejects_unknown_approval_mode_before_rpc() -> None:
"""Unknown async approval modes should fail before awaiting client calls."""
async def scenario() -> None:
"""Exercise async validation without starting a real app-server process."""
calls: list[str] = []
class FailOnAsyncRpcClient:
async def thread_start(self, _params: object) -> SimpleNamespace:
calls.append("thread_start")
return SimpleNamespace(thread=SimpleNamespace(id="thread-started"))
async def turn_start(
self,
_thread_id: str,
_wire_input: object,
*,
params: object | None = None, # noqa: ARG002
) -> SimpleNamespace:
calls.append("turn_start")
return SimpleNamespace(turn=SimpleNamespace(id="turn-1"))
codex = AsyncCodex()
codex._client = FailOnAsyncRpcClient()
codex._initialized = True
errors: list[str] = []
for call in (
lambda: codex.thread_start(approval_mode="allow_all"), # type: ignore[arg-type]
lambda: AsyncThread(codex, "thread-1").turn( # type: ignore[arg-type]
TextInput("hello"),
approval_mode="allow_all",
),
):
with pytest.raises(ValueError) as exc_info:
await call()
errors.append(str(exc_info.value))
assert (errors, calls) == (
[
"approval_mode must be one of: deny_all, auto_review",
"approval_mode must be one of: deny_all, auto_review",
],
[],
)
asyncio.run(scenario())
def test_turn_streams_can_consume_multiple_turns_on_one_client() -> None:
"""Two sync TurnHandle streams should advance independently on one client."""
client = AppServerClient()

View File

@@ -121,7 +121,10 @@ def test_root_exports_run_result() -> None:
def test_root_exports_approval_mode() -> None:
"""The root package should expose the high-level approval mode enum."""
assert ApprovalMode.deny_all.value == "deny_all"
assert [(mode.name, mode.value) for mode in ApprovalMode] == [
("deny_all", "deny_all"),
("auto_review", "auto_review"),
]
def test_package_and_default_client_versions_follow_project_version() -> None: