feat(doctor): add opt-in `hermes doctor --live` real-call backend probes
Adds a bounded, read-only health probe per CONFIGURED tool backend, run only when the user explicitly passes `--live` (real network calls): - Firecrawl: credit-usage metadata GET (auth check, no scrape spend) - FAL: models metadata GET (never a generation) - Browser: headless launch + about:blank + close (full cleanup) - MCP: initialize + tools/list per configured server (reuses the `hermes mcp test` machinery in mcp_config._probe_single_server) - TTS/STT: provider models/voices list GET (openai/groq/elevenlabs); local providers (edge/piper/faster-whisper/...) skipped Invariants: - Opt-in only: zero probes without --live (default False) - Bounded: sequential, per-probe timeout (doctor.live_probe_timeout, default 10s, config.yaml knob) - Never mutates state; unconfigured backends skip with a note - Failure isolation: every probe wrapped in a catch-all; a probe crash can never break the doctor run; failures append to the issues summary New: hermes_cli/doctor_live.py, tests/hermes_cli/test_doctor_live.py (23 tests, probes mocked at the HTTP/client seam). Wired: --live flag in subcommands/doctor.py; run_doctor calls maybe_run_live_checks after all static checks. Coordination: PR #70124 (--probe-routes) probes LLM routes; this flag probes TOOL backends — different surface, no code-region collision (the run_doctor hook here sits at the end-of-run summary, not the API Connectivity section #70124 extends). Inspired by: paradigmxyz/centaur tool-health-smoke (MIT/Apache-2.0); sibling: #70124 (LLM route probes — different surface)
This commit is contained in:
Teknium
parent
0ebaa490b5
commit
1006faa6f8
|
|
@ -2799,6 +2799,13 @@ DEFAULT_CONFIG = {
|
|||
},
|
||||
},
|
||||
|
||||
# ``hermes doctor`` behaviour.
|
||||
"doctor": {
|
||||
# Per-probe timeout (seconds) for the opt-in `hermes doctor --live`
|
||||
# real-call backend probes (Firecrawl/FAL/browser/MCP/TTS/STT).
|
||||
"live_probe_timeout": 10,
|
||||
},
|
||||
|
||||
# ``hermes update`` behaviour.
|
||||
"updates": {
|
||||
# Pre-update safety backup — ONE consolidated mechanism, three modes:
|
||||
|
|
|
|||
|
|
@ -2747,6 +2747,14 @@ def run_doctor(args):
|
|||
except Exception:
|
||||
pass
|
||||
|
||||
# Opt-in live backend probes run AFTER all static checks, only with
|
||||
# `hermes doctor --live` (real network calls; bounded + read-only).
|
||||
try:
|
||||
from hermes_cli.doctor_live import maybe_run_live_checks
|
||||
maybe_run_live_checks(args, manual_issues)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
print()
|
||||
remaining_issues = issues + manual_issues
|
||||
if should_fix and fixed_count > 0:
|
||||
|
|
|
|||
|
|
@ -0,0 +1,316 @@
|
|||
"""``hermes doctor --live`` — opt-in bounded real-call tool-backend probes.
|
||||
|
||||
Design invariants:
|
||||
|
||||
- **Opt-in only.** These probes make real (cheap, metadata/read-only) network
|
||||
calls and may spend a trivial amount of quota. They run ONLY when the user
|
||||
passes ``hermes doctor --live``.
|
||||
- **Bounded.** One probe per configured backend, sequential, each with a
|
||||
~10s timeout (configurable via ``doctor.live_probe_timeout`` in
|
||||
config.yaml).
|
||||
- **Read-only.** Metadata GETs only — no generation, no scrapes that spend
|
||||
credits, no state mutation anywhere.
|
||||
- **Failure-isolated.** A probe crashing must never crash the doctor run;
|
||||
every probe is wrapped in a catch-all.
|
||||
- **Configured-only.** Backends without credentials / config are skipped with
|
||||
a note, never failed.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from dataclasses import dataclass
|
||||
from typing import Callable, List, Optional
|
||||
|
||||
from hermes_cli.doctor import (
|
||||
_section,
|
||||
check_fail,
|
||||
check_info,
|
||||
check_ok,
|
||||
check_warn,
|
||||
)
|
||||
|
||||
DEFAULT_PROBE_TIMEOUT = 10.0
|
||||
|
||||
# Metadata-only endpoints. None of these spend generation credits.
|
||||
FIRECRAWL_HEALTH_URL = "https://api.firecrawl.dev/v2/team/credit-usage"
|
||||
FAL_MODELS_URL = "https://fal.ai/api/models?page=1"
|
||||
OPENAI_MODELS_URL = "https://api.openai.com/v1/models"
|
||||
GROQ_MODELS_URL = "https://api.groq.com/openai/v1/models"
|
||||
ELEVENLABS_VOICES_URL = "https://api.elevenlabs.io/v1/voices"
|
||||
|
||||
# TTS/STT providers that never touch the network (nothing to probe).
|
||||
_LOCAL_AUDIO_PROVIDERS = {"", "local", "edge", "neutts", "kittentts", "piper"}
|
||||
|
||||
|
||||
@dataclass
|
||||
class ProbeResult:
|
||||
"""Outcome of one backend probe."""
|
||||
|
||||
name: str
|
||||
status: str # "pass" | "warn" | "fail" | "skip"
|
||||
detail: str = ""
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Small seams (monkeypatchable in tests, and single points of control).
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _load_config() -> dict:
|
||||
try:
|
||||
from hermes_cli.config import load_config
|
||||
|
||||
return load_config() or {}
|
||||
except Exception:
|
||||
return {}
|
||||
|
||||
|
||||
def _http_get(url: str, headers: Optional[dict] = None,
|
||||
timeout: Optional[float] = None):
|
||||
"""Single HTTP GET seam for all metadata probes."""
|
||||
import httpx
|
||||
|
||||
return httpx.get(url, headers=headers or {}, timeout=timeout)
|
||||
|
||||
|
||||
def _browser_available() -> bool:
|
||||
"""Is the local browser automation backend (agent-browser) installed?"""
|
||||
import shutil
|
||||
|
||||
if shutil.which("agent-browser"):
|
||||
return True
|
||||
try:
|
||||
from hermes_cli.doctor import HERMES_HOME, PROJECT_ROOT
|
||||
|
||||
if (PROJECT_ROOT / "node_modules" / "agent-browser").exists():
|
||||
return True
|
||||
for candidate in (HERMES_HOME / "node" / "bin",
|
||||
HERMES_HOME / "node",
|
||||
HERMES_HOME / "node_modules" / ".bin"):
|
||||
if shutil.which("agent-browser", path=str(candidate)):
|
||||
return True
|
||||
except Exception:
|
||||
pass
|
||||
return False
|
||||
|
||||
|
||||
def _launch_browser_probe(timeout: float) -> tuple:
|
||||
"""Launch a browser, open about:blank, close. Returns (ok, detail).
|
||||
|
||||
Uses Playwright directly (what agent-browser drives underneath) so the
|
||||
probe owns the full lifecycle and always cleans up.
|
||||
"""
|
||||
try:
|
||||
from playwright.sync_api import sync_playwright
|
||||
except ImportError:
|
||||
return (False, "playwright not installed")
|
||||
|
||||
with sync_playwright() as p:
|
||||
browser = p.chromium.launch(headless=True,
|
||||
timeout=timeout * 1000)
|
||||
try:
|
||||
page = browser.new_page()
|
||||
page.goto("about:blank", timeout=timeout * 1000)
|
||||
finally:
|
||||
browser.close()
|
||||
return (True, "launched + about:blank + closed")
|
||||
|
||||
|
||||
def _probe_mcp_server(name: str, config: dict, timeout: float):
|
||||
"""initialize + tools/list against one configured MCP server.
|
||||
|
||||
Reuses the exact machinery behind ``hermes mcp test``.
|
||||
"""
|
||||
from hermes_cli.mcp_config import _probe_single_server
|
||||
|
||||
return _probe_single_server(name, config, connect_timeout=timeout)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Per-backend probes. Each returns a ProbeResult and never raises upward
|
||||
# beyond what run_live_checks' catch-all handles.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _classify_http(name: str, resp, key_hint: str) -> ProbeResult:
|
||||
code = getattr(resp, "status_code", None)
|
||||
if code is not None and 200 <= code < 300:
|
||||
return ProbeResult(name, "pass", f"(HTTP {code})")
|
||||
if code in (401, 403):
|
||||
return ProbeResult(name, "fail",
|
||||
f"(HTTP {code} — check {key_hint})")
|
||||
return ProbeResult(name, "fail", f"(HTTP {code})")
|
||||
|
||||
|
||||
def _probe_firecrawl(timeout: float) -> ProbeResult:
|
||||
key = os.getenv("FIRECRAWL_API_KEY", "").strip()
|
||||
if not key:
|
||||
return ProbeResult("Firecrawl", "skip", "(not configured)")
|
||||
resp = _http_get(FIRECRAWL_HEALTH_URL,
|
||||
headers={"Authorization": f"Bearer {key}"},
|
||||
timeout=timeout)
|
||||
return _classify_http("Firecrawl", resp, "FIRECRAWL_API_KEY")
|
||||
|
||||
|
||||
def _probe_fal(timeout: float) -> ProbeResult:
|
||||
key = os.getenv("FAL_KEY", "").strip()
|
||||
if not key:
|
||||
return ProbeResult("FAL", "skip", "(not configured)")
|
||||
# Metadata GET only — never a generation call.
|
||||
resp = _http_get(FAL_MODELS_URL,
|
||||
headers={"Authorization": f"Key {key}"},
|
||||
timeout=timeout)
|
||||
return _classify_http("FAL", resp, "FAL_KEY")
|
||||
|
||||
|
||||
def _probe_browser(timeout: float) -> ProbeResult:
|
||||
if not _browser_available():
|
||||
return ProbeResult("Browser", "skip", "(not configured)")
|
||||
ok, detail = _launch_browser_probe(timeout)
|
||||
return ProbeResult("Browser", "pass" if ok else "fail", f"({detail})")
|
||||
|
||||
|
||||
def _audio_provider_probe(kind: str, provider: str,
|
||||
timeout: float) -> ProbeResult:
|
||||
"""Shared TTS/STT metadata probe (voices/models list GET only)."""
|
||||
name = kind.upper()
|
||||
provider = (provider or "").strip().lower()
|
||||
if provider in _LOCAL_AUDIO_PROVIDERS:
|
||||
return ProbeResult(name, "skip",
|
||||
f"(provider '{provider or 'local'}' — no remote "
|
||||
"backend to probe)")
|
||||
|
||||
probes = {
|
||||
"openai": (OPENAI_MODELS_URL, "OPENAI_API_KEY", "Bearer"),
|
||||
"groq": (GROQ_MODELS_URL, "GROQ_API_KEY", "Bearer"),
|
||||
"elevenlabs": (ELEVENLABS_VOICES_URL, "ELEVENLABS_API_KEY", "xi"),
|
||||
}
|
||||
entry = probes.get(provider)
|
||||
if entry is None:
|
||||
return ProbeResult(name, "skip",
|
||||
f"(provider '{provider}' — no live probe "
|
||||
"implemented)")
|
||||
url, env_var, scheme = entry
|
||||
key = os.getenv(env_var, "").strip()
|
||||
if not key:
|
||||
return ProbeResult(name, "warn",
|
||||
f"(provider '{provider}' configured but "
|
||||
f"{env_var} is not set)")
|
||||
if scheme == "xi":
|
||||
headers = {"xi-api-key": key}
|
||||
else:
|
||||
headers = {"Authorization": f"Bearer {key}"}
|
||||
resp = _http_get(url, headers=headers, timeout=timeout)
|
||||
result = _classify_http(name, resp, env_var)
|
||||
result.detail = f"({provider}) {result.detail}"
|
||||
return result
|
||||
|
||||
|
||||
def _probe_tts(config: dict, timeout: float) -> ProbeResult:
|
||||
provider = ((config.get("tts") or {}).get("provider")) or ""
|
||||
return _audio_provider_probe("tts", provider, timeout)
|
||||
|
||||
|
||||
def _probe_stt(config: dict, timeout: float) -> ProbeResult:
|
||||
provider = ((config.get("stt") or {}).get("provider")) or ""
|
||||
return _audio_provider_probe("stt", provider, timeout)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Orchestration
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _report(result: ProbeResult, issues: List[str]) -> None:
|
||||
if result.status == "pass":
|
||||
check_ok(result.name, result.detail)
|
||||
elif result.status == "warn":
|
||||
check_warn(result.name, result.detail)
|
||||
elif result.status == "fail":
|
||||
check_fail(result.name, result.detail)
|
||||
issues.append(f"Live probe failed: {result.name} {result.detail}")
|
||||
else: # skip
|
||||
check_info(f"{result.name} {result.detail} — skipped")
|
||||
|
||||
|
||||
def _run_one(name: str, fn: Callable[[], ProbeResult],
|
||||
issues: List[str]) -> ProbeResult:
|
||||
"""Run one probe with a catch-all so a crash never kills doctor."""
|
||||
try:
|
||||
result = fn()
|
||||
except TimeoutError as exc:
|
||||
result = ProbeResult(name, "fail", f"(timed out: {exc})")
|
||||
except Exception as exc:
|
||||
msg = str(exc) or exc.__class__.__name__
|
||||
if "time" in msg.lower():
|
||||
result = ProbeResult(name, "fail", f"(timed out: {msg})")
|
||||
else:
|
||||
result = ProbeResult(name, "fail", f"({msg})")
|
||||
_report(result, issues)
|
||||
return result
|
||||
|
||||
|
||||
def run_live_checks(issues: List[str]) -> List[ProbeResult]:
|
||||
"""Run one bounded, read-only probe per configured tool backend.
|
||||
|
||||
Sequential by design (bounded, predictable output ordering). Appends a
|
||||
remediation line to ``issues`` for each failed probe. Skipped backends
|
||||
never fail and never append issues.
|
||||
"""
|
||||
config = _load_config()
|
||||
try:
|
||||
timeout = float(
|
||||
(config.get("doctor") or {}).get("live_probe_timeout",
|
||||
DEFAULT_PROBE_TIMEOUT))
|
||||
except (TypeError, ValueError):
|
||||
timeout = DEFAULT_PROBE_TIMEOUT
|
||||
timeout = max(1.0, timeout)
|
||||
|
||||
_section("Live Backend Probes (opt-in, real calls)")
|
||||
results: List[ProbeResult] = []
|
||||
|
||||
results.append(_run_one(
|
||||
"Firecrawl", lambda: _probe_firecrawl(timeout), issues))
|
||||
results.append(_run_one(
|
||||
"FAL", lambda: _probe_fal(timeout), issues))
|
||||
results.append(_run_one(
|
||||
"Browser", lambda: _probe_browser(timeout), issues))
|
||||
|
||||
servers = config.get("mcp_servers") or {}
|
||||
if isinstance(servers, dict) and servers:
|
||||
for name in sorted(servers):
|
||||
entry = servers[name]
|
||||
label = f"MCP: {name}"
|
||||
|
||||
def _probe(n=name, e=entry) -> ProbeResult:
|
||||
if not isinstance(e, dict):
|
||||
return ProbeResult(f"MCP: {n}", "skip",
|
||||
"(malformed config entry)")
|
||||
tools = _probe_mcp_server(n, e, timeout)
|
||||
return ProbeResult(f"MCP: {n}", "pass",
|
||||
f"({len(tools)} tool(s))")
|
||||
|
||||
results.append(_run_one(label, _probe, issues))
|
||||
else:
|
||||
results.append(ProbeResult("MCP", "skip", "(no servers configured)"))
|
||||
_report(results[-1], issues)
|
||||
|
||||
results.append(_run_one(
|
||||
"TTS", lambda: _probe_tts(config, timeout), issues))
|
||||
results.append(_run_one(
|
||||
"STT", lambda: _probe_stt(config, timeout), issues))
|
||||
|
||||
return results
|
||||
|
||||
|
||||
def maybe_run_live_checks(args, issues: List[str]):
|
||||
"""Entry point called from ``run_doctor`` after the static checks.
|
||||
|
||||
No-ops (returns None) unless the user explicitly passed ``--live``.
|
||||
A crash anywhere in the live subsystem must never break doctor.
|
||||
"""
|
||||
if not getattr(args, "live", False):
|
||||
return None
|
||||
try:
|
||||
return run_live_checks(issues)
|
||||
except Exception as exc: # catch-all: doctor must survive
|
||||
check_warn("Live backend probes crashed", f"({exc})")
|
||||
return None
|
||||
|
|
@ -22,6 +22,15 @@ def build_doctor_parser(subparsers, *, cmd_doctor: Callable) -> None:
|
|||
doctor_parser.add_argument(
|
||||
"--fix", action="store_true", help="Attempt to fix issues automatically"
|
||||
)
|
||||
doctor_parser.add_argument(
|
||||
"--live",
|
||||
action="store_true",
|
||||
help=(
|
||||
"Opt-in: run one bounded, read-only real-call health probe per "
|
||||
"configured tool backend (Firecrawl/FAL/browser/MCP/TTS/STT) "
|
||||
"after the static checks. Makes real network calls."
|
||||
),
|
||||
)
|
||||
doctor_parser.add_argument(
|
||||
"--ack",
|
||||
metavar="ADVISORY_ID",
|
||||
|
|
|
|||
|
|
@ -0,0 +1,270 @@
|
|||
"""Tests for ``hermes doctor --live`` — opt-in bounded real-call tool-backend probes.
|
||||
|
||||
All probes are mocked at the HTTP/client layer; no real network calls are made.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
from hermes_cli import doctor_live
|
||||
from hermes_cli.doctor_live import (
|
||||
ProbeResult,
|
||||
maybe_run_live_checks,
|
||||
run_live_checks,
|
||||
)
|
||||
|
||||
|
||||
def _args(live: bool = True) -> argparse.Namespace:
|
||||
return argparse.Namespace(live=live)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _clean_env(monkeypatch):
|
||||
"""Strip backend credentials so each test opts in explicitly."""
|
||||
for var in ("FIRECRAWL_API_KEY", "FAL_KEY", "OPENAI_API_KEY",
|
||||
"ELEVENLABS_API_KEY", "GROQ_API_KEY"):
|
||||
monkeypatch.delenv(var, raising=False)
|
||||
# Default: empty config, no MCP servers, local tts/stt.
|
||||
monkeypatch.setattr(doctor_live, "_load_config", lambda: {})
|
||||
# Default: browser not installed.
|
||||
monkeypatch.setattr(doctor_live, "_browser_available", lambda: False)
|
||||
|
||||
|
||||
class TestLiveFlagGating:
|
||||
def test_parser_has_live_flag_default_false(self):
|
||||
from hermes_cli.subcommands.doctor import build_doctor_parser
|
||||
|
||||
parser = argparse.ArgumentParser()
|
||||
sub = parser.add_subparsers(dest="command")
|
||||
build_doctor_parser(sub, cmd_doctor=lambda a: None)
|
||||
args = parser.parse_args(["doctor"])
|
||||
assert args.live is False
|
||||
args = parser.parse_args(["doctor", "--live"])
|
||||
assert args.live is True
|
||||
|
||||
def test_no_live_flag_means_zero_probes(self, monkeypatch):
|
||||
called = []
|
||||
monkeypatch.setattr(
|
||||
doctor_live, "run_live_checks",
|
||||
lambda *a, **k: called.append(True))
|
||||
result = maybe_run_live_checks(_args(live=False), [])
|
||||
assert result is None
|
||||
assert called == []
|
||||
|
||||
def test_missing_live_attr_means_zero_probes(self, monkeypatch):
|
||||
called = []
|
||||
monkeypatch.setattr(
|
||||
doctor_live, "run_live_checks",
|
||||
lambda *a, **k: called.append(True))
|
||||
assert maybe_run_live_checks(SimpleNamespace(), []) is None
|
||||
assert called == []
|
||||
|
||||
def test_live_flag_runs_checks(self, monkeypatch):
|
||||
called = []
|
||||
monkeypatch.setattr(
|
||||
doctor_live, "run_live_checks",
|
||||
lambda issues, **k: called.append(issues) or [])
|
||||
issues: list[str] = []
|
||||
maybe_run_live_checks(_args(live=True), issues)
|
||||
assert called == [issues]
|
||||
|
||||
def test_live_check_crash_never_propagates(self, monkeypatch, capsys):
|
||||
def _boom(*a, **k):
|
||||
raise RuntimeError("probe subsystem exploded")
|
||||
|
||||
monkeypatch.setattr(doctor_live, "run_live_checks", _boom)
|
||||
# Must not raise.
|
||||
maybe_run_live_checks(_args(live=True), [])
|
||||
|
||||
|
||||
class TestConfiguredOnlySelection:
|
||||
def test_all_unconfigured_all_skipped(self, capsys):
|
||||
results = run_live_checks([])
|
||||
assert results, "expected one result per backend"
|
||||
assert all(r.status == "skip" for r in results)
|
||||
# No issues appended for skips.
|
||||
|
||||
def test_unconfigured_backends_do_not_touch_network(self, monkeypatch):
|
||||
def _no_net(*a, **k):
|
||||
raise AssertionError("HTTP call made for unconfigured backend")
|
||||
|
||||
monkeypatch.setattr(doctor_live, "_http_get", _no_net)
|
||||
results = run_live_checks([])
|
||||
assert all(r.status == "skip" for r in results)
|
||||
|
||||
def test_firecrawl_probed_when_key_present(self, monkeypatch):
|
||||
monkeypatch.setenv("FIRECRAWL_API_KEY", "fc-test")
|
||||
calls = []
|
||||
|
||||
def _fake_get(url, headers=None, timeout=None):
|
||||
calls.append(url)
|
||||
return SimpleNamespace(status_code=200)
|
||||
|
||||
monkeypatch.setattr(doctor_live, "_http_get", _fake_get)
|
||||
results = {r.name: r for r in run_live_checks([])}
|
||||
assert results["Firecrawl"].status == "pass"
|
||||
assert any("firecrawl" in u for u in calls)
|
||||
|
||||
def test_firecrawl_invalid_key_fails_and_appends_issue(self, monkeypatch):
|
||||
monkeypatch.setenv("FIRECRAWL_API_KEY", "fc-bad")
|
||||
monkeypatch.setattr(
|
||||
doctor_live, "_http_get",
|
||||
lambda *a, **k: SimpleNamespace(status_code=401))
|
||||
issues: list[str] = []
|
||||
results = {r.name: r for r in run_live_checks(issues)}
|
||||
assert results["Firecrawl"].status == "fail"
|
||||
assert any("FIRECRAWL" in i or "Firecrawl" in i for i in issues)
|
||||
|
||||
def test_fal_probed_when_key_present(self, monkeypatch):
|
||||
monkeypatch.setenv("FAL_KEY", "fal-test")
|
||||
monkeypatch.setattr(
|
||||
doctor_live, "_http_get",
|
||||
lambda *a, **k: SimpleNamespace(status_code=200))
|
||||
results = {r.name: r for r in run_live_checks([])}
|
||||
assert results["FAL"].status == "pass"
|
||||
|
||||
def test_mcp_servers_probed_per_configured_server(self, monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
doctor_live, "_load_config",
|
||||
lambda: {"mcp_servers": {"alpha": {"url": "https://x"},
|
||||
"beta": {"command": "foo"}}})
|
||||
probed = []
|
||||
monkeypatch.setattr(
|
||||
doctor_live, "_probe_mcp_server",
|
||||
lambda name, cfg, timeout: probed.append(name) or [("t", "d")])
|
||||
results = [r for r in run_live_checks([]) if r.name.startswith("MCP")]
|
||||
assert sorted(probed) == ["alpha", "beta"]
|
||||
assert len(results) == 2
|
||||
assert all(r.status == "pass" for r in results)
|
||||
|
||||
def test_tts_local_provider_skipped(self, monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
doctor_live, "_load_config",
|
||||
lambda: {"tts": {"provider": "edge"}})
|
||||
results = {r.name: r for r in run_live_checks([])}
|
||||
assert results["TTS"].status == "skip"
|
||||
|
||||
def test_tts_openai_probed_with_key(self, monkeypatch):
|
||||
monkeypatch.setenv("OPENAI_API_KEY", "sk-test")
|
||||
monkeypatch.setattr(
|
||||
doctor_live, "_load_config",
|
||||
lambda: {"tts": {"provider": "openai"}})
|
||||
monkeypatch.setattr(
|
||||
doctor_live, "_http_get",
|
||||
lambda *a, **k: SimpleNamespace(status_code=200))
|
||||
results = {r.name: r for r in run_live_checks([])}
|
||||
assert results["TTS"].status == "pass"
|
||||
|
||||
def test_stt_groq_probed_with_key(self, monkeypatch):
|
||||
monkeypatch.setenv("GROQ_API_KEY", "gsk-test")
|
||||
monkeypatch.setattr(
|
||||
doctor_live, "_load_config",
|
||||
lambda: {"stt": {"provider": "groq"}})
|
||||
monkeypatch.setattr(
|
||||
doctor_live, "_http_get",
|
||||
lambda *a, **k: SimpleNamespace(status_code=200))
|
||||
results = {r.name: r for r in run_live_checks([])}
|
||||
assert results["STT"].status == "pass"
|
||||
|
||||
def test_stt_provider_configured_but_key_missing_warns(self, monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
doctor_live, "_load_config",
|
||||
lambda: {"stt": {"provider": "groq"}})
|
||||
results = {r.name: r for r in run_live_checks([])}
|
||||
assert results["STT"].status == "warn"
|
||||
|
||||
def test_browser_probed_when_available(self, monkeypatch):
|
||||
monkeypatch.setattr(doctor_live, "_browser_available", lambda: True)
|
||||
monkeypatch.setattr(
|
||||
doctor_live, "_launch_browser_probe",
|
||||
lambda timeout: (True, "about:blank ok"))
|
||||
results = {r.name: r for r in run_live_checks([])}
|
||||
assert results["Browser"].status == "pass"
|
||||
|
||||
|
||||
class TestFailureIsolation:
|
||||
def test_one_probe_raising_does_not_stop_others(self, monkeypatch):
|
||||
monkeypatch.setenv("FIRECRAWL_API_KEY", "fc-test")
|
||||
monkeypatch.setenv("FAL_KEY", "fal-test")
|
||||
|
||||
def _get(url, headers=None, timeout=None):
|
||||
if "firecrawl" in url:
|
||||
raise RuntimeError("connection reset")
|
||||
return SimpleNamespace(status_code=200)
|
||||
|
||||
monkeypatch.setattr(doctor_live, "_http_get", _get)
|
||||
issues: list[str] = []
|
||||
results = {r.name: r for r in run_live_checks(issues)}
|
||||
assert results["Firecrawl"].status == "fail"
|
||||
assert results["FAL"].status == "pass"
|
||||
|
||||
def test_mcp_probe_failure_isolated_per_server(self, monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
doctor_live, "_load_config",
|
||||
lambda: {"mcp_servers": {"bad": {"url": "https://x"},
|
||||
"good": {"url": "https://y"}}})
|
||||
|
||||
def _probe(name, cfg, timeout):
|
||||
if name == "bad":
|
||||
raise ConnectionError("refused")
|
||||
return [("tool", "desc")]
|
||||
|
||||
monkeypatch.setattr(doctor_live, "_probe_mcp_server", _probe)
|
||||
results = {r.name: r for r in run_live_checks([])}
|
||||
assert results["MCP: bad"].status == "fail"
|
||||
assert results["MCP: good"].status == "pass"
|
||||
|
||||
|
||||
class TestTimeoutHandling:
|
||||
def test_timeout_reported_as_fail(self, monkeypatch):
|
||||
monkeypatch.setenv("FIRECRAWL_API_KEY", "fc-test")
|
||||
|
||||
def _slow(*a, **k):
|
||||
raise TimeoutError("timed out")
|
||||
|
||||
monkeypatch.setattr(doctor_live, "_http_get", _slow)
|
||||
results = {r.name: r for r in run_live_checks([])}
|
||||
assert results["Firecrawl"].status == "fail"
|
||||
assert "time" in (results["Firecrawl"].detail or "").lower()
|
||||
|
||||
def test_probe_timeout_bounded_and_configurable(self, monkeypatch):
|
||||
monkeypatch.setenv("FIRECRAWL_API_KEY", "fc-test")
|
||||
monkeypatch.setattr(
|
||||
doctor_live, "_load_config",
|
||||
lambda: {"doctor": {"live_probe_timeout": 3}})
|
||||
seen = {}
|
||||
|
||||
def _get(url, headers=None, timeout=None):
|
||||
seen["timeout"] = timeout
|
||||
return SimpleNamespace(status_code=200)
|
||||
|
||||
monkeypatch.setattr(doctor_live, "_http_get", _get)
|
||||
run_live_checks([])
|
||||
assert seen["timeout"] == 3
|
||||
|
||||
def test_default_timeout_is_10s(self, monkeypatch):
|
||||
monkeypatch.setenv("FIRECRAWL_API_KEY", "fc-test")
|
||||
seen = {}
|
||||
|
||||
def _get(url, headers=None, timeout=None):
|
||||
seen["timeout"] = timeout
|
||||
return SimpleNamespace(status_code=200)
|
||||
|
||||
monkeypatch.setattr(doctor_live, "_http_get", _get)
|
||||
run_live_checks([])
|
||||
assert seen["timeout"] == 10.0
|
||||
|
||||
|
||||
class TestReadOnly:
|
||||
def test_probe_result_is_plain_record(self):
|
||||
r = ProbeResult(name="X", status="skip", detail="not configured")
|
||||
assert (r.name, r.status, r.detail) == ("X", "skip", "not configured")
|
||||
|
||||
def test_skips_never_append_issues(self, capsys):
|
||||
issues: list[str] = []
|
||||
run_live_checks(issues)
|
||||
assert issues == []
|
||||
Loading…
Reference in New Issue