mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-27 14:17:10 -04:00
feat(cli): add headroom doctor setup diagnostics (#926)
## Description Headroom fails silently: a client not routed through the proxy (or a proxy running stale code) keeps working — it just stops saving tokens. State that determines whether you are actually saving lives in five places nothing reconciles. `headroom doctor` correlates them in one command (the diagnostic idiom of `claude doctor` / `pnpm doctor`, and the repo's own `headroom tools doctor`). Closes # <!-- no tracked issue; setup-diagnosis gap found this session --> ## Type of Change - [ ] Bug fix (non-breaking change that fixes an issue) - [x] New feature (non-breaking change that adds functionality) - [ ] Breaking change (fix or feature that would cause existing functionality to change) - [ ] Documentation update - [ ] Performance improvement - [ ] Code refactoring (no functional changes) ## Changes Made - `headroom/cli/doctor.py`: new command with 8 pure checks (proxy liveness, version drift, claude/codex routing, shell env, savings flow, budget, deployments); exit codes 0/1/2; `--json`; `--port`/`HEADROOM_PORT`. - `headroom/proxy/cost.py`: expose `budget_limit_usd`/`budget_period` in `CostTracker.stats()` so the budget check can read it (older proxies degrade to a warning). - `headroom/cli/main.py`: register the command. - `tests/test_cli_doctor.py`: 41 tests, zero network (probed payloads / paths / env injected). ## Testing - [x] Unit tests pass (`pytest`) - [x] Linting passes (`ruff check .`) - [x] Type checking passes (`mypy headroom`) - [x] New tests added for new functionality - [x] Manual testing performed ### Test Output ```text $ pytest tests/test_cli_doctor.py -q 41 passed ``` ## Real Behavior Proof - Environment: local macOS, Python 3.11, against a real proxy running for 3 days, branch `feat/doctor-command`. - Exact command / steps: `headroom doctor` (live), plus `pytest tests/test_cli_doctor.py -q`. - Observed result: Correctly flagged real version drift (proxy 0.25.0 vs installed 0.26.0), an unrouted claude client, and a shell `OPENAI_BASE_URL` pointed at a non-Headroom gateway; savings check showed 17.6M tokens / $7.82 saved; exit code 1 (warnings). - Not tested: Windows path handling for client config files (logic is OS-agnostic via pathlib). ## Review Readiness - [x] I have performed a self-review - [x] This PR is ready for human review ## Checklist - [x] My code follows the project's style guidelines - [x] I have performed a self-review of my code - [x] I have commented my code, particularly in hard-to-understand areas - [ ] I have made corresponding changes to the documentation - [x] My changes generate no new warnings - [x] I have added tests that prove my fix is effective or that my feature works - [x] New and existing unit tests pass locally with my changes - [ ] I have updated the CHANGELOG.md if applicable ## Screenshots (if applicable) Terminal output of `headroom doctor` (rich table) can be attached; the rendered table is reproduced in the live-proof bullet above. ## Additional Notes Branched fresh from main. The budget check connects to the enforcement fix in #885. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com> Co-authored-by: JD Davis <mxjerrett@gmail.com> Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
Ashish
GitHub
parent
b0cd0329c7
commit
e45cf4e061
4 changed files with 766 additions and 0 deletions
412
headroom/cli/doctor.py
Normal file
412
headroom/cli/doctor.py
Normal file
|
|
@ -0,0 +1,412 @@
|
|||
"""`headroom doctor` — diagnose whether the local Headroom setup is working.
|
||||
|
||||
Headroom's failure mode is silent: when a client is not routed through the
|
||||
proxy (or the proxy runs stale code), everything still works — you just
|
||||
stop saving tokens. This command correlates the state nothing else
|
||||
reconciles: the proxy process, per-client wrap configs, the current shell
|
||||
environment, savings flow, and budget configuration.
|
||||
|
||||
Exit codes: 0 = all checks pass, 1 = warnings only, 2 = any failure.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
from collections.abc import Mapping
|
||||
from dataclasses import asdict, dataclass
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import click
|
||||
|
||||
from headroom.install.health import probe_json
|
||||
from headroom.install.paths import claude_settings_path, codex_config_path
|
||||
from headroom.install.state import list_manifests
|
||||
from headroom.paths import savings_path
|
||||
|
||||
from .main import get_version, main
|
||||
|
||||
PASS = "pass"
|
||||
WARN = "warn"
|
||||
FAIL = "fail"
|
||||
SKIP = "skip"
|
||||
|
||||
_LOOPBACK_URL_RE = re.compile(r"https?://(?:127\.0\.0\.1|localhost):(\d+)")
|
||||
_CODEX_BASE_URL_RE = re.compile(r'base_url\s*=\s*"https?://(?:127\.0\.0\.1|localhost):(\d+)')
|
||||
|
||||
|
||||
@dataclass
|
||||
class CheckResult:
|
||||
"""One diagnostic outcome."""
|
||||
|
||||
name: str
|
||||
status: str # pass | warn | fail | skip
|
||||
summary: str
|
||||
hint: str | None = None
|
||||
|
||||
|
||||
def _format_uptime(seconds: float) -> str:
|
||||
total = int(seconds)
|
||||
days, rem = divmod(total, 86400)
|
||||
hours, rem = divmod(rem, 3600)
|
||||
minutes = rem // 60
|
||||
if days:
|
||||
return f"{days}d {hours}h"
|
||||
if hours:
|
||||
return f"{hours}h {minutes}m"
|
||||
return f"{minutes}m"
|
||||
|
||||
|
||||
def _format_since(iso_ts: str) -> str | None:
|
||||
try:
|
||||
then = datetime.fromisoformat(iso_ts.replace("Z", "+00:00"))
|
||||
except (ValueError, AttributeError):
|
||||
return None
|
||||
delta = datetime.now(then.tzinfo) - then
|
||||
seconds = max(0, int(delta.total_seconds()))
|
||||
if seconds < 60:
|
||||
return "just now"
|
||||
if seconds < 3600:
|
||||
return f"{seconds // 60}m ago"
|
||||
if seconds < 86400:
|
||||
return f"{seconds // 3600}h ago"
|
||||
return f"{seconds // 86400}d ago"
|
||||
|
||||
|
||||
def check_proxy_liveness(livez: dict[str, Any] | None, base_url: str) -> CheckResult:
|
||||
"""Is the proxy process up and answering /livez?"""
|
||||
if livez is None:
|
||||
return CheckResult(
|
||||
name="proxy",
|
||||
status=FAIL,
|
||||
summary=f"not reachable at {base_url}",
|
||||
hint="start it with: headroom proxy",
|
||||
)
|
||||
version = livez.get("version", "unknown")
|
||||
uptime = livez.get("uptime_seconds")
|
||||
uptime_text = f"up {_format_uptime(uptime)}" if isinstance(uptime, (int, float)) else "up"
|
||||
return CheckResult(
|
||||
name="proxy",
|
||||
status=PASS,
|
||||
summary=f"running at {base_url} ({uptime_text}, v{version})",
|
||||
)
|
||||
|
||||
|
||||
def check_version_drift(livez: dict[str, Any] | None, installed: str) -> CheckResult:
|
||||
"""Does the running proxy match the installed package version?"""
|
||||
if livez is None:
|
||||
return CheckResult(name="version", status=SKIP, summary="proxy not reachable")
|
||||
running = str(livez.get("version") or "unknown")
|
||||
if "unknown" in (running, installed):
|
||||
return CheckResult(
|
||||
name="version",
|
||||
status=WARN,
|
||||
summary=f"cannot compare versions (proxy {running}, installed {installed})",
|
||||
)
|
||||
if running != installed:
|
||||
return CheckResult(
|
||||
name="version",
|
||||
status=WARN,
|
||||
summary=f"version drift: proxy {running}, installed {installed}",
|
||||
hint="restart the proxy to pick up new code: headroom proxy",
|
||||
)
|
||||
return CheckResult(name="version", status=PASS, summary=f"proxy matches installed v{installed}")
|
||||
|
||||
|
||||
def check_claude_routing(settings_path: Path, port: int) -> CheckResult:
|
||||
"""Is Claude Code configured to route through the proxy?"""
|
||||
name = "claude"
|
||||
if not settings_path.exists():
|
||||
return CheckResult(
|
||||
name=name,
|
||||
status=WARN,
|
||||
summary="not routed (no ~/.claude/settings.json)",
|
||||
hint="wrap it: headroom wrap claude",
|
||||
)
|
||||
try:
|
||||
payload = json.loads(settings_path.read_text(encoding="utf-8"))
|
||||
except (OSError, ValueError) as exc:
|
||||
return CheckResult(
|
||||
name=name,
|
||||
status=WARN,
|
||||
summary=f"could not parse {settings_path}: {exc}",
|
||||
)
|
||||
base_url = ""
|
||||
env_block = payload.get("env")
|
||||
if isinstance(env_block, dict):
|
||||
base_url = str(env_block.get("ANTHROPIC_BASE_URL", "") or "")
|
||||
if not base_url:
|
||||
return CheckResult(
|
||||
name=name,
|
||||
status=WARN,
|
||||
summary="not routed (no ANTHROPIC_BASE_URL in settings env)",
|
||||
hint="wrap it: headroom wrap claude",
|
||||
)
|
||||
return _classify_routing_url(name, base_url, port, source=str(settings_path))
|
||||
|
||||
|
||||
def check_codex_routing(config_path: Path, port: int) -> CheckResult:
|
||||
"""Is Codex configured to route through the proxy?
|
||||
|
||||
Detection keys on the ``[model_providers.headroom]`` section, which both
|
||||
writers emit (install's persistent block and wrap's auto-injected block).
|
||||
Substring matching keeps malformed TOML a WARN instead of a crash.
|
||||
"""
|
||||
name = "codex"
|
||||
if not config_path.exists():
|
||||
return CheckResult(
|
||||
name=name,
|
||||
status=WARN,
|
||||
summary="not routed (no ~/.codex/config.toml)",
|
||||
hint="wrap it: headroom wrap codex",
|
||||
)
|
||||
try:
|
||||
text = config_path.read_text(encoding="utf-8", errors="replace")
|
||||
except OSError as exc:
|
||||
return CheckResult(name=name, status=WARN, summary=f"could not read {config_path}: {exc}")
|
||||
if "[model_providers.headroom]" not in text:
|
||||
return CheckResult(
|
||||
name=name,
|
||||
status=WARN,
|
||||
summary="not routed (no Headroom provider in config.toml)",
|
||||
hint="wrap it: headroom wrap codex",
|
||||
)
|
||||
match = _CODEX_BASE_URL_RE.search(text)
|
||||
if match and int(match.group(1)) != port:
|
||||
return CheckResult(
|
||||
name=name,
|
||||
status=WARN,
|
||||
summary=f"routed to port {match.group(1)}, but doctor probed port {port}",
|
||||
hint=f"re-run with: headroom doctor --port {match.group(1)}",
|
||||
)
|
||||
return CheckResult(name=name, status=PASS, summary=f"routed ({config_path})")
|
||||
|
||||
|
||||
def check_shell_env(environ: Mapping[str, str], port: int) -> CheckResult:
|
||||
"""Is the *current shell* pointed at the proxy for ad-hoc runs?"""
|
||||
name = "shell env"
|
||||
for var in ("ANTHROPIC_BASE_URL", "OPENAI_BASE_URL"):
|
||||
value = environ.get(var, "")
|
||||
if value:
|
||||
return _classify_routing_url(name, value, port, source=var)
|
||||
return CheckResult(
|
||||
name=name,
|
||||
status=WARN,
|
||||
summary="ANTHROPIC_BASE_URL / OPENAI_BASE_URL unset — this shell bypasses the proxy",
|
||||
hint=f"export ANTHROPIC_BASE_URL=http://127.0.0.1:{port} (or launch via headroom wrap)",
|
||||
)
|
||||
|
||||
|
||||
def _classify_routing_url(name: str, url: str, port: int, *, source: str) -> CheckResult:
|
||||
match = _LOOPBACK_URL_RE.match(url.strip())
|
||||
if match is None:
|
||||
return CheckResult(
|
||||
name=name,
|
||||
status=WARN,
|
||||
summary=f"points at {url}, not the local Headroom proxy ({source})",
|
||||
)
|
||||
found_port = int(match.group(1))
|
||||
if found_port != port:
|
||||
return CheckResult(
|
||||
name=name,
|
||||
status=WARN,
|
||||
summary=f"routed to port {found_port}, but doctor probed port {port} ({source})",
|
||||
hint=f"re-run with: headroom doctor --port {found_port}",
|
||||
)
|
||||
return CheckResult(name=name, status=PASS, summary=f"routed via {source}")
|
||||
|
||||
|
||||
def check_savings(stats: dict[str, Any] | None, savings_file: Path) -> CheckResult:
|
||||
"""Are savings actually flowing? Lifetime totals + last activity."""
|
||||
name = "savings"
|
||||
payload: dict[str, Any] | None = None
|
||||
source = "proxy /stats"
|
||||
if stats is not None and isinstance(stats.get("persistent_savings"), dict):
|
||||
payload = stats["persistent_savings"]
|
||||
elif savings_file.exists():
|
||||
source = str(savings_file)
|
||||
try:
|
||||
payload = json.loads(savings_file.read_text(encoding="utf-8"))
|
||||
except (OSError, ValueError):
|
||||
return CheckResult(
|
||||
name=name, status=WARN, summary=f"could not read savings file {savings_file}"
|
||||
)
|
||||
if payload is None:
|
||||
return CheckResult(
|
||||
name=name,
|
||||
status=WARN,
|
||||
summary="no savings recorded yet",
|
||||
hint="route a client through the proxy and make a request",
|
||||
)
|
||||
|
||||
lifetime = payload.get("lifetime") or {}
|
||||
tokens = lifetime.get("tokens_saved", 0) or 0
|
||||
usd = lifetime.get("compression_savings_usd", 0.0) or 0.0
|
||||
if not tokens:
|
||||
return CheckResult(
|
||||
name=name,
|
||||
status=WARN,
|
||||
summary="no tokens saved yet",
|
||||
hint="route a client through the proxy and make a request",
|
||||
)
|
||||
|
||||
session = payload.get("display_session") or {}
|
||||
freshness = None
|
||||
last_activity = session.get("last_activity_at")
|
||||
if isinstance(last_activity, str):
|
||||
freshness = _format_since(last_activity)
|
||||
summary = f"{tokens:,} tokens / ${usd:,.2f} saved lifetime"
|
||||
if freshness:
|
||||
summary += f" — last request {freshness}"
|
||||
return CheckResult(name=name, status=PASS, summary=f"{summary} ({source})")
|
||||
|
||||
|
||||
def check_budget(stats: dict[str, Any] | None) -> CheckResult:
|
||||
"""Is a spend budget configured on the proxy?"""
|
||||
name = "budget"
|
||||
if stats is None:
|
||||
return CheckResult(name=name, status=SKIP, summary="proxy not reachable")
|
||||
cost = stats.get("cost")
|
||||
if not isinstance(cost, dict):
|
||||
return CheckResult(name=name, status=WARN, summary="cost tracking disabled (--no-cost)")
|
||||
if "budget_limit_usd" not in cost:
|
||||
return CheckResult(
|
||||
name=name,
|
||||
status=WARN,
|
||||
summary="proxy does not report budget config (older version?)",
|
||||
hint="restart the proxy on the current version",
|
||||
)
|
||||
limit = cost.get("budget_limit_usd")
|
||||
if limit is None:
|
||||
return CheckResult(
|
||||
name=name,
|
||||
status=WARN,
|
||||
summary="no budget configured — spend is unlimited",
|
||||
hint="set one: headroom proxy --budget 10 (env: HEADROOM_BUDGET)",
|
||||
)
|
||||
period = cost.get("budget_period", "daily")
|
||||
return CheckResult(name=name, status=PASS, summary=f"${limit}/{period} budget enforced")
|
||||
|
||||
|
||||
def check_deployments(manifests: list[Any], probe: Any = probe_json) -> CheckResult | None:
|
||||
"""Probe persistent deployment health URLs. None when no deployments."""
|
||||
if not manifests:
|
||||
return None
|
||||
down = []
|
||||
for manifest in manifests:
|
||||
payload = probe(manifest.health_url)
|
||||
ready = bool(payload and (payload.get("ready") or payload.get("status") == "healthy"))
|
||||
if not ready:
|
||||
down.append(manifest.profile)
|
||||
if down:
|
||||
return CheckResult(
|
||||
name="deployments",
|
||||
status=FAIL,
|
||||
summary=f"{len(down)} of {len(manifests)} deployment(s) down: {', '.join(down)}",
|
||||
hint="inspect with: headroom install status --profile <name>",
|
||||
)
|
||||
return CheckResult(
|
||||
name="deployments",
|
||||
status=PASS,
|
||||
summary=f"{len(manifests)} deployment(s) healthy",
|
||||
)
|
||||
|
||||
|
||||
_STATUS_STYLE = {PASS: "green", WARN: "yellow", FAIL: "red", SKIP: "dim"}
|
||||
_STATUS_GLYPH = {PASS: "✓", WARN: "⚠", FAIL: "✗", SKIP: "·"}
|
||||
|
||||
|
||||
def _render(checks: list[CheckResult], port: int, installed: str) -> None:
|
||||
from rich.console import Console
|
||||
from rich.markup import escape
|
||||
from rich.table import Table
|
||||
|
||||
console = Console()
|
||||
console.print(f"[bold]Headroom Doctor[/bold] [dim]v{installed} · port {port}[/dim]\n")
|
||||
table = Table(show_header=True, header_style="bold")
|
||||
table.add_column("check")
|
||||
table.add_column("status")
|
||||
table.add_column("summary")
|
||||
for check in checks:
|
||||
style = _STATUS_STYLE.get(check.status, "white")
|
||||
glyph = _STATUS_GLYPH.get(check.status, "?")
|
||||
table.add_row(
|
||||
check.name,
|
||||
f"[{style}]{glyph} {check.status}[/{style}]",
|
||||
escape(check.summary),
|
||||
)
|
||||
console.print(table)
|
||||
for check in checks:
|
||||
if check.hint:
|
||||
console.print(f"[dim]{check.name}:[/dim] {escape(check.hint)}")
|
||||
|
||||
fails = sum(1 for c in checks if c.status == FAIL)
|
||||
warns = sum(1 for c in checks if c.status == WARN)
|
||||
if fails or warns:
|
||||
console.print(f"\n[bold]{fails} failure(s), {warns} warning(s)[/bold]")
|
||||
else:
|
||||
console.print("\n[green bold]all checks passed[/green bold]")
|
||||
|
||||
|
||||
@main.command()
|
||||
@click.option(
|
||||
"--port",
|
||||
"-p",
|
||||
default=8787,
|
||||
type=int,
|
||||
envvar="HEADROOM_PORT",
|
||||
help="Proxy port to check (default: 8787, env: HEADROOM_PORT)",
|
||||
)
|
||||
@click.option("--json", "emit_json", is_flag=True, help="Emit JSON instead of formatted output.")
|
||||
def doctor(port: int, emit_json: bool) -> None:
|
||||
"""Check that the Headroom proxy and client routing are working.
|
||||
|
||||
\b
|
||||
Exit codes:
|
||||
0 everything healthy
|
||||
1 warnings only (working, but not optimally wired)
|
||||
2 at least one failure (proxy down / deployment down)
|
||||
"""
|
||||
base_url = f"http://127.0.0.1:{port}"
|
||||
livez = probe_json(f"{base_url}/livez")
|
||||
stats = probe_json(f"{base_url}/stats", timeout=5.0) if livez else None
|
||||
installed = get_version()
|
||||
|
||||
checks = [
|
||||
check_proxy_liveness(livez, base_url),
|
||||
check_version_drift(livez, installed),
|
||||
check_claude_routing(claude_settings_path(), port),
|
||||
check_codex_routing(codex_config_path(), port),
|
||||
check_shell_env(os.environ, port),
|
||||
check_savings(stats, savings_path()),
|
||||
check_budget(stats),
|
||||
]
|
||||
deployments = check_deployments(list_manifests())
|
||||
if deployments is not None:
|
||||
checks.append(deployments)
|
||||
|
||||
if any(c.status == FAIL for c in checks):
|
||||
exit_code = 2
|
||||
elif any(c.status == WARN for c in checks):
|
||||
exit_code = 1
|
||||
else:
|
||||
exit_code = 0
|
||||
|
||||
if emit_json:
|
||||
click.echo(
|
||||
json.dumps(
|
||||
{
|
||||
"port": port,
|
||||
"installed_version": installed,
|
||||
"exit_code": exit_code,
|
||||
"checks": [asdict(c) for c in checks],
|
||||
},
|
||||
indent=2,
|
||||
)
|
||||
)
|
||||
else:
|
||||
_render(checks, port, installed)
|
||||
raise SystemExit(exit_code)
|
||||
|
|
@ -40,6 +40,7 @@ def _register_commands() -> None:
|
|||
audit, # noqa: F401
|
||||
capture, # noqa: F401
|
||||
copilot_auth, # noqa: F401
|
||||
doctor, # noqa: F401
|
||||
evals, # noqa: F401
|
||||
init, # noqa: F401
|
||||
install, # noqa: F401
|
||||
|
|
|
|||
|
|
@ -853,4 +853,8 @@ class CostTracker:
|
|||
"per_model": per_model,
|
||||
"cost_with_headroom_usd": round(cost_with_headroom, 4),
|
||||
"savings_usd": round(savings_usd, 4),
|
||||
# Budget config passthrough — surfaces in /stats["cost"] so
|
||||
# `headroom doctor` can report whether a budget is set.
|
||||
"budget_limit_usd": self.budget_limit_usd,
|
||||
"budget_period": self.budget_period,
|
||||
}
|
||||
|
|
|
|||
349
tests/test_cli_doctor.py
Normal file
349
tests/test_cli_doctor.py
Normal file
|
|
@ -0,0 +1,349 @@
|
|||
"""Tests for `headroom doctor`."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from dataclasses import dataclass
|
||||
|
||||
import pytest
|
||||
from click.testing import CliRunner
|
||||
|
||||
import headroom.cli.doctor as doctor_mod
|
||||
from headroom.cli.doctor import (
|
||||
FAIL,
|
||||
PASS,
|
||||
SKIP,
|
||||
WARN,
|
||||
check_budget,
|
||||
check_claude_routing,
|
||||
check_codex_routing,
|
||||
check_deployments,
|
||||
check_proxy_liveness,
|
||||
check_savings,
|
||||
check_shell_env,
|
||||
check_version_drift,
|
||||
)
|
||||
from headroom.cli.main import main
|
||||
|
||||
LIVEZ_OK = {
|
||||
"service": "headroom-proxy",
|
||||
"status": "healthy",
|
||||
"alive": True,
|
||||
"version": "0.26.0",
|
||||
"uptime_seconds": 260135.0,
|
||||
}
|
||||
|
||||
STATS_OK = {
|
||||
"persistent_savings": {
|
||||
"lifetime": {"tokens_saved": 17_583_102, "compression_savings_usd": 7.81701},
|
||||
"display_session": {"last_activity_at": "2026-06-12T12:00:00Z"},
|
||||
},
|
||||
"cost": {"budget_limit_usd": 10.0, "budget_period": "daily"},
|
||||
}
|
||||
|
||||
|
||||
class TestProxyLiveness:
|
||||
def test_down_is_fail_with_hint(self):
|
||||
result = check_proxy_liveness(None, "http://127.0.0.1:8787")
|
||||
assert result.status == FAIL
|
||||
assert "headroom proxy" in (result.hint or "")
|
||||
|
||||
def test_up_mentions_version_and_uptime(self):
|
||||
result = check_proxy_liveness(LIVEZ_OK, "http://127.0.0.1:8787")
|
||||
assert result.status == PASS
|
||||
assert "v0.26.0" in result.summary
|
||||
assert "3d" in result.summary
|
||||
|
||||
|
||||
class TestVersionDrift:
|
||||
def test_match_passes(self):
|
||||
assert check_version_drift(LIVEZ_OK, "0.26.0").status == PASS
|
||||
|
||||
def test_mismatch_warns_with_restart_hint(self):
|
||||
result = check_version_drift(LIVEZ_OK, "0.27.0")
|
||||
assert result.status == WARN
|
||||
assert "drift" in result.summary
|
||||
assert "restart" in (result.hint or "")
|
||||
|
||||
def test_proxy_down_skips(self):
|
||||
assert check_version_drift(None, "0.26.0").status == SKIP
|
||||
|
||||
def test_unknown_version_warns(self):
|
||||
assert check_version_drift({"version": "unknown"}, "0.26.0").status == WARN
|
||||
assert check_version_drift(LIVEZ_OK, "unknown").status == WARN
|
||||
|
||||
|
||||
class TestClaudeRouting:
|
||||
def test_missing_file_warns(self, tmp_path):
|
||||
result = check_claude_routing(tmp_path / "settings.json", 8787)
|
||||
assert result.status == WARN
|
||||
assert "wrap claude" in (result.hint or "")
|
||||
|
||||
def test_malformed_json_warns(self, tmp_path):
|
||||
path = tmp_path / "settings.json"
|
||||
path.write_text("{not json", encoding="utf-8")
|
||||
assert check_claude_routing(path, 8787).status == WARN
|
||||
|
||||
def test_no_env_key_warns(self, tmp_path):
|
||||
path = tmp_path / "settings.json"
|
||||
path.write_text(json.dumps({"env": {}}), encoding="utf-8")
|
||||
assert check_claude_routing(path, 8787).status == WARN
|
||||
|
||||
def test_correct_url_passes(self, tmp_path):
|
||||
path = tmp_path / "settings.json"
|
||||
path.write_text(
|
||||
json.dumps({"env": {"ANTHROPIC_BASE_URL": "http://127.0.0.1:8787"}}),
|
||||
encoding="utf-8",
|
||||
)
|
||||
assert check_claude_routing(path, 8787).status == PASS
|
||||
|
||||
def test_port_mismatch_warns(self, tmp_path):
|
||||
path = tmp_path / "settings.json"
|
||||
path.write_text(
|
||||
json.dumps({"env": {"ANTHROPIC_BASE_URL": "http://127.0.0.1:8788"}}),
|
||||
encoding="utf-8",
|
||||
)
|
||||
result = check_claude_routing(path, 8787)
|
||||
assert result.status == WARN
|
||||
assert "8788" in result.summary
|
||||
|
||||
def test_non_headroom_url_warns(self, tmp_path):
|
||||
path = tmp_path / "settings.json"
|
||||
path.write_text(
|
||||
json.dumps({"env": {"ANTHROPIC_BASE_URL": "https://gateway.corp.example/v1"}}),
|
||||
encoding="utf-8",
|
||||
)
|
||||
result = check_claude_routing(path, 8787)
|
||||
assert result.status == WARN
|
||||
assert "gateway.corp.example" in result.summary
|
||||
|
||||
|
||||
class TestCodexRouting:
|
||||
def test_missing_file_warns(self, tmp_path):
|
||||
assert check_codex_routing(tmp_path / "config.toml", 8787).status == WARN
|
||||
|
||||
def test_marker_block_right_port_passes(self, tmp_path):
|
||||
path = tmp_path / "config.toml"
|
||||
path.write_text(
|
||||
'model_provider = "headroom"\n'
|
||||
"[model_providers.headroom]\n"
|
||||
'base_url = "http://127.0.0.1:8787/v1"\n',
|
||||
encoding="utf-8",
|
||||
)
|
||||
assert check_codex_routing(path, 8787).status == PASS
|
||||
|
||||
def test_port_mismatch_warns(self, tmp_path):
|
||||
path = tmp_path / "config.toml"
|
||||
path.write_text(
|
||||
'[model_providers.headroom]\nbase_url = "http://127.0.0.1:9999/v1"\n',
|
||||
encoding="utf-8",
|
||||
)
|
||||
result = check_codex_routing(path, 8787)
|
||||
assert result.status == WARN
|
||||
assert "9999" in result.summary
|
||||
|
||||
def test_no_marker_warns(self, tmp_path):
|
||||
path = tmp_path / "config.toml"
|
||||
path.write_text('model = "gpt-5"\n', encoding="utf-8")
|
||||
assert check_codex_routing(path, 8787).status == WARN
|
||||
|
||||
def test_garbage_bytes_warn_not_crash(self, tmp_path):
|
||||
path = tmp_path / "config.toml"
|
||||
path.write_bytes(b"\xff\xfe garbage \x00")
|
||||
assert check_codex_routing(path, 8787).status == WARN
|
||||
|
||||
|
||||
class TestShellEnv:
|
||||
def test_unset_warns(self):
|
||||
result = check_shell_env({}, 8787)
|
||||
assert result.status == WARN
|
||||
assert "bypasses" in result.summary
|
||||
|
||||
def test_matching_anthropic_url_passes(self):
|
||||
env = {"ANTHROPIC_BASE_URL": "http://127.0.0.1:8787"}
|
||||
assert check_shell_env(env, 8787).status == PASS
|
||||
|
||||
def test_localhost_also_passes(self):
|
||||
env = {"OPENAI_BASE_URL": "http://localhost:8787/v1"}
|
||||
assert check_shell_env(env, 8787).status == PASS
|
||||
|
||||
def test_other_url_warns(self):
|
||||
env = {"ANTHROPIC_BASE_URL": "https://api.anthropic.com"}
|
||||
assert check_shell_env(env, 8787).status == WARN
|
||||
|
||||
|
||||
class TestSavings:
|
||||
def test_from_stats_passes_with_totals(self, tmp_path):
|
||||
result = check_savings(STATS_OK, tmp_path / "missing.json")
|
||||
assert result.status == PASS
|
||||
assert "17,583,102" in result.summary
|
||||
assert "$7.82" in result.summary
|
||||
|
||||
def test_falls_back_to_file_when_proxy_down(self, tmp_path):
|
||||
savings_file = tmp_path / "proxy_savings.json"
|
||||
savings_file.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"lifetime": {"tokens_saved": 500, "compression_savings_usd": 0.01},
|
||||
"display_session": {"last_activity_at": "2026-06-12T11:00:00Z"},
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
result = check_savings(None, savings_file)
|
||||
assert result.status == PASS
|
||||
assert "500" in result.summary
|
||||
assert str(savings_file) in result.summary
|
||||
|
||||
def test_no_data_warns(self, tmp_path):
|
||||
assert check_savings(None, tmp_path / "missing.json").status == WARN
|
||||
|
||||
def test_zero_tokens_warns(self, tmp_path):
|
||||
stats = {"persistent_savings": {"lifetime": {"tokens_saved": 0}}}
|
||||
assert check_savings(stats, tmp_path / "missing.json").status == WARN
|
||||
|
||||
|
||||
class TestBudget:
|
||||
def test_proxy_down_skips(self):
|
||||
assert check_budget(None).status == SKIP
|
||||
|
||||
def test_cost_tracking_disabled_warns(self):
|
||||
assert check_budget({"cost": None}).status == WARN
|
||||
|
||||
def test_old_proxy_without_keys_warns(self):
|
||||
result = check_budget({"cost": {"savings_usd": 1.0}})
|
||||
assert result.status == WARN
|
||||
assert "older version" in result.summary
|
||||
|
||||
def test_unset_budget_warns_with_hint(self):
|
||||
result = check_budget({"cost": {"budget_limit_usd": None}})
|
||||
assert result.status == WARN
|
||||
assert "--budget" in (result.hint or "")
|
||||
|
||||
def test_configured_budget_passes(self):
|
||||
result = check_budget(STATS_OK)
|
||||
assert result.status == PASS
|
||||
assert "$10.0/daily" in result.summary
|
||||
|
||||
|
||||
@dataclass
|
||||
class _FakeManifest:
|
||||
profile: str
|
||||
health_url: str
|
||||
|
||||
|
||||
class TestDeployments:
|
||||
def test_no_manifests_omits_section(self):
|
||||
assert check_deployments([]) is None
|
||||
|
||||
def test_all_healthy_passes(self):
|
||||
manifests = [_FakeManifest("default", "http://127.0.0.1:8787/readyz")]
|
||||
result = check_deployments(manifests, probe=lambda url: {"ready": True})
|
||||
assert result is not None and result.status == PASS
|
||||
|
||||
def test_unhealthy_fails_naming_profile(self):
|
||||
manifests = [_FakeManifest("prod", "http://127.0.0.1:9999/readyz")]
|
||||
result = check_deployments(manifests, probe=lambda url: None)
|
||||
assert result is not None and result.status == FAIL
|
||||
assert "prod" in result.summary
|
||||
|
||||
|
||||
class TestDoctorCommand:
|
||||
@pytest.fixture
|
||||
def runner(self):
|
||||
return CliRunner()
|
||||
|
||||
@pytest.fixture
|
||||
def isolated(self, tmp_path, monkeypatch):
|
||||
"""Point all filesystem/network surfaces at controlled fakes."""
|
||||
monkeypatch.setattr(doctor_mod, "claude_settings_path", lambda: tmp_path / "settings.json")
|
||||
monkeypatch.setattr(doctor_mod, "codex_config_path", lambda: tmp_path / "config.toml")
|
||||
monkeypatch.setattr(doctor_mod, "savings_path", lambda: tmp_path / "savings.json")
|
||||
monkeypatch.setattr(doctor_mod, "list_manifests", lambda: [])
|
||||
for var in ("ANTHROPIC_BASE_URL", "OPENAI_BASE_URL", "HEADROOM_PORT"):
|
||||
monkeypatch.delenv(var, raising=False)
|
||||
return tmp_path
|
||||
|
||||
def _probe(self, livez, stats):
|
||||
def fake_probe(url, timeout=2.0):
|
||||
if url.endswith("/livez"):
|
||||
return livez
|
||||
if url.endswith("/stats"):
|
||||
return stats
|
||||
return None
|
||||
|
||||
return fake_probe
|
||||
|
||||
def test_proxy_down_exits_2(self, runner, isolated, monkeypatch):
|
||||
monkeypatch.setattr(doctor_mod, "probe_json", self._probe(None, None))
|
||||
result = runner.invoke(main, ["doctor"])
|
||||
assert result.exit_code == 2
|
||||
assert "not reachable" in result.output
|
||||
|
||||
def test_warnings_only_exits_1(self, runner, isolated, monkeypatch):
|
||||
monkeypatch.setattr(doctor_mod, "probe_json", self._probe(LIVEZ_OK, STATS_OK))
|
||||
monkeypatch.setattr(doctor_mod, "get_version", lambda: "0.26.0")
|
||||
# proxy healthy, but clients unwrapped + shell env unset -> warns
|
||||
result = runner.invoke(main, ["doctor"])
|
||||
assert result.exit_code == 1
|
||||
|
||||
def test_all_pass_exits_0(self, runner, isolated, monkeypatch):
|
||||
monkeypatch.setattr(doctor_mod, "probe_json", self._probe(LIVEZ_OK, STATS_OK))
|
||||
monkeypatch.setattr(doctor_mod, "get_version", lambda: "0.26.0")
|
||||
(isolated / "settings.json").write_text(
|
||||
json.dumps({"env": {"ANTHROPIC_BASE_URL": "http://127.0.0.1:8787"}}),
|
||||
encoding="utf-8",
|
||||
)
|
||||
(isolated / "config.toml").write_text(
|
||||
'[model_providers.headroom]\nbase_url = "http://127.0.0.1:8787/v1"\n',
|
||||
encoding="utf-8",
|
||||
)
|
||||
result = runner.invoke(
|
||||
main, ["doctor"], env={"ANTHROPIC_BASE_URL": "http://127.0.0.1:8787"}
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
assert "all checks passed" in result.output
|
||||
|
||||
def test_json_output_parses(self, runner, isolated, monkeypatch):
|
||||
monkeypatch.setattr(doctor_mod, "probe_json", self._probe(LIVEZ_OK, STATS_OK))
|
||||
result = runner.invoke(main, ["doctor", "--json"])
|
||||
payload = json.loads(result.output)
|
||||
assert payload["port"] == 8787
|
||||
assert {c["name"] for c in payload["checks"]} >= {"proxy", "version", "budget"}
|
||||
assert all(c["status"] in ("pass", "warn", "fail", "skip") for c in payload["checks"])
|
||||
|
||||
def test_port_option_changes_probe_url(self, runner, isolated, monkeypatch):
|
||||
seen: list[str] = []
|
||||
|
||||
def recording_probe(url, timeout=2.0):
|
||||
seen.append(url)
|
||||
return None
|
||||
|
||||
monkeypatch.setattr(doctor_mod, "probe_json", recording_probe)
|
||||
runner.invoke(main, ["doctor", "--port", "9999"])
|
||||
assert "http://127.0.0.1:9999/livez" in seen
|
||||
|
||||
def test_port_env_var_respected(self, runner, isolated, monkeypatch):
|
||||
seen: list[str] = []
|
||||
|
||||
def recording_probe(url, timeout=2.0):
|
||||
seen.append(url)
|
||||
return None
|
||||
|
||||
monkeypatch.setattr(doctor_mod, "probe_json", recording_probe)
|
||||
runner.invoke(main, ["doctor"], env={"HEADROOM_PORT": "9999"})
|
||||
assert "http://127.0.0.1:9999/livez" in seen
|
||||
|
||||
|
||||
class TestCostTrackerBudgetKeys:
|
||||
def test_stats_exposes_budget_config(self):
|
||||
from headroom.proxy.cost import CostTracker
|
||||
|
||||
stats = CostTracker(budget_limit_usd=5.0, budget_period="monthly").stats()
|
||||
assert stats["budget_limit_usd"] == 5.0
|
||||
assert stats["budget_period"] == "monthly"
|
||||
|
||||
def test_stats_budget_none_when_unset(self):
|
||||
from headroom.proxy.cost import CostTracker
|
||||
|
||||
assert CostTracker().stats()["budget_limit_usd"] is None
|
||||
Loading…
Add table
Add a link
Reference in a new issue