test(init): extend Docker e2e with bare/shim/per-subcommand cases

Port e2e/init/run.py onto the shared harness and extend coverage so
issue #245 (bare ``headroom init -g`` with no agents) is locked in:

* ``seq_claude_local`` / ``seq_copilot_global`` / ``seq_codex_local`` —
  the original scenario, now expressed as a sequence of Cases sharing
  one scratch so the manifest-merge behavior (claude + codex targets)
  is still exercised end-to-end
* ``bare_init_g_no_shims`` — regression guard for issue #245: asserts
  the new guided error mentions every probed target and the concrete
  ``headroom init -g <agent>`` example
* ``bare_init_g_with_all_shims`` — complementary happy path with all
  four shims present; asserts all three configurable agents report
  ``Configured ... (user scope)`` on stdout
* ``init_g_{claude,codex,copilot}_explicit`` — one case per
  subcommand, each with only its own shim on PATH, asserting exit 0
  and the correct per-agent settings file is written
* ``init_g_openclaw_missing`` — negative path for openclaw when its
  binary isn't installed (delegates to ``headroom wrap openclaw`` which
  can't be shimmed cheaply)
* ``init_verbose_no_shims`` — smoke test for ``headroom init -v``
  ensuring ``detect_init_targets``, ``global_scope=True``, and every
  agent name appear on stderr

Dockerfile is updated to COPY e2e/__init__.py and e2e/_lib/ so the
harness is importable inside the container. A new e2e/__init__.py
marks the tree as a package.

One small harness fix rides along: ``_resolve_headroom_bin`` captures
the absolute path to headroom before ``with_clean_path`` narrows PATH.
This is required for any case run inside a venv-scoped image - the
real ``headroom`` lives outside the shim dir and would otherwise be
hidden by the scrubbed PATH. Same bug would have bitten every future
command suite, so the fix belongs in the harness rather than run.py.

Verified locally inside the Docker image: all 10 cases pass.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
JerrettDavis 2026-04-23 16:12:27 -05:00
parent 6269a7e1fc
commit 48e2431510
4 changed files with 378 additions and 239 deletions

7
e2e/__init__.py Normal file
View file

@ -0,0 +1,7 @@
"""End-to-end test suites for Headroom CLI commands.
Subpackages:
_lib shared harness and helpers
init ``headroom init`` coverage
wrap ``headroom wrap`` coverage
"""

View file

