mirror of
https://github.com/openai/codex.git
synced 2026-05-18 10:12:59 +00:00
362 lines
10 KiB
Python
362 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.resources as resources
|
|
import inspect
|
|
import tomllib
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import codex_app_server
|
|
from codex_app_server import (
|
|
AppServerConfig,
|
|
AsyncCodex,
|
|
AsyncThread,
|
|
Codex,
|
|
RunResult,
|
|
Thread,
|
|
)
|
|
from codex_app_server.models import InitializeResponse
|
|
|
|
EXPECTED_ROOT_EXPORTS = [
|
|
"__version__",
|
|
"AppServerConfig",
|
|
"Codex",
|
|
"AsyncCodex",
|
|
"Thread",
|
|
"AsyncThread",
|
|
"TurnHandle",
|
|
"AsyncTurnHandle",
|
|
"RunResult",
|
|
"Input",
|
|
"InputItem",
|
|
"TextInput",
|
|
"ImageInput",
|
|
"LocalImageInput",
|
|
"SkillInput",
|
|
"MentionInput",
|
|
"ApprovalsReviewer",
|
|
"AskForApproval",
|
|
"Personality",
|
|
"PlanType",
|
|
"ReasoningEffort",
|
|
"ReasoningSummary",
|
|
"SandboxMode",
|
|
"SandboxPolicy",
|
|
"SortDirection",
|
|
"ThreadListCwdFilter",
|
|
"ThreadSortKey",
|
|
"ThreadSource",
|
|
"ThreadSourceKind",
|
|
"ThreadStartSource",
|
|
"TurnStatus",
|
|
"retry_on_overload",
|
|
"AppServerError",
|
|
"TransportClosedError",
|
|
"JsonRpcError",
|
|
"AppServerRpcError",
|
|
"ParseError",
|
|
"InvalidRequestError",
|
|
"MethodNotFoundError",
|
|
"InvalidParamsError",
|
|
"InternalRpcError",
|
|
"ServerBusyError",
|
|
"RetryLimitExceededError",
|
|
"is_retryable_error",
|
|
]
|
|
|
|
|
|
def _keyword_only_names(fn: object) -> list[str]:
|
|
signature = inspect.signature(fn)
|
|
return [
|
|
param.name
|
|
for param in signature.parameters.values()
|
|
if param.kind == inspect.Parameter.KEYWORD_ONLY
|
|
]
|
|
|
|
|
|
def _assert_no_any_annotations(fn: object) -> None:
|
|
signature = inspect.signature(fn)
|
|
for param in signature.parameters.values():
|
|
if param.annotation is Any:
|
|
raise AssertionError(
|
|
f"{fn} has public parameter typed as Any: {param.name}"
|
|
)
|
|
if signature.return_annotation is Any:
|
|
raise AssertionError(f"{fn} has public return annotation typed as Any")
|
|
|
|
|
|
def test_root_exports_app_server_config() -> None:
|
|
assert AppServerConfig.__name__ == "AppServerConfig"
|
|
|
|
|
|
def test_root_exports_run_result() -> None:
|
|
assert RunResult.__name__ == "RunResult"
|
|
|
|
|
|
def test_package_and_default_client_versions_follow_project_version() -> None:
|
|
pyproject_path = Path(__file__).resolve().parents[1] / "pyproject.toml"
|
|
pyproject = tomllib.loads(pyproject_path.read_text())
|
|
|
|
assert codex_app_server.__version__ == pyproject["project"]["version"]
|
|
assert AppServerConfig().client_version == codex_app_server.__version__
|
|
|
|
|
|
def test_package_includes_py_typed_marker() -> None:
|
|
marker = resources.files("codex_app_server").joinpath("py.typed")
|
|
assert marker.is_file()
|
|
|
|
|
|
def test_package_root_exports_only_public_api() -> None:
|
|
"""The package root should expose the supported SDK surface, not internals."""
|
|
assert codex_app_server.__all__ == EXPECTED_ROOT_EXPORTS
|
|
assert {
|
|
name: hasattr(codex_app_server, name) for name in EXPECTED_ROOT_EXPORTS
|
|
} == {name: True for name in EXPECTED_ROOT_EXPORTS}
|
|
assert {
|
|
"AppServerClient": hasattr(codex_app_server, "AppServerClient"),
|
|
"AsyncAppServerClient": hasattr(codex_app_server, "AsyncAppServerClient"),
|
|
"InitializeResponse": hasattr(codex_app_server, "InitializeResponse"),
|
|
"ThreadStartParams": hasattr(codex_app_server, "ThreadStartParams"),
|
|
"TurnStartParams": hasattr(codex_app_server, "TurnStartParams"),
|
|
} == {
|
|
"AppServerClient": False,
|
|
"AsyncAppServerClient": False,
|
|
"InitializeResponse": False,
|
|
"ThreadStartParams": False,
|
|
"TurnStartParams": False,
|
|
}
|
|
|
|
|
|
def test_package_star_import_matches_public_api() -> None:
|
|
"""Star imports should follow the same explicit public API list."""
|
|
namespace: dict[str, object] = {}
|
|
exec("from codex_app_server import *", namespace)
|
|
|
|
exported = set(namespace) - {"__builtins__"}
|
|
assert exported == set(EXPECTED_ROOT_EXPORTS)
|
|
|
|
|
|
def test_generated_public_signatures_are_snake_case_and_typed() -> None:
|
|
"""Generated convenience methods should expose typed Pythonic keyword names."""
|
|
expected = {
|
|
Codex.thread_start: [
|
|
"approval_policy",
|
|
"approvals_reviewer",
|
|
"base_instructions",
|
|
"config",
|
|
"cwd",
|
|
"developer_instructions",
|
|
"ephemeral",
|
|
"model",
|
|
"model_provider",
|
|
"personality",
|
|
"sandbox",
|
|
"service_name",
|
|
"service_tier",
|
|
"session_start_source",
|
|
"thread_source",
|
|
],
|
|
Codex.thread_list: [
|
|
"archived",
|
|
"cursor",
|
|
"cwd",
|
|
"limit",
|
|
"model_providers",
|
|
"search_term",
|
|
"sort_direction",
|
|
"sort_key",
|
|
"source_kinds",
|
|
"use_state_db_only",
|
|
],
|
|
Codex.thread_resume: [
|
|
"approval_policy",
|
|
"approvals_reviewer",
|
|
"base_instructions",
|
|
"config",
|
|
"cwd",
|
|
"developer_instructions",
|
|
"model",
|
|
"model_provider",
|
|
"personality",
|
|
"sandbox",
|
|
"service_tier",
|
|
],
|
|
Codex.thread_fork: [
|
|
"approval_policy",
|
|
"approvals_reviewer",
|
|
"base_instructions",
|
|
"config",
|
|
"cwd",
|
|
"developer_instructions",
|
|
"ephemeral",
|
|
"model",
|
|
"model_provider",
|
|
"sandbox",
|
|
"service_tier",
|
|
"thread_source",
|
|
],
|
|
Thread.turn: [
|
|
"approval_policy",
|
|
"approvals_reviewer",
|
|
"cwd",
|
|
"effort",
|
|
"model",
|
|
"output_schema",
|
|
"personality",
|
|
"sandbox_policy",
|
|
"service_tier",
|
|
"summary",
|
|
],
|
|
Thread.run: [
|
|
"approval_policy",
|
|
"approvals_reviewer",
|
|
"cwd",
|
|
"effort",
|
|
"model",
|
|
"output_schema",
|
|
"personality",
|
|
"sandbox_policy",
|
|
"service_tier",
|
|
"summary",
|
|
],
|
|
AsyncCodex.thread_start: [
|
|
"approval_policy",
|
|
"approvals_reviewer",
|
|
"base_instructions",
|
|
"config",
|
|
"cwd",
|
|
"developer_instructions",
|
|
"ephemeral",
|
|
"model",
|
|
"model_provider",
|
|
"personality",
|
|
"sandbox",
|
|
"service_name",
|
|
"service_tier",
|
|
"session_start_source",
|
|
"thread_source",
|
|
],
|
|
AsyncCodex.thread_list: [
|
|
"archived",
|
|
"cursor",
|
|
"cwd",
|
|
"limit",
|
|
"model_providers",
|
|
"search_term",
|
|
"sort_direction",
|
|
"sort_key",
|
|
"source_kinds",
|
|
"use_state_db_only",
|
|
],
|
|
AsyncCodex.thread_resume: [
|
|
"approval_policy",
|
|
"approvals_reviewer",
|
|
"base_instructions",
|
|
"config",
|
|
"cwd",
|
|
"developer_instructions",
|
|
"model",
|
|
"model_provider",
|
|
"personality",
|
|
"sandbox",
|
|
"service_tier",
|
|
],
|
|
AsyncCodex.thread_fork: [
|
|
"approval_policy",
|
|
"approvals_reviewer",
|
|
"base_instructions",
|
|
"config",
|
|
"cwd",
|
|
"developer_instructions",
|
|
"ephemeral",
|
|
"model",
|
|
"model_provider",
|
|
"sandbox",
|
|
"service_tier",
|
|
"thread_source",
|
|
],
|
|
AsyncThread.turn: [
|
|
"approval_policy",
|
|
"approvals_reviewer",
|
|
"cwd",
|
|
"effort",
|
|
"model",
|
|
"output_schema",
|
|
"personality",
|
|
"sandbox_policy",
|
|
"service_tier",
|
|
"summary",
|
|
],
|
|
AsyncThread.run: [
|
|
"approval_policy",
|
|
"approvals_reviewer",
|
|
"cwd",
|
|
"effort",
|
|
"model",
|
|
"output_schema",
|
|
"personality",
|
|
"sandbox_policy",
|
|
"service_tier",
|
|
"summary",
|
|
],
|
|
}
|
|
|
|
for fn, expected_kwargs in expected.items():
|
|
actual = _keyword_only_names(fn)
|
|
assert actual == expected_kwargs, f"unexpected kwargs for {fn}: {actual}"
|
|
assert all(name == name.lower() for name in actual), (
|
|
f"non snake_case kwargs in {fn}: {actual}"
|
|
)
|
|
_assert_no_any_annotations(fn)
|
|
|
|
|
|
def test_lifecycle_methods_are_codex_scoped() -> None:
|
|
assert hasattr(Codex, "thread_resume")
|
|
assert hasattr(Codex, "thread_fork")
|
|
assert hasattr(Codex, "thread_archive")
|
|
assert hasattr(Codex, "thread_unarchive")
|
|
assert hasattr(AsyncCodex, "thread_resume")
|
|
assert hasattr(AsyncCodex, "thread_fork")
|
|
assert hasattr(AsyncCodex, "thread_archive")
|
|
assert hasattr(AsyncCodex, "thread_unarchive")
|
|
assert not hasattr(Codex, "thread")
|
|
assert not hasattr(AsyncCodex, "thread")
|
|
|
|
assert not hasattr(Thread, "resume")
|
|
assert not hasattr(Thread, "fork")
|
|
assert not hasattr(Thread, "archive")
|
|
assert not hasattr(Thread, "unarchive")
|
|
assert not hasattr(AsyncThread, "resume")
|
|
assert not hasattr(AsyncThread, "fork")
|
|
assert not hasattr(AsyncThread, "archive")
|
|
assert not hasattr(AsyncThread, "unarchive")
|
|
|
|
for fn in (
|
|
Codex.thread_archive,
|
|
Codex.thread_unarchive,
|
|
AsyncCodex.thread_archive,
|
|
AsyncCodex.thread_unarchive,
|
|
):
|
|
_assert_no_any_annotations(fn)
|
|
|
|
|
|
def test_initialize_metadata_parses_user_agent_shape() -> None:
|
|
payload = InitializeResponse.model_validate({"userAgent": "codex-cli/1.2.3"})
|
|
parsed = Codex._validate_initialize(payload)
|
|
assert parsed is payload
|
|
assert parsed.userAgent == "codex-cli/1.2.3"
|
|
assert parsed.serverInfo is not None
|
|
assert parsed.serverInfo.name == "codex-cli"
|
|
assert parsed.serverInfo.version == "1.2.3"
|
|
|
|
|
|
def test_initialize_metadata_requires_non_empty_information() -> None:
|
|
try:
|
|
Codex._validate_initialize(InitializeResponse.model_validate({}))
|
|
except RuntimeError as exc:
|
|
assert "missing required metadata" in str(exc)
|
|
else:
|
|
raise AssertionError(
|
|
"expected RuntimeError when initialize metadata is missing"
|
|
)
|