262 lines
9.5 KiB
Python
262 lines
9.5 KiB
Python
"""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 <fresh>` for `--resume <existing>`.
|
|
|
|
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}"
|
|
)
|