@ -84,6 +84,27 @@ def _resolve_placeholder(spec: str, *, home: Path, project: Path) -> Path:
return Path(spec.format(home=str(home), project=str(project)))
def _resolve_headroom_bin(name: str) -> str:
"""Return the absolute path to the headroom binary before PATH is scrubbed.
``with_clean_path`` intentionally narrows PATH so agent shims dominate;
that would also hide the real ``headroom`` binary (typically at
``/opt/*venv/bin/headroom`` or similar). Resolving up-front lets the
subprocess launch even after PATH is cleaned.
"""
if os.sep in name or (os.altsep and os.altsep in name):
return name
import shutil
resolved = shutil.which(name)
if resolved:
return resolved
# Fall back to the bare name; subprocess will raise a clear
# FileNotFoundError that the case output surfaces.
return name
def _run_single(case: Case, headroom_bin: str = "headroom") -> bool:
"""Execute one case. Return True on pass, False on fail."""
@ -99,6 +120,10 @@ def _run_single(case: Case, headroom_bin: str = "headroom") -> bool:
for shim_name, behavior in case.shims.items():
make_shim(shim_name, shim_dir, behavior=behavior)
# Resolve headroom to its absolute path BEFORE mutating PATH so the
# shim dir can dominate PATH without losing the headroom binary.
resolved_bin = _resolve_headroom_bin(headroom_bin)
with with_clean_path([shim_dir]) as env:
env["HOME"] = str(home)
env["USERPROFILE"] = str(home)
@ -106,7 +131,7 @@ def _run_single(case: Case, headroom_bin: str = "headroom") -> bool:
env.update(case.env_extra)
proc = subprocess.run(
[headroom_bin, *case.argv],
[resolved_bin, *case.argv],
env=env,
cwd=str(project),
capture_output=True,
@ -169,6 +194,8 @@ def _run_in_scratch(
for shim_name, behavior in case.shims.items():
make_shim(shim_name, shim_dir, behavior=behavior)
resolved_bin = _resolve_headroom_bin(headroom_bin)
with with_clean_path([shim_dir]) as env:
env["HOME"] = str(home)
env["USERPROFILE"] = str(home)
@ -176,7 +203,7 @@ def _run_in_scratch(
env.update(case.env_extra)
proc = subprocess.run(
[headroom_bin, *case.argv],
[resolved_bin, *case.argv],
env=env,
cwd=str(project),
capture_output=True,

View file

@ -24,10 +24,15 @@ COPY headroom ./headroom
COPY .claude-plugin ./.claude-plugin
COPY .github/plugin ./.github/plugin
COPY plugins/headroom-agent-hooks ./plugins/headroom-agent-hooks
# The init e2e harness imports from e2e._lib; both directories must be
# present and each must contain an __init__.py so Python sees them as
# packages rooted at /workspace.
COPY e2e/__init__.py ./e2e/__init__.py
COPY e2e/_lib ./e2e/_lib
COPY e2e/init ./e2e/init
RUN python -m venv /opt/headroom-venv && \
/opt/headroom-venv/bin/python -m pip install --upgrade pip && \
/opt/headroom-venv/bin/python -m pip install --upgrade "pip<25" && \
/opt/headroom-venv/bin/python -m pip install -e ".[proxy]"
CMD ["python", "e2e/init/run.py"]

View file

@ -1,236 +1,336 @@
from __future__ import annotations
import json
import os
import stat
import subprocess
import sys
import tempfile
import textwrap
from pathlib import Path
from headroom.cli import init as init_cli
REPO_ROOT = Path("/workspace")
HEADROOM = "headroom"
def log(message: str) -> None:
print(f"[init-e2e] {message}", flush=True)
def run(
cmd: list[str],
*,
env: dict[str, str],
cwd: Path,
timeout: int = 180,
) -> subprocess.CompletedProcess[str]:
log(f"$ {' '.join(cmd)}")
result = subprocess.run(
cmd,
env=env,
cwd=str(cwd),
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
timeout=timeout,
)
if result.stdout.strip():
print(result.stdout.rstrip(), flush=True)
if result.stderr.strip():
print(result.stderr.rstrip(), file=sys.stderr, flush=True)
if result.returncode != 0:
raise RuntimeError(f"Command failed with exit code {result.returncode}: {' '.join(cmd)}")
return result
def assert_true(condition: bool, message: str) -> None:
if not condition:
raise AssertionError(message)
def write_executable(path: Path, content: str) -> None:
path.write_text(content, encoding="utf-8")
path.chmod(path.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
def read_jsonl(path: Path) -> list[dict[str, object]]:
if not path.exists():
return []
return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line]
def create_agent_shims(shim_dir: Path, log_path: Path) -> None:
shim = textwrap.dedent(
"""\
#!/usr/bin/env python3
from __future__ import annotations
import json
import os
import sys
from pathlib import Path
record = {
"tool": Path(sys.argv[0]).name,
"argv": sys.argv[1:],
"cwd": os.getcwd(),
}
log_path = Path(os.environ["HEADROOM_INIT_E2E_LOG"])
log_path.parent.mkdir(parents=True, exist_ok=True)
with log_path.open("a", encoding="utf-8") as handle:
handle.write(json.dumps(record) + "\\n")
print(f"{record['tool']} shim executed")
raise SystemExit(0)
"""
)
shim_dir.mkdir(parents=True, exist_ok=True)
for name in ("claude", "copilot"):
write_executable(shim_dir / name, shim)
def expect_hook_command(command: str, profile: str) -> None:
assert_true("init hook ensure" in command, f"missing init hook ensure in: {command}")
assert_true(f"--profile {profile}" in command, f"missing profile {profile} in: {command}")
def read_manifest(home_dir: Path, profile: str) -> dict[str, object]:
path = home_dir / ".headroom" / "deploy" / profile / "manifest.json"
assert_true(path.exists(), f"Expected manifest at {path}")
return json.loads(path.read_text(encoding="utf-8"))
def verify_claude_local(home_dir: Path, project_dir: Path, shim_log: Path) -> None:
settings = json.loads(
(project_dir / ".claude" / "settings.local.json").read_text(encoding="utf-8")
)
assert_true(
settings["env"]["ANTHROPIC_BASE_URL"] == "http://127.0.0.1:9011",
"Claude local settings should point at the requested proxy port",
)
session_start = settings["hooks"]["SessionStart"][0]["hooks"][0]["command"]
pre_tool = settings["hooks"]["PreToolUse"][0]["hooks"][0]["command"]
profile = init_cli._local_profile(project_dir)
expect_hook_command(session_start, profile)
expect_hook_command(pre_tool, profile)
manifest = read_manifest(home_dir, profile)
assert_true("claude" in manifest["targets"], "Claude init should register the claude target")
claude_calls = [record["argv"] for record in read_jsonl(shim_log) if record["tool"] == "claude"]
assert_true(
claude_calls
== [
["plugin", "marketplace", "add", str(REPO_ROOT)],
["plugin", "install", "headroom@headroom-marketplace", "--scope", "local"],
],
f"Unexpected Claude install commands: {claude_calls}",
)
def verify_copilot_global(home_dir: Path, shim_log: Path) -> None:
config = json.loads((home_dir / ".copilot" / "config.json").read_text(encoding="utf-8"))
assert_true(
"SessionStart" in config["hooks"], "Copilot config should include SessionStart hooks"
)
assert_true("PreToolUse" in config["hooks"], "Copilot config should include PreToolUse hooks")
session_start = config["hooks"]["SessionStart"][0]["command"]
expect_hook_command(session_start, "init-user")
for shell_file in (home_dir / ".bashrc", home_dir / ".zshrc", home_dir / ".profile"):
content = shell_file.read_text(encoding="utf-8")
assert_true(
'export COPILOT_PROVIDER_TYPE="openai"' in content,
f"{shell_file.name} should contain the Copilot provider type",
)
assert_true(
'export COPILOT_PROVIDER_BASE_URL="http://127.0.0.1:9005/v1"' in content,
f"{shell_file.name} should contain the Copilot provider base URL",
)
assert_true(
'export COPILOT_PROVIDER_WIRE_API="completions"' in content,
f"{shell_file.name} should contain the Copilot wire API",
)
copilot_calls = [
record["argv"] for record in read_jsonl(shim_log) if record["tool"] == "copilot"
]
assert_true(
copilot_calls
== [
["plugin", "marketplace", "add", str(REPO_ROOT)],
["plugin", "install", "headroom@headroom-marketplace"],
],
f"Unexpected Copilot install commands: {copilot_calls}",
)
def verify_codex_local(home_dir: Path, project_dir: Path) -> None:
config_path = project_dir / ".codex" / "config.toml"
hooks_path = project_dir / ".codex" / "hooks.json"
config = config_path.read_text(encoding="utf-8")
hooks = json.loads(hooks_path.read_text(encoding="utf-8"))
profile = init_cli._local_profile(project_dir)
assert_true(
'base_url = "http://127.0.0.1:9012/v1"' in config,
"Codex config should point at the requested proxy port",
)
assert_true(
config.count("[features]") == 1, "Codex config should keep a single [features] table"
)
assert_true("codex_hooks = true" in config, "Codex config should enable codex_hooks")
command = hooks["hooks"]["SessionStart"][0]["hooks"][0]["command"]
expect_hook_command(command, profile)
manifest = read_manifest(home_dir, profile)
targets = manifest["targets"]
assert_true(set(targets) == {"claude", "codex"}, f"Unexpected merged targets: {targets}")
def main() -> None:
with tempfile.TemporaryDirectory(prefix="headroom-init-e2e-") as temp_root_raw:
temp_root = Path(temp_root_raw)
home_dir = temp_root / "home"
project_dir = temp_root / "project"
shim_dir = temp_root / "bin"
shim_log = temp_root / "shim-log.jsonl"
home_dir.mkdir(parents=True)
project_dir.mkdir(parents=True)
create_agent_shims(shim_dir, shim_log)
env = os.environ.copy()
env["HOME"] = str(home_dir)
env["USERPROFILE"] = str(home_dir)
env["HEADROOM_INIT_E2E_LOG"] = str(shim_log)
env["PATH"] = f"{shim_dir}:{env['PATH']}"
run([HEADROOM, "init", "--port", "9011", "claude"], env=env, cwd=project_dir)
verify_claude_local(home_dir, project_dir, shim_log)
run(
[
HEADROOM,
"init",
"-g",
"--port",
"9005",
"--backend",
"openai",
"copilot",
],
env=env,
cwd=project_dir,
)
verify_copilot_global(home_dir, shim_log)
run([HEADROOM, "init", "--port", "9012", "codex"], env=env, cwd=project_dir)
verify_codex_local(home_dir, project_dir)
log("Init e2e completed successfully")
if __name__ == "__main__":
main()
"""Docker e2e cases for ``headroom init``.
Every case is described declaratively with :class:`Case` from
``e2e/_lib/harness.py``. Three groups run in order:
1. **existing sequence**: preserves the original scenario that exercised
``headroom init claude`` (local) -> ``init -g copilot`` (global) ->
``init codex`` (local), sharing scratch state so manifest-merge is
exercised end-to-end.
2. **bare ``init -g`` detection**: verifies the UX regression from #245
stays fixed both "no shims found" (friendly error, exit 1) and
"all shims found" (exit 0, all four agents configured).
3. **per-subcommand**: one case per ``init -g <agent>`` with only that
agent's shim on PATH, so the explicit path is covered independently.
The fourth group covers ``--verbose`` output going to stderr.
Run directly: ``python e2e/init/run.py`` (inside the Docker image built
from ``e2e/init/Dockerfile``).
"""
from __future__ import annotations
import json
import sys
from pathlib import Path
# Add repo root to sys.path so the harness import works whether the file is
# invoked as ``python e2e/init/run.py`` or ``python -m e2e.init.run``.
_REPO_ROOT = Path(__file__).resolve().parents[2]
if str(_REPO_ROOT) not in sys.path:
sys.path.insert(0, str(_REPO_ROOT))
from e2e._lib import ( # noqa: E402
Case,
CaseContext,
run_case_sequence,
run_cases,
)
from headroom.cli import init as init_cli # noqa: E402
# ----- helpers reused across cases --------------------------------------------
# Docker image builds the workspace at /workspace; the marketplace source
# falls back to that repo checkout when a local marketplace manifest is found.
REPO_ROOT_IN_CONTAINER = Path("/workspace")
def _read_jsonl(path: Path) -> list[dict[str, object]]:
if not path.exists():
return []
return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line]
def _expect_hook_command(command: str, profile: str) -> None:
if "init hook ensure" not in command:
raise AssertionError(f"missing 'init hook ensure' in: {command}")
if f"--profile {profile}" not in command:
raise AssertionError(f"missing '--profile {profile}' in: {command}")
def _read_manifest(home: Path, profile: str) -> dict[str, object]:
path = home / ".headroom" / "deploy" / profile / "manifest.json"
if not path.exists():
raise AssertionError(f"Expected manifest at {path}")
return json.loads(path.read_text(encoding="utf-8"))
# ----- existing-flow assertions (ported verbatim from the old run.py) ---------
def _verify_claude_local(ctx: CaseContext) -> None:
settings_path = ctx.project / ".claude" / "settings.local.json"
settings = json.loads(settings_path.read_text(encoding="utf-8"))
if settings["env"]["ANTHROPIC_BASE_URL"] != "http://127.0.0.1:9011":
raise AssertionError(
f"Claude local settings should point at port 9011, got "
f"{settings['env']['ANTHROPIC_BASE_URL']!r}"
)
session_start = settings["hooks"]["SessionStart"][0]["hooks"][0]["command"]
pre_tool = settings["hooks"]["PreToolUse"][0]["hooks"][0]["command"]
profile = init_cli._local_profile(ctx.project)
_expect_hook_command(session_start, profile)
_expect_hook_command(pre_tool, profile)
manifest = _read_manifest(ctx.home, profile)
if "claude" not in manifest["targets"]:
raise AssertionError(
f"Claude init should register the claude target, got {manifest['targets']}"
)
claude_calls = [
record["argv"] for record in _read_jsonl(ctx.shim_log) if record["tool"] == "claude"
]
expected = [
["plugin", "marketplace", "add", str(REPO_ROOT_IN_CONTAINER)],
["plugin", "install", "headroom@headroom-marketplace", "--scope", "local"],
]
if claude_calls != expected:
raise AssertionError(f"Unexpected Claude install commands: {claude_calls}")
def _verify_copilot_global(ctx: CaseContext) -> None:
config = json.loads((ctx.home / ".copilot" / "config.json").read_text(encoding="utf-8"))
if "SessionStart" not in config["hooks"]:
raise AssertionError("Copilot config missing SessionStart hooks")
if "PreToolUse" not in config["hooks"]:
raise AssertionError("Copilot config missing PreToolUse hooks")
session_start = config["hooks"]["SessionStart"][0]["command"]
_expect_hook_command(session_start, "init-user")
for shell_file in (ctx.home / ".bashrc", ctx.home / ".zshrc", ctx.home / ".profile"):
content = shell_file.read_text(encoding="utf-8")
for literal in (
'export COPILOT_PROVIDER_TYPE="openai"',
'export COPILOT_PROVIDER_BASE_URL="http://127.0.0.1:9005/v1"',
'export COPILOT_PROVIDER_WIRE_API="completions"',
):
if literal not in content:
raise AssertionError(f"{shell_file.name} missing {literal!r}")
copilot_calls = [
record["argv"] for record in _read_jsonl(ctx.shim_log) if record["tool"] == "copilot"
]
expected = [
["plugin", "marketplace", "add", str(REPO_ROOT_IN_CONTAINER)],
["plugin", "install", "headroom@headroom-marketplace"],
]
if copilot_calls != expected:
raise AssertionError(f"Unexpected Copilot install commands: {copilot_calls}")
def _verify_codex_local(ctx: CaseContext) -> None:
config = (ctx.project / ".codex" / "config.toml").read_text(encoding="utf-8")
hooks = json.loads((ctx.project / ".codex" / "hooks.json").read_text(encoding="utf-8"))
profile = init_cli._local_profile(ctx.project)
if 'base_url = "http://127.0.0.1:9012/v1"' not in config:
raise AssertionError("Codex config should point at the requested proxy port (9012)")
if config.count("[features]") != 1:
raise AssertionError("Codex config should keep a single [features] table")
if "codex_hooks = true" not in config:
raise AssertionError("Codex config should enable codex_hooks")
command = hooks["hooks"]["SessionStart"][0]["hooks"][0]["command"]
_expect_hook_command(command, profile)
manifest = _read_manifest(ctx.home, profile)
targets = manifest["targets"]
if set(targets) != {"claude", "codex"}:
raise AssertionError(f"Unexpected merged targets: {targets}")
# ----- new cases (issue #245 fix + per-subcommand coverage) -------------------
def _verify_claude_global(ctx: CaseContext) -> None:
settings = json.loads((ctx.home / ".claude" / "settings.json").read_text(encoding="utf-8"))
if settings["env"]["ANTHROPIC_BASE_URL"] != "http://127.0.0.1:8787":
raise AssertionError(
f"Claude user settings should default to port 8787, got "
f"{settings['env']['ANTHROPIC_BASE_URL']!r}"
)
_expect_hook_command(
settings["hooks"]["SessionStart"][0]["hooks"][0]["command"],
init_cli._GLOBAL_PROFILE,
)
def _verify_codex_global(ctx: CaseContext) -> None:
config = (ctx.home / ".codex" / "config.toml").read_text(encoding="utf-8")
if 'base_url = "http://127.0.0.1:8787/v1"' not in config:
raise AssertionError("Codex user config should point at port 8787 by default")
if "codex_hooks = true" not in config:
raise AssertionError("Codex user config should enable codex_hooks")
hooks = json.loads((ctx.home / ".codex" / "hooks.json").read_text(encoding="utf-8"))
_expect_hook_command(
hooks["hooks"]["SessionStart"][0]["hooks"][0]["command"],
init_cli._GLOBAL_PROFILE,
)
# ----- case tables ------------------------------------------------------------
def existing_sequence_cases() -> list[Case]:
"""Preserves the original run.py scenario in one shared scratch."""
return [
Case(
name="seq_claude_local",
argv=["init", "--port", "9011", "claude"],
shims={"claude": "record-args", "copilot": "record-args"},
expected_exit=0,
expected_stdout_contains=["Configured Claude Code (local scope)"],
extra_assertions=[_verify_claude_local],
),
Case(
name="seq_copilot_global",
argv=["init", "-g", "--port", "9005", "--backend", "openai", "copilot"],
shims={}, # reuse shims from prior case in the sequence
expected_exit=0,
expected_stdout_contains=["Configured GitHub Copilot CLI (user scope)"],
extra_assertions=[_verify_copilot_global],
),
Case(
name="seq_codex_local",
argv=["init", "--port", "9012", "codex"],
shims={},
expected_exit=0,
expected_stdout_contains=["Configured Codex (local scope)"],
extra_assertions=[_verify_codex_local],
),
]
def bare_init_g_cases() -> list[Case]:
"""Bare ``headroom init -g`` — the direct coverage of issue #245."""
return [
Case(
name="bare_init_g_no_shims",
argv=["init", "-g"],
shims={}, # nothing on PATH
expected_exit=1,
expected_stderr_contains=[
# every target should be listed so the user knows what was tried
"claude",
"codex",
"copilot",
"openclaw",
# concrete escape hatch — exactly what the user should type next
"headroom init -g claude",
# confirm -g itself is still the right flag
"-g",
],
),
Case(
name="bare_init_g_with_all_shims",
argv=["init", "-g"],
shims={
"claude": "record-args",
"codex": "noop",
"copilot": "record-args",
"openclaw": "noop",
},
expected_exit=0,
expected_stdout_contains=[
"Configured Claude Code (user scope)",
"Configured GitHub Copilot CLI (user scope)",
"Configured Codex (user scope)",
],
),
]
def per_subcommand_cases() -> list[Case]:
"""One case per ``headroom init -g <agent>`` with only that agent's shim."""
return [
Case(
name="init_g_claude_explicit",
argv=["init", "-g", "claude"],
shims={"claude": "record-args"},
expected_exit=0,
expected_stdout_contains=["Configured Claude Code (user scope)"],
expected_files=["{home}/.claude/settings.json"],
extra_assertions=[_verify_claude_global],
),
Case(
name="init_g_codex_explicit",
argv=["init", "-g", "codex"],
shims={"codex": "noop"},
expected_exit=0,
expected_stdout_contains=["Configured Codex (user scope)"],
expected_files=[
"{home}/.codex/config.toml",
"{home}/.codex/hooks.json",
],
extra_assertions=[_verify_codex_global],
),
Case(
name="init_g_copilot_explicit",
argv=["init", "-g", "copilot"],
shims={"copilot": "record-args"},
expected_exit=0,
expected_stdout_contains=["Configured GitHub Copilot CLI (user scope)"],
expected_files=["{home}/.copilot/config.json"],
),
# openclaw delegates to `headroom wrap openclaw` which has its own
# (more expensive) init path and isn't stubbable with a simple shim.
# We assert it fails fast with a clear error when not installed, and
# rely on the `bare_init_g_with_all_shims` case (which uses a noop
# openclaw shim + claude/codex/copilot shims) to cover the success
# path alongside the other agents.
Case(
name="init_g_openclaw_missing",
argv=["init", "-g", "openclaw"],
shims={},
expected_exit=1,
),
]
def verbose_cases() -> list[Case]:
"""Verbose flag smoke tests — debug lines should appear on stderr."""
return [
Case(
name="init_verbose_no_shims",
argv=["init", "-v", "-g"],
shims={},
expected_exit=1,
expected_stderr_contains=[
# A few structural markers from the verbose log. Kept loose so
# minor wording tweaks don't break the test.
"detect_init_targets",
"claude",
"global_scope=True",
],
),
]
def main() -> None:
rc = 0
rc |= run_case_sequence(existing_sequence_cases(), label="existing-sequence")
rc |= run_cases(bare_init_g_cases())
rc |= run_cases(per_subcommand_cases())
rc |= run_cases(verbose_cases())
if rc != 0:
raise SystemExit(rc)
print("[e2e] init e2e completed successfully", flush=True)
if __name__ == "__main__":
main()