"""Unit + smoke tests for Layer 1 (`PtyClaudeProcess`). Unit tests exercise pure argv/env construction and don't require `claude`. The smoke test spawns the real binary and is opt-in via env var because it hits the user's OAuth state and the wider system. """ from __future__ import annotations import asyncio import os import pytest from claude_code_api import CLINotFoundError from claude_code_api.pty import ( PtyClaudeProcess, PtyProcessOptions, build_argv, build_env, ) # --- argv construction ---------------------------------------------------- def test_build_argv_minimal_uses_session_id_and_permission_mode() -> None: opts = PtyProcessOptions(cwd="/tmp") argv = build_argv(opts, session_id="abc-123") assert argv[0] == "claude" # --session-id must come early so it can be observed in `ps` output even # if later flags are mistyped/dropped. assert argv[1:3] == ["--session-id", "abc-123"] assert "--permission-mode" in argv pm_index = argv.index("--permission-mode") assert argv[pm_index + 1] == "bypassPermissions" # Must never contain headless-only flags. for forbidden in ("--print", "-p", "--output-format", "--input-format"): assert forbidden not in argv def test_build_argv_dangerously_skip_permissions_excludes_permission_mode() -> None: opts = PtyProcessOptions(cwd="/tmp", dangerously_skip_permissions=True) argv = build_argv(opts, session_id="s") assert "--dangerously-skip-permissions" in argv assert "--permission-mode" not in argv def test_build_argv_includes_optional_flags_when_set() -> None: opts = PtyProcessOptions( cwd="/tmp", model="claude-opus-4-7", system_prompt="be brief", append_system_prompt="also be kind", allowed_tools=("Read", "Glob"), disallowed_tools=("Bash",), mcp_config=("/tmp/a.json", "/tmp/b.json"), add_dir=("/srv/x", "/srv/y"), effort="high", settings="/tmp/settings.json", extra_args=("--brief",), ) argv = build_argv(opts, session_id="s") # Each flag should pair with its value. def _pairs(flag: str) -> list[str]: return [argv[i + 1] for i, v in enumerate(argv) if v == flag and i + 1 < len(argv)] assert _pairs("--model") == ["claude-opus-4-7"] assert _pairs("--system-prompt") == ["be brief"] assert _pairs("--append-system-prompt") == ["also be kind"] # CSV form per claude CLI conventions. assert _pairs("--allowedTools") == ["Read,Glob"] assert _pairs("--disallowedTools") == ["Bash"] assert _pairs("--mcp-config") == ["/tmp/a.json", "/tmp/b.json"] assert _pairs("--effort") == ["high"] assert _pairs("--settings") == ["/tmp/settings.json"] # --add-dir is variadic in claude CLI: one flag, multiple values. add_dir_at = argv.index("--add-dir") assert argv[add_dir_at + 1 : add_dir_at + 3] == ["/srv/x", "/srv/y"] # extra_args are passthrough at the end. assert argv[-1] == "--brief" def test_build_argv_omits_unset_optionals() -> None: opts = PtyProcessOptions(cwd="/tmp") argv = build_argv(opts, session_id="s") for flag in ( "--model", "--system-prompt", "--append-system-prompt", "--allowedTools", "--disallowedTools", "--mcp-config", "--add-dir", "--effort", "--settings", ): assert flag not in argv def test_build_argv_resume_session_id_replaces_session_id_flag() -> None: """Resume mode swaps `--session-id ` for `--resume `. claude rejects the two flags together unless `--fork-session` is also passed (which would branch the session into a new JSONL). Higher layers pick resume mode when they've seeded a JSONL by hand and need claude to pick it up rather than create a new one. """ opts = PtyProcessOptions(cwd="/tmp", resume_session_id="resume-uuid") argv = build_argv(opts, session_id="ignored-fresh-uuid") assert argv[1:3] == ["--resume", "resume-uuid"] assert "--session-id" not in argv def test_options_reject_session_id_with_resume_session_id() -> None: with pytest.raises(ValueError, match="session_id"): PtyProcessOptions(cwd="/tmp", session_id="a", resume_session_id="b") def test_pty_process_reports_resume_session_id_as_session_id() -> None: """When constructed in resume mode, the process advertises the resumed session id (the id of the JSONL on disk) — not a fresh uuid. Higher layers rely on `pty.session_id` to compute the JSONL path.""" proc = PtyClaudeProcess(PtyProcessOptions(cwd="/tmp", resume_session_id="seeded-123")) assert proc.session_id == "seeded-123" assert "--resume" in proc.argv assert "--session-id" not in proc.argv def test_options_reject_invalid_permission_mode() -> None: with pytest.raises(ValueError, match="permission_mode"): PtyProcessOptions(cwd="/tmp", permission_mode="banana") def test_options_reject_nonpositive_dimensions() -> None: with pytest.raises(ValueError, match="dimensions"): PtyProcessOptions(cwd="/tmp", dimensions=(0, 80)) # --- env construction ----------------------------------------------------- def test_build_env_strips_provider_vars_by_default() -> None: base = { "PATH": "/usr/bin", "HOME": "/home/x", "ANTHROPIC_API_KEY": "sk-xxx", "ANTHROPIC_AUTH_TOKEN": "tok", "ANTHROPIC_BASE_URL": "https://x.example", } env = build_env(PtyProcessOptions(cwd="/tmp"), base=base) for name in ("ANTHROPIC_API_KEY", "ANTHROPIC_AUTH_TOKEN", "ANTHROPIC_BASE_URL"): assert name not in env assert env["PATH"] == "/usr/bin" assert env["HOME"] == "/home/x" assert env["TERM"] == "xterm-256color" assert env["NO_COLOR"] == "1" def test_build_env_preserve_provider_env_keeps_keys() -> None: base = {"ANTHROPIC_API_KEY": "sk-xxx", "PATH": "/usr/bin"} opts = PtyProcessOptions(cwd="/tmp", preserve_provider_env=True) env = build_env(opts, base=base) assert env["ANTHROPIC_API_KEY"] == "sk-xxx" def test_build_env_extra_env_overrides_base() -> None: base = {"PATH": "/usr/bin", "TERM": "dumb"} opts = PtyProcessOptions(cwd="/tmp", extra_env={"FOO": "bar", "TERM": "vt100"}) env = build_env(opts, base=base) assert env["FOO"] == "bar" # Explicit override should win over the default TERM we set in build_env. assert env["TERM"] == "vt100" # --- construction-only PtyClaudeProcess sanity ---------------------------- def test_session_id_is_autogenerated_when_omitted() -> None: proc = PtyClaudeProcess(PtyProcessOptions(cwd="/tmp")) # UUID4 is 36 chars including dashes. assert len(proc.session_id) == 36 assert proc.is_alive() is False assert proc.pid is None def test_session_id_is_passed_through_when_provided() -> None: proc = PtyClaudeProcess(PtyProcessOptions(cwd="/tmp", session_id="custom-id")) assert proc.session_id == "custom-id" assert "--session-id" in proc.argv assert proc.argv[proc.argv.index("--session-id") + 1] == "custom-id" # --- error mapping (Stage 10) --------------------------------------------- @pytest.mark.asyncio async def test_start_raises_cli_not_found_when_executable_missing(tmp_path) -> None: """`PtyClaudeProcess.start()` lifts ptyprocess's `FileNotFoundError` (which fires from the pre-fork `which()` lookup) into our typed `CLINotFoundError` so callers don't need to know about the underlying library.""" opts = PtyProcessOptions( cwd=str(tmp_path), executable="claude-binary-that-does-not-exist-xyz", dangerously_skip_permissions=True, ) proc = PtyClaudeProcess(opts) with pytest.raises(CLINotFoundError) as info: await proc.start() assert "claude-binary-that-does-not-exist-xyz" in str(info.value) assert info.value.executable == "claude-binary-that-does-not-exist-xyz" # --- smoke test (real claude) --------------------------------------------- _SMOKE_ENV = "RUN_CLAUDE_SMOKE" @pytest.mark.skipif( os.environ.get(_SMOKE_ENV) != "1", reason=f"set {_SMOKE_ENV}=1 to run the real-`claude` smoke test", ) @pytest.mark.asyncio async def test_smoke_start_write_terminate(tmp_path) -> None: """End-to-end Layer 1 check against the installed `claude` binary. Spawns claude under a PTY, confirms it's alive, sends a no-op message (which we don't expect a turn to complete in this test), then terminates cleanly via SIGTERM. We only assert lifecycle invariants here — JSONL parsing and turn semantics live in later layers. """ opts = PtyProcessOptions( cwd=str(tmp_path), dangerously_skip_permissions=True, ) proc = PtyClaudeProcess(opts) await proc.start() pid = proc.pid try: assert pid is not None and pid > 0 # Give claude a moment to paint the TUI before we ask it to die. # If it can't even stay alive for a beat, something is fundamentally # wrong with the spawn (auth blocked, missing HOME, etc.). await asyncio.sleep(0.5) captured = proc.captured_output() assert proc.is_alive(), ( f"claude exited within 0.5s of spawn; captured {len(captured)} bytes:\n" f"{captured[:1000]!r}" ) await proc.write("hello") finally: exit_status = await proc.terminate(grace=5.0) assert proc.is_alive() is False # Either an exit code or a signal — anything other than `None` is fine. assert exit_status is not None, ( f"terminate() returned None for pid={pid}; output:\n{proc.captured_output()[:1000]!r}" )