akadmin/headroom
1
0
Fork 0
mirror of https://github.com/headroomlabs-ai/headroom.git synced 2026-08-27 14:17:10 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix: stabilize cli test isolation

Browse source
This commit is contained in:
Kayzo 2026-04-22 11:28:30 +00:00
parent d17aea2fe2
commit 3045b36b52
5 changed files with 176 additions and 53 deletions
Download patch file Download diff file Expand all files Collapse all files

37
headroom/cli/__init__.py
View file

@ -1,5 +1,42 @@
"""Headroom CLI - Command-line interface for memory and proxy management."""
import sys
from importlib import import_module
from .main import main
_LAZY_SUBMODULES = {
"evals",
"init",
"install",
"learn",
"mcp",
"memory",
"perf",
"proxy",
"tools",
"wrap",
}
__all__ = ["main"]
for _name in _LAZY_SUBMODULES:
_module = sys.modules.get(f"{__name__}.{_name}")
if _module is not None:
globals()[_name] = _module
def __getattr__(name: str) -> object:
if name == "__path__":
raise AttributeError(name)
if name in _LAZY_SUBMODULES:
module = import_module(f"{__name__}.{name}")
globals()[name] = module
return module
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
def __dir__() -> list[str]:
return sorted(set(globals()) | set(__all__) | _LAZY_SUBMODULES)

62
headroom/cli/wrap.py
View file

@ -26,7 +26,7 @@ import time
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any
from typing import Any, cast
# Fix Windows cp1252 encoding — box-drawing characters require UTF-8
if sys.platform == "win32" and hasattr(sys.stdout, "buffer"):
@ -42,6 +42,18 @@ from headroom.copilot_auth import has_oauth_auth, resolve_client_bearer_token
from .main import main
def _live_wrap_module() -> Any:
"""Return the current live wrap module instance.
CLI tests sometimes reload `headroom.cli.wrap` while still invoking Click
command callbacks that were registered from an older module instance. By
resolving helper calls through `sys.modules[__name__]`, patched helpers on
the live module continue to affect those callbacks.
"""
return cast(Any, sys.modules[__name__])
def _print_telemetry_notice() -> None:
"""Print a telemetry notice when anonymous telemetry is enabled.
@ -651,7 +663,8 @@ def _recover_persistent_proxy(port: int) -> bool:
from headroom.install.runtime import start_detached_agent, start_persistent_docker, wait_ready
from headroom.install.supervisors import start_supervisor
manifest = _find_persistent_manifest(port)
helpers = _live_wrap_module()
manifest = helpers._find_persistent_manifest(port)
if manifest is None:
return False
@ -735,24 +748,30 @@ def _ensure_proxy(
openai_api_url: str | None = None,
) -> subprocess.Popen | None:
"""Start or verify proxy. Returns process handle if we started it."""
helpers = _live_wrap_module()
if not no_proxy:
manifest = _find_persistent_manifest(port)
manifest = helpers._find_persistent_manifest(port)
if manifest is not None:
from headroom.install.health import probe_ready
if probe_ready(manifest.health_url):
click.echo(f" Proxy already running on port {port}")
return None
if _recover_persistent_proxy(port):
if helpers._recover_persistent_proxy(port):
return None
raise click.ClickException(
f"Persistent deployment '{manifest.profile}' on port {port} is not healthy."
if helpers._check_proxy(port):
raise click.ClickException(
f"Persistent deployment '{manifest.profile}' on port {port} is not healthy."
)
click.echo(
f" Warning: persistent deployment '{manifest.profile}' on port {port} "
"is stale; starting a fresh proxy instead."
)
if _check_proxy(port):
if helpers._check_proxy(port):
# Proxy is running — check if it has the features we need
needs_restart = False
running_config = _query_proxy_config(port)
running_config = helpers._query_proxy_config(port)
if running_config is not None:
missing = []
@ -776,7 +795,7 @@ def _ensure_proxy(
proxy_pid = running_config.get("pid")
if proxy_pid is not None:
if not _kill_proxy_by_pid(int(proxy_pid), port):
if not helpers._kill_proxy_by_pid(int(proxy_pid), port):
raise click.ClickException(
f"Failed to stop existing proxy (PID {proxy_pid}) on port {port}. "
"Stop it manually and retry."
@ -799,7 +818,7 @@ def _ensure_proxy(
# Start (or restart) the proxy with the requested flags
click.echo(f" Starting Headroom proxy on port {port}...")
try:
proc = _start_proxy(
proc = helpers._start_proxy(
port,
learn=learn,
memory=memory,
@ -816,7 +835,7 @@ def _ensure_proxy(
click.echo(f" Error: {e}")
raise SystemExit(1) from e
else:
if not _check_proxy(port):
if not helpers._check_proxy(port):
click.echo(f" Warning: No proxy detected on port {port}")
return None
@ -1234,9 +1253,11 @@ def claude(
headroom wrap claude --code-graph # With code graph intelligence
headroom wrap claude --no-rtk # Skip rtk (proxy only)
"""
helpers = _live_wrap_module()
if prepare_only:
if not no_rtk:
_prepare_wrap_rtk(verbose=verbose, label="Claude")
helpers._prepare_wrap_rtk(verbose=verbose, label="Claude")
return
claude_bin = shutil.which("claude")
@ -1247,7 +1268,7 @@ def claude(
# Setup rtk before launching (Claude-specific)
proxy_holder: list[subprocess.Popen | None] = [None]
cleanup = _make_cleanup(proxy_holder, port)
cleanup = helpers._make_cleanup(proxy_holder, port)
signal.signal(signal.SIGINT, cleanup)
signal.signal(signal.SIGTERM, cleanup)
@ -1300,25 +1321,30 @@ def claude(
click.echo(" ╚═══════════════════════════════════════════════╝")
click.echo()
proxy_holder[0] = _ensure_proxy(
port, no_proxy, learn=learn, memory=memory, agent_type="claude", code_graph=code_graph
proxy_holder[0] = helpers._ensure_proxy(
port,
no_proxy,
learn=learn,
memory=memory,
agent_type="claude",
code_graph=code_graph,
)
if not no_rtk:
click.echo(" Setting up rtk...")
_setup_rtk(verbose=verbose)
helpers._setup_rtk(verbose=verbose)
elif verbose:
click.echo(" Skipping rtk (--no-rtk)")
if code_graph:
_setup_code_graph(verbose=verbose)
helpers._setup_code_graph(verbose=verbose)
click.echo()
click.echo(" Launching Claude Code (API routed through Headroom)...")
click.echo(f" ANTHROPIC_BASE_URL=http://127.0.0.1:{port}")
if claude_args:
click.echo(f" Extra args: {' '.join(claude_args)}")
_print_telemetry_notice()
helpers._print_telemetry_notice()
click.echo()
env = os.environ.copy()

78
tests/test_cli/test_wrap_copilot.py
View file

@ -14,29 +14,45 @@ from click.testing import CliRunner
from headroom.copilot_auth import DEFAULT_API_URL
fake_main_module = types.ModuleType("headroom.cli.main")
fake_main_module.main = click.Group()
sys.modules["headroom.cli.main"] = fake_main_module
sys.modules.pop("headroom.cli", None)
sys.modules.pop("headroom.cli.wrap", None)
wrap_cli = importlib.import_module("headroom.cli.wrap")
main = fake_main_module.main
@pytest.fixture
def runner() -> CliRunner:
return CliRunner()
def _load_wrap_cli(monkeypatch: pytest.MonkeyPatch) -> tuple[types.ModuleType, click.Group]:
fake_main_module = types.ModuleType("headroom.cli.main")
fake_main_module.main = click.Group()
monkeypatch.setitem(sys.modules, "headroom.cli.main", fake_main_module)
monkeypatch.delitem(sys.modules, "headroom.cli", raising=False)
monkeypatch.delitem(sys.modules, "headroom.cli.wrap", raising=False)
importlib.invalidate_caches()
wrap_cli = importlib.import_module("headroom.cli.wrap")
return wrap_cli, fake_main_module.main
@pytest.fixture
def copilot_cli(monkeypatch: pytest.MonkeyPatch) -> tuple[types.ModuleType, click.Group]:
return _load_wrap_cli(monkeypatch)
@pytest.fixture(autouse=True)
def no_running_proxy(monkeypatch: pytest.MonkeyPatch) -> None:
def no_running_proxy(
monkeypatch: pytest.MonkeyPatch, copilot_cli: tuple[types.ModuleType, click.Group]
) -> None:
wrap_cli, _ = copilot_cli
monkeypatch.setattr(wrap_cli, "_check_proxy", lambda _port: False)
def test_wrap_copilot_auto_anthropic_injects_instructions(
runner: CliRunner, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
copilot_cli: tuple[types.ModuleType, click.Group],
) -> None:
wrap_cli, main = copilot_cli
monkeypatch.chdir(tmp_path)
monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-test-dummy")
captured: dict[str, object] = {}
@ -70,8 +86,11 @@ def test_wrap_copilot_auto_anthropic_injects_instructions(
def test_wrap_copilot_openai_backend_sets_completions_env(
runner: CliRunner, monkeypatch: pytest.MonkeyPatch
runner: CliRunner,
monkeypatch: pytest.MonkeyPatch,
copilot_cli: tuple[types.ModuleType, click.Group],
) -> None:
_, main = copilot_cli
monkeypatch.setenv("OPENAI_API_KEY", "sk-test-dummy")
captured: dict[str, object] = {}
@ -112,8 +131,11 @@ def test_wrap_copilot_openai_backend_sets_completions_env(
def test_wrap_copilot_auto_detects_running_proxy_backend(
runner: CliRunner, monkeypatch: pytest.MonkeyPatch
runner: CliRunner,
monkeypatch: pytest.MonkeyPatch,
copilot_cli: tuple[types.ModuleType, click.Group],
) -> None:
_, main = copilot_cli
monkeypatch.setenv("OPENAI_API_KEY", "sk-test-dummy")
captured: dict[str, object] = {}
@ -138,8 +160,11 @@ def test_wrap_copilot_auto_detects_running_proxy_backend(
def test_wrap_copilot_prefers_existing_oauth_session(
runner: CliRunner, monkeypatch: pytest.MonkeyPatch
runner: CliRunner,
monkeypatch: pytest.MonkeyPatch,
copilot_cli: tuple[types.ModuleType, click.Group],
) -> None:
_, main = copilot_cli
monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-test-dummy")
captured: dict[str, object] = {}
@ -168,7 +193,9 @@ def test_wrap_copilot_prefers_existing_oauth_session(
def test_wrap_copilot_translated_backend_still_requires_byok(
runner: CliRunner,
copilot_cli: tuple[types.ModuleType, click.Group],
) -> None:
_, main = copilot_cli
with patch("headroom.cli.wrap.shutil.which", return_value="copilot"):
with patch("headroom.cli.wrap.has_oauth_auth", return_value=True):
result = runner.invoke(
@ -188,7 +215,11 @@ def test_wrap_copilot_translated_backend_still_requires_byok(
assert "Copilot BYOK mode requires a provider API key" in result.output
def test_wrap_copilot_rejects_wire_api_for_anthropic_provider(runner: CliRunner) -> None:
def test_wrap_copilot_rejects_wire_api_for_anthropic_provider(
runner: CliRunner,
copilot_cli: tuple[types.ModuleType, click.Group],
) -> None:
_, main = copilot_cli
with patch("headroom.cli.wrap.shutil.which", return_value="copilot"):
result = runner.invoke(
main,
@ -207,7 +238,11 @@ def test_wrap_copilot_rejects_wire_api_for_anthropic_provider(runner: CliRunner)
assert "--wire-api is only valid" in result.output
def test_wrap_copilot_rejects_responses_for_translated_backends(runner: CliRunner) -> None:
def test_wrap_copilot_rejects_responses_for_translated_backends(
runner: CliRunner,
copilot_cli: tuple[types.ModuleType, click.Group],
) -> None:
_, main = copilot_cli
with patch("headroom.cli.wrap.shutil.which", return_value="copilot"):
result = runner.invoke(
main,
@ -229,8 +264,11 @@ def test_wrap_copilot_rejects_responses_for_translated_backends(runner: CliRunne
def test_wrap_copilot_clears_stale_wire_api_in_anthropic_mode(
runner: CliRunner, monkeypatch: pytest.MonkeyPatch
runner: CliRunner,
monkeypatch: pytest.MonkeyPatch,
copilot_cli: tuple[types.ModuleType, click.Group],
) -> None:
_, main = copilot_cli
monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-test-dummy")
captured: dict[str, object] = {}
@ -255,7 +293,11 @@ def test_wrap_copilot_clears_stale_wire_api_in_anthropic_mode(
assert "COPILOT_PROVIDER_WIRE_API" not in env
def test_wrap_copilot_fails_when_binary_missing(runner: CliRunner) -> None:
def test_wrap_copilot_fails_when_binary_missing(
runner: CliRunner,
copilot_cli: tuple[types.ModuleType, click.Group],
) -> None:
_, main = copilot_cli
with patch("headroom.cli.wrap.shutil.which", return_value=None):
result = runner.invoke(main, ["wrap", "copilot", "--", "--model", "gpt-4o"])

50
tests/test_cli/test_wrap_persistent.py
View file

@ -2,7 +2,7 @@ from __future__ import annotations
import click
from headroom.cli.wrap import _ensure_proxy, _find_persistent_manifest, _recover_persistent_proxy
import headroom.cli.wrap as wrap_cli
class _Manifest:
@ -15,8 +15,8 @@ class _Manifest:
def test_ensure_proxy_recovers_matching_persistent_deployment(monkeypatch) -> None:
calls: list[str] = []
monkeypatch.setattr("headroom.cli.wrap._check_proxy", lambda port: False)
monkeypatch.setattr("headroom.cli.wrap._find_persistent_manifest", lambda port: _Manifest())
monkeypatch.setattr(wrap_cli, "_check_proxy", lambda port: False)
monkeypatch.setattr(wrap_cli, "_find_persistent_manifest", lambda port: _Manifest())
monkeypatch.setattr("headroom.install.health.probe_ready", lambda url: False)
monkeypatch.setattr(
"headroom.install.supervisors.start_supervisor",
@ -26,13 +26,14 @@ def test_ensure_proxy_recovers_matching_persistent_deployment(monkeypatch) -> No
"headroom.install.runtime.wait_ready", lambda manifest, timeout_seconds=45: True
)
monkeypatch.setattr(
"headroom.cli.wrap._start_proxy",
wrap_cli,
"_start_proxy",
lambda *args, **kwargs: (_ for _ in ()).throw(
AssertionError("ephemeral proxy should not start")
),
)
result = _ensure_proxy(8787, False)
result = wrap_cli._ensure_proxy(8787, False)
assert result is None
assert calls == ["start:default"]
@ -41,8 +42,8 @@ def test_ensure_proxy_recovers_matching_persistent_deployment(monkeypatch) -> No
def test_ensure_proxy_recovers_persistent_deployment_when_socket_is_bound(monkeypatch) -> None:
calls: list[str] = []
monkeypatch.setattr("headroom.cli.wrap._check_proxy", lambda port: True)
monkeypatch.setattr("headroom.cli.wrap._find_persistent_manifest", lambda port: _Manifest())
monkeypatch.setattr(wrap_cli, "_check_proxy", lambda port: True)
monkeypatch.setattr(wrap_cli, "_find_persistent_manifest", lambda port: _Manifest())
monkeypatch.setattr("headroom.install.health.probe_ready", lambda url: False)
monkeypatch.setattr(
"headroom.install.supervisors.start_supervisor",
@ -52,26 +53,41 @@ def test_ensure_proxy_recovers_persistent_deployment_when_socket_is_bound(monkey
"headroom.install.runtime.wait_ready", lambda manifest, timeout_seconds=45: True
)
result = _ensure_proxy(8787, False)
result = wrap_cli._ensure_proxy(8787, False)
assert result is None
assert calls == ["start:default"]
def test_ensure_proxy_rejects_unhealthy_persistent_deployment(monkeypatch) -> None:
monkeypatch.setattr("headroom.cli.wrap._check_proxy", lambda port: True)
monkeypatch.setattr("headroom.cli.wrap._find_persistent_manifest", lambda port: _Manifest())
monkeypatch.setattr(wrap_cli, "_check_proxy", lambda port: True)
monkeypatch.setattr(wrap_cli, "_find_persistent_manifest", lambda port: _Manifest())
monkeypatch.setattr("headroom.install.health.probe_ready", lambda url: False)
monkeypatch.setattr("headroom.cli.wrap._recover_persistent_proxy", lambda port: False)
monkeypatch.setattr(wrap_cli, "_recover_persistent_proxy", lambda port: False)
try:
_ensure_proxy(8787, False)
wrap_cli._ensure_proxy(8787, False)
except click.ClickException as exc:
assert "is not healthy" in str(exc)
else:
raise AssertionError("expected unhealthy persistent deployment to raise")
def test_ensure_proxy_falls_back_when_persistent_manifest_is_stale(monkeypatch) -> None:
calls: list[str] = []
monkeypatch.setattr(wrap_cli, "_check_proxy", lambda port: False)
monkeypatch.setattr(wrap_cli, "_find_persistent_manifest", lambda port: _Manifest())
monkeypatch.setattr("headroom.install.health.probe_ready", lambda url: False)
monkeypatch.setattr(wrap_cli, "_recover_persistent_proxy", lambda port: False)
monkeypatch.setattr(wrap_cli, "_start_proxy", lambda *args, **kwargs: calls.append("start"))
result = wrap_cli._ensure_proxy(8787, False)
assert result is None
assert calls == ["start"]
def test_find_persistent_manifest_prefers_default_profile(monkeypatch) -> None:
class DefaultManifest:
profile = "default"
@ -86,23 +102,23 @@ def test_find_persistent_manifest_prefers_default_profile(monkeypatch) -> None:
lambda: [OtherManifest(), DefaultManifest()],
)
manifest = _find_persistent_manifest(8787)
manifest = wrap_cli._find_persistent_manifest(8787)
assert manifest.profile == "default"
def test_recover_persistent_proxy_reuses_healthy_deployment(monkeypatch) -> None:
monkeypatch.setattr("headroom.cli.wrap._find_persistent_manifest", lambda port: _Manifest())
monkeypatch.setattr(wrap_cli, "_find_persistent_manifest", lambda port: _Manifest())
monkeypatch.setattr("headroom.install.health.probe_ready", lambda url: True)
assert _recover_persistent_proxy(8787) is True
assert wrap_cli._recover_persistent_proxy(8787) is True
def test_recover_persistent_proxy_warns_for_task_deployment(monkeypatch) -> None:
class TaskManifest(_Manifest):
supervisor_kind = "task"
monkeypatch.setattr("headroom.cli.wrap._find_persistent_manifest", lambda port: TaskManifest())
monkeypatch.setattr(wrap_cli, "_find_persistent_manifest", lambda port: TaskManifest())
monkeypatch.setattr("headroom.install.health.probe_ready", lambda url: False)
assert _recover_persistent_proxy(8787) is False
assert wrap_cli._recover_persistent_proxy(8787) is False

2
tests/test_release_workflows.py
View file

@ -53,6 +53,8 @@ def test_macos_native_wrapper_dependency_install_retries_pypi_downloads() -> Non
content = (ROOT / ".github" / "workflows" / "ci.yml").read_text(encoding="utf-8")
assert "python -m pip install --retries 10 --timeout 60 pytest" in content
def test_ci_commitlint_skips_default_github_merge_commits() -> None:
content = (ROOT / ".github" / "workflows" / "ci.yml").read_text(encoding="utf-8")