diff --git a/README.md b/README.md index 64096c1d..c8b515a5 100644 --- a/README.md +++ b/README.md @@ -225,6 +225,7 @@ pick the new config up on their next ucode run. | Command | Description | |---------|-------------| | `ucode status` | Show current workspace, base URLs, managed config files, and selected models | +| `ucode doctor` | Diagnose the local setup (uv, npm, Databricks CLI, workspace, credentials, agent CLIs, tracing) and offer to fix any problems found | | `ucode usage` | Show AI Gateway usage summary, plus your budget spend against its alert threshold when the workspace reports one | | `ucode usage --warehouse-id ` | Query a specific SQL warehouse instead of discovering one | | `ucode revert` | Clear saved state and restore backed-up config files | diff --git a/src/ucode/agents/__init__.py b/src/ucode/agents/__init__.py index 3afd4fea..e038c777 100644 --- a/src/ucode/agents/__init__.py +++ b/src/ucode/agents/__init__.py @@ -248,6 +248,38 @@ def ensure_tool_binary_available(tool: str) -> None: ) +def tool_binary_installed(tool: str) -> bool: + """True when the agent's CLI binary is on PATH. Read-only — for ``ucode doctor``.""" + return bool(shutil.which(TOOL_SPECS[tool]["binary"])) + + +def tool_update_available(tool: str) -> tuple[str, str] | None: + """Return ``(current, latest)`` when a newer agent CLI is published, else None. + Read-only wrapper over the per-agent update check — for ``ucode doctor``.""" + return _MODULES[tool].is_update_available() + + +def update_tool_binary(tool: str) -> bool: + """Install the latest agent CLI, returning True on success. Public entry + point over the internal updater so ``ucode doctor`` can apply the fix.""" + return _update_installed_tool_binary(tool) + + +def tracing_mlflow_ok() -> bool: + """True when the `mlflow` CLI that Claude tracing needs is installed and in + the supported version range. Read-only — for ``ucode doctor``.""" + current = claude._installed_mlflow_version() + return bool( + current and claude.MINIMUM_MLFLOW_VERSION <= current < claude.MAXIMUM_MLFLOW_VERSION + ) + + +def ensure_tracing_mlflow_cli() -> bool: + """Install/repair the pinned `mlflow` CLI for Claude tracing, returning True + on success. Public entry point so ``ucode doctor`` can apply the fix.""" + return claude._ensure_mlflow_cli() + + def ensure_bootstrap_dependencies( tool: str, *, diff --git a/src/ucode/cli.py b/src/ucode/cli.py index a68c90d1..ad652273 100644 --- a/src/ucode/cli.py +++ b/src/ucode/cli.py @@ -2387,6 +2387,18 @@ def revert_cmd() -> None: raise typer.Exit(1) from None +@app.command("doctor") +def doctor_cmd() -> None: + """Diagnose the local ucode setup and offer to fix any problems found.""" + from ucode.doctor import doctor + + try: + doctor() + except RuntimeError as exc: + print_err(str(exc)) + raise typer.Exit(1) from None + + @app.command("usage") def usage_cmd( warehouse_id: Annotated[ diff --git a/src/ucode/databricks.py b/src/ucode/databricks.py index ec79b989..a8e457a5 100644 --- a/src/ucode/databricks.py +++ b/src/ucode/databricks.py @@ -687,6 +687,38 @@ def ensure_databricks_cli_version() -> None: ensure_databricks_cli_version() +def databricks_cli_version() -> tuple[int, int, int] | None: + """Return the installed Databricks CLI's (major, minor, patch), or None if + the CLI is absent or its version can't be read/parsed. Unlike + ``ensure_databricks_cli_version`` this only reports — it never upgrades — so + ``ucode doctor`` can decide what to recommend.""" + if not shutil.which("databricks"): + return None + try: + result = run( + ["databricks", "--version"], + check=False, + capture_output=True, + text=True, + timeout=10, + ) + except (OSError, subprocess.TimeoutExpired): + return None + raw = result.stdout or result.stderr or "" + output = (raw if isinstance(raw, str) else raw.decode(errors="replace")).strip() + return _parse_databricks_cli_version(output) + + +def upgrade_databricks_cli() -> bool: + """Upgrade an already-installed Databricks CLI to the latest release. + Returns True on success, False if the installer failed.""" + try: + _run_databricks_cli_installer(brew_subcommand="upgrade") + except RuntimeError: + return False + return True + + def install_databricks_cli() -> None: if shutil.which("databricks"): ensure_databricks_cli_version() diff --git a/src/ucode/doctor.py b/src/ucode/doctor.py new file mode 100644 index 00000000..31d8bc39 --- /dev/null +++ b/src/ucode/doctor.py @@ -0,0 +1,327 @@ +"""`ucode doctor` — diagnose the local ucode setup and offer to fix what it can. + +Mirrors the `brew doctor` / `flutter doctor` / `npm doctor` pattern: run a +series of independent checks, print a status line for each, and for any problem +ucode knows how to fix, prompt the user to apply the fix and report whether it +worked. The command is read-only until the user says yes to a specific +suggestion, and a declined or piped run (no tty) changes nothing. +""" + +from __future__ import annotations + +import os +import shutil +import subprocess +from collections.abc import Callable +from dataclasses import dataclass + +from ucode.agents import ( + TOOL_SPECS, + ensure_tracing_mlflow_cli, + tool_binary_installed, + tool_update_available, + tracing_mlflow_ok, + update_tool_binary, +) +from ucode.databricks import ( + MIN_DATABRICKS_CLI_VERSION, + databricks_cli_version, + has_valid_databricks_auth, + install_databricks_cli, + run_databricks_login, + upgrade_databricks_cli, +) +from ucode.state import load_state +from ucode.telemetry import ucode_version +from ucode.tracing import tracing_config +from ucode.ui import ( + console, + heading, + label, + print_note, + print_success, + print_warning, + prompt_yes_no_default, + spinner, + status_badge, +) + +# Env vars that shadow the credential ucode configures for Claude Code. Claude +# warns when its own token and one of these are both set, so we surface them. +_CLAUDE_TOKEN_ENV_VARS = ("ANTHROPIC_AUTH_TOKEN", "ANTHROPIC_API_KEY") + +UCODE_GIT_URL = "git+https://github.com/databricks/ucode" + +# status -> (glyph, status_badge kind). "info" is a healthy line that still +# carries an optional suggestion (e.g. the ucode self-upgrade). +_BADGES = { + "ok": ("✓", "ok"), + "warn": ("!", "warn"), + "error": ("✗", "error"), + "info": ("•", "info"), +} + + +@dataclass +class Suggestion: + """A fix ucode offers to apply. ``apply`` returns True on success.""" + + prompt: str + apply: Callable[[], bool] + + +@dataclass +class Check: + name: str + status: str # one of _BADGES + detail: str + suggestion: Suggestion | None = None + + +def _fmt_version(version: tuple[int, int, int]) -> str: + return ".".join(str(n) for n in version) + + +# ── individual checks ────────────────────────────────────────────────────── + + +def _check_uv() -> Check: + if shutil.which("uv"): + return Check("uv", "ok", "found on PATH") + return Check( + "uv", + "error", + "not found — needed to install and upgrade ucode. Install it from " + "https://docs.astral.sh/uv/getting-started/installation/.", + ) + + +def _check_npm() -> Check: + if shutil.which("npm"): + return Check("npm", "ok", "found on PATH") + return Check( + "npm", + "warn", + "not found — needed to install coding-agent CLIs automatically. " + "Install Node.js/npm from https://nodejs.org/.", + ) + + +def _install_databricks() -> bool: + try: + install_databricks_cli() + except RuntimeError: + return False + return shutil.which("databricks") is not None + + +def _check_databricks_cli() -> Check: + if not shutil.which("databricks"): + return Check( + "Databricks CLI", + "error", + "not installed", + Suggestion("Install the Databricks CLI?", _install_databricks), + ) + version = databricks_cli_version() + if version is None: + return Check("Databricks CLI", "warn", "installed, but its version could not be read") + current = _fmt_version(version) + if version < MIN_DATABRICKS_CLI_VERSION: + floor = _fmt_version(MIN_DATABRICKS_CLI_VERSION) + return Check( + "Databricks CLI", + "warn", + f"v{current} is below v{floor}, the release that ships `databricks aitools`", + Suggestion("Upgrade the Databricks CLI to the latest release?", upgrade_databricks_cli), + ) + return Check("Databricks CLI", "ok", f"v{current}") + + +def _check_workspace() -> Check: + workspace = load_state().get("workspace") + if workspace: + return Check("Workspace", "ok", str(workspace)) + return Check( + "Workspace", + "warn", + "not configured — run `ucode configure` to set your Databricks workspace", + ) + + +def _check_agent_clis() -> list[Check]: + """One check per configured coding agent: installed and up to date?""" + tools = load_state().get("available_tools") or [] + checks: list[Check] = [] + for tool in tools: + if tool not in TOOL_SPECS: + continue + spec = TOOL_SPECS[tool] + display = spec["display"] + if not tool_binary_installed(tool): + checks.append( + Check( + display, + "warn", + f"`{spec['binary']}` not found on PATH", + Suggestion(f"Install {display}?", lambda t=tool: update_tool_binary(t)), + ) + ) + continue + with spinner(f"Checking {display} for updates..."): + update = tool_update_available(tool) + if update: + current, latest = update + checks.append( + Check( + display, + "warn", + f"{current} installed; {latest} available", + Suggestion( + f"Update {display} to {latest}?", lambda t=tool: update_tool_binary(t) + ), + ) + ) + else: + checks.append(Check(display, "ok", "installed and up to date")) + return checks + + +def _check_databricks_auth() -> Check | None: + """Validate the configured workspace's Databricks credentials. + + The most common ucode failure at launch is an expired/invalid token + surfacing as a `403 Invalid Token` from the agent. Checking auth up front + (and offering to re-run `databricks auth login`) catches it before launch. + Returns None when there's no workspace yet — `_check_workspace` covers that. + """ + state = load_state() + workspace = state.get("workspace") + if not workspace: + return None + profile = state.get("profile") + with spinner("Verifying Databricks credentials..."): + ok = has_valid_databricks_auth(workspace, profile) + if ok: + return Check("Databricks auth", "ok", "credentials are valid") + + def _login() -> bool: + try: + run_databricks_login(workspace, profile) + except RuntimeError: + return False + return has_valid_databricks_auth(workspace, profile) + + return Check( + "Databricks auth", + "warn", + "no valid credentials for this workspace (launches will fail to authenticate)", + Suggestion("Log in to Databricks now?", _login), + ) + + +def _check_anthropic_env_collision() -> Check | None: + """Warn when a Claude token env var is set that collides with ucode's own. + + ucode authenticates Claude Code through the gateway; a stray + `ANTHROPIC_AUTH_TOKEN`/`ANTHROPIC_API_KEY` in the environment shadows that + and makes Claude complain. We can't unset a parent shell's env, so this is + advisory (no fix). Returns None when nothing collides. + """ + set_vars = [name for name in _CLAUDE_TOKEN_ENV_VARS if os.environ.get(name, "").strip()] + if not set_vars: + return None + joined = ", ".join(set_vars) + return Check( + "Claude auth env", + "warn", + f"{joined} is set and can collide with ucode's Claude auth; unset it in your shell", + ) + + +def _check_tracing_mlflow() -> Check | None: + """When tracing is enabled, check the `mlflow` CLI it needs is installed. + + Only relevant if the user turned on tracing (`ucode configure tracing`); + otherwise there's nothing to check. A missing/out-of-range mlflow is offered + as an install. Returns None when tracing is disabled. + """ + if tracing_config(load_state()) is None: + return None + if tracing_mlflow_ok(): + return Check("Tracing (mlflow CLI)", "ok", "installed and in the supported range") + return Check( + "Tracing (mlflow CLI)", + "warn", + "tracing is enabled but the required `mlflow` CLI is missing or out of range", + Suggestion("Install the mlflow CLI for tracing?", ensure_tracing_mlflow_cli), + ) + + +def _upgrade_ucode() -> bool: + if not shutil.which("uv"): + print_warning("`uv` is not on PATH; cannot upgrade ucode.") + return False + try: + subprocess.run(["uv", "tool", "install", "--reinstall", UCODE_GIT_URL], check=True) + except (FileNotFoundError, subprocess.CalledProcessError): + return False + return True + + +def _check_ucode() -> Check: + """ucode installs from GitHub (no release tags), so there's no version to + diff against. Report the installed build and offer a reinstall-to-latest as + an optional maintenance action rather than claiming it's out of date.""" + version = ucode_version() + suggestion = ( + Suggestion("Reinstall ucode from GitHub to pick up the latest changes?", _upgrade_ucode) + if shutil.which("uv") + else None + ) + return Check("ucode", "info", f"v{version} (installed from GitHub)", suggestion) + + +# ── orchestration ────────────────────────────────────────────────────────── + + +def _gather_checks() -> list[Check]: + checks: list[Check] = [_check_uv(), _check_npm(), _check_databricks_cli(), _check_workspace()] + # These return None when they don't apply (no workspace, no env collision, + # tracing disabled), so drop the Nones before display. + optional = [_check_databricks_auth(), _check_anthropic_env_collision(), _check_tracing_mlflow()] + checks.extend(c for c in optional if c is not None) + checks.extend(_check_agent_clis()) + checks.append(_check_ucode()) + return checks + + +def doctor() -> int: + """Run every check, print its status, and prompt to apply any offered fix.""" + console.print(heading("ucode doctor")) + console.print() + + checks = _gather_checks() + problems = 0 + applied = 0 + for check in checks: + glyph, kind = _BADGES[check.status] + console.print(f" {status_badge(glyph, kind)} {label(check.name)}: {check.detail}") + if check.status in ("warn", "error"): + problems += 1 + if check.suggestion is None: + continue + if prompt_yes_no_default(f" {check.suggestion.prompt}", default=False): + if check.suggestion.apply(): + print_success(f"{check.name}: fixed") + applied += 1 + else: + print_warning(f"{check.name}: fix did not complete") + + console.print() + if problems == 0: + print_success("No problems detected.") + else: + noun = "issue" if problems == 1 else "issues" + print_note(f"{problems} {noun} found; {applied} fix(es) applied.") + return 0 diff --git a/tests/test_cli.py b/tests/test_cli.py index c6d36080..6d281040 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -715,6 +715,20 @@ def test_reverts_mcp_configs_before_clearing_state(self): assert "Claude Code MCP config: restored" in result.output +class TestDoctorCommand: + def test_invokes_doctor(self): + with patch("ucode.doctor.doctor", return_value=0) as mock_doctor: + result = runner.invoke(app, ["doctor"]) + assert result.exit_code == 0, result.output + mock_doctor.assert_called_once_with() + + def test_reports_runtime_error(self): + with patch("ucode.doctor.doctor", side_effect=RuntimeError("boom")): + result = runner.invoke(app, ["doctor"]) + assert result.exit_code == 1 + assert "boom" in _strip_ansi(result.output) + + class TestAutoConfigureOnFirstRun: def test_triggers_when_no_workspace(self): """Auto-configure runs when state has no workspace.""" diff --git a/tests/test_databricks.py b/tests/test_databricks.py index 155dcae3..cb218b55 100644 --- a/tests/test_databricks.py +++ b/tests/test_databricks.py @@ -27,6 +27,7 @@ build_skills_mcp_url, build_tool_base_url, classify_model_family, + databricks_cli_version, discover_sql_warehouses, ensure_databricks_cli_version, ensure_pat_bearer, @@ -36,6 +37,7 @@ list_databricks_connections, list_genie_spaces, resolve_current_budget_spend, + upgrade_databricks_cli, workspace_hostname, ) @@ -1872,6 +1874,50 @@ def test_raises_when_version_unparseable(self, tmp_path, monkeypatch): ensure_databricks_cli_version() +class TestDatabricksCliVersion: + def test_none_when_absent(self, monkeypatch): + monkeypatch.setattr(db_mod.shutil, "which", lambda cmd: None) + assert databricks_cli_version() is None + + def test_parses_installed_version(self, monkeypatch): + monkeypatch.setattr(db_mod.shutil, "which", lambda cmd: "/usr/bin/databricks") + monkeypatch.setattr( + db_mod, + "run", + lambda *a, **kw: subprocess.CompletedProcess(a, 0, "Databricks CLI v0.299.2", ""), + ) + assert databricks_cli_version() == (0, 299, 2) + + def test_none_on_unparseable_output(self, monkeypatch): + monkeypatch.setattr(db_mod.shutil, "which", lambda cmd: "/usr/bin/databricks") + monkeypatch.setattr( + db_mod, "run", lambda *a, **kw: subprocess.CompletedProcess(a, 0, "garbage", "") + ) + assert databricks_cli_version() is None + + def test_never_raises_on_subprocess_error(self, monkeypatch): + monkeypatch.setattr(db_mod.shutil, "which", lambda cmd: "/usr/bin/databricks") + + def boom(*a, **kw): + raise OSError("nope") + + monkeypatch.setattr(db_mod, "run", boom) + assert databricks_cli_version() is None + + +class TestUpgradeDatabricksCli: + def test_true_on_success(self, monkeypatch): + monkeypatch.setattr(db_mod, "_run_databricks_cli_installer", lambda **kw: None) + assert upgrade_databricks_cli() is True + + def test_false_when_installer_fails(self, monkeypatch): + def boom(**kw): + raise RuntimeError("install failed") + + monkeypatch.setattr(db_mod, "_run_databricks_cli_installer", boom) + assert upgrade_databricks_cli() is False + + class TestRunDatabricksCliInstaller: @pytest.mark.parametrize("brew_subcommand", ["install", "upgrade"]) def test_macos_uses_fully_qualified_tap_formula(self, monkeypatch, brew_subcommand): diff --git a/tests/test_doctor.py b/tests/test_doctor.py new file mode 100644 index 00000000..9aea20ce --- /dev/null +++ b/tests/test_doctor.py @@ -0,0 +1,281 @@ +"""Tests for `ucode doctor` — check classification and the fix-apply flow.""" + +from __future__ import annotations + +from unittest.mock import patch + +import ucode.doctor as doctor_mod +from ucode.databricks import MIN_DATABRICKS_CLI_VERSION +from ucode.doctor import ( + Check, + Suggestion, + _check_agent_clis, + _check_anthropic_env_collision, + _check_databricks_auth, + _check_databricks_cli, + _check_npm, + _check_tracing_mlflow, + _check_uv, + _check_workspace, + doctor, +) + + +class TestUvCheck: + def test_ok_when_present(self): + with patch.object(doctor_mod.shutil, "which", return_value="/usr/bin/uv"): + check = _check_uv() + assert check.status == "ok" + assert check.suggestion is None + + def test_error_when_missing(self): + with patch.object(doctor_mod.shutil, "which", return_value=None): + check = _check_uv() + assert check.status == "error" + + +class TestNpmCheck: + def test_ok_when_present(self): + with patch.object(doctor_mod.shutil, "which", return_value="/usr/bin/npm"): + assert _check_npm().status == "ok" + + def test_warn_when_missing(self): + with patch.object(doctor_mod.shutil, "which", return_value=None): + assert _check_npm().status == "warn" + + +class TestDatabricksCliCheck: + def test_error_and_install_suggestion_when_missing(self): + with patch.object(doctor_mod.shutil, "which", return_value=None): + check = _check_databricks_cli() + assert check.status == "error" + assert check.suggestion is not None + assert "Install" in check.suggestion.prompt + + def test_warn_and_upgrade_suggestion_when_below_floor(self): + # A public-preview build below the aitools floor: recommend an upgrade, + # not a hard failure. + old = (MIN_DATABRICKS_CLI_VERSION[0], MIN_DATABRICKS_CLI_VERSION[1] - 1, 0) + with ( + patch.object(doctor_mod.shutil, "which", return_value="/usr/bin/databricks"), + patch.object(doctor_mod, "databricks_cli_version", return_value=old), + ): + check = _check_databricks_cli() + assert check.status == "warn" + assert check.suggestion is not None + assert "Upgrade" in check.suggestion.prompt + + def test_ok_when_at_or_above_floor(self): + with ( + patch.object(doctor_mod.shutil, "which", return_value="/usr/bin/databricks"), + patch.object( + doctor_mod, "databricks_cli_version", return_value=MIN_DATABRICKS_CLI_VERSION + ), + ): + check = _check_databricks_cli() + assert check.status == "ok" + assert check.suggestion is None + + def test_warn_when_version_unreadable(self): + with ( + patch.object(doctor_mod.shutil, "which", return_value="/usr/bin/databricks"), + patch.object(doctor_mod, "databricks_cli_version", return_value=None), + ): + check = _check_databricks_cli() + assert check.status == "warn" + assert check.suggestion is None + + +class TestWorkspaceCheck: + def test_ok_when_configured(self): + with patch.object(doctor_mod, "load_state", return_value={"workspace": "https://ws"}): + check = _check_workspace() + assert check.status == "ok" + assert "https://ws" in check.detail + + def test_warn_when_unconfigured(self): + with patch.object(doctor_mod, "load_state", return_value={}): + assert _check_workspace().status == "warn" + + +class TestAgentCliChecks: + def test_missing_binary_offers_install(self): + state = {"available_tools": ["claude"]} + with ( + patch.object(doctor_mod, "load_state", return_value=state), + patch.object(doctor_mod, "tool_binary_installed", return_value=False), + ): + checks = _check_agent_clis() + assert len(checks) == 1 + assert checks[0].status == "warn" + assert checks[0].suggestion is not None + + def test_outdated_offers_update(self): + state = {"available_tools": ["claude"]} + with ( + patch.object(doctor_mod, "load_state", return_value=state), + patch.object(doctor_mod, "tool_binary_installed", return_value=True), + patch.object(doctor_mod, "tool_update_available", return_value=("1.0.0", "1.2.0")), + ): + checks = _check_agent_clis() + assert checks[0].status == "warn" + assert "1.2.0" in checks[0].detail + assert checks[0].suggestion is not None + + def test_up_to_date_is_ok(self): + state = {"available_tools": ["claude"]} + with ( + patch.object(doctor_mod, "load_state", return_value=state), + patch.object(doctor_mod, "tool_binary_installed", return_value=True), + patch.object(doctor_mod, "tool_update_available", return_value=None), + ): + checks = _check_agent_clis() + assert checks[0].status == "ok" + assert checks[0].suggestion is None + + def test_unknown_tool_is_skipped(self): + state = {"available_tools": ["not-a-real-tool"]} + with patch.object(doctor_mod, "load_state", return_value=state): + assert _check_agent_clis() == [] + + +class TestDatabricksAuthCheck: + def test_none_when_no_workspace(self): + with patch.object(doctor_mod, "load_state", return_value={}): + assert _check_databricks_auth() is None + + def test_ok_when_valid(self): + with ( + patch.object(doctor_mod, "load_state", return_value={"workspace": "https://ws"}), + patch.object(doctor_mod, "has_valid_databricks_auth", return_value=True), + ): + check = _check_databricks_auth() + assert check.status == "ok" + assert check.suggestion is None + + def test_warn_and_login_suggestion_when_invalid(self): + with ( + patch.object(doctor_mod, "load_state", return_value={"workspace": "https://ws"}), + patch.object(doctor_mod, "has_valid_databricks_auth", return_value=False), + ): + check = _check_databricks_auth() + assert check.status == "warn" + assert check.suggestion is not None + assert "Log in" in check.suggestion.prompt + + def test_login_fix_reports_success(self): + with ( + patch.object(doctor_mod, "load_state", return_value={"workspace": "https://ws"}), + # invalid at first, then valid after login + patch.object(doctor_mod, "has_valid_databricks_auth", side_effect=[False, True]), + patch.object(doctor_mod, "run_databricks_login") as login, + ): + check = _check_databricks_auth() + assert check.suggestion.apply() is True + login.assert_called_once() + + def test_login_fix_reports_failure_when_login_raises(self): + with ( + patch.object(doctor_mod, "load_state", return_value={"workspace": "https://ws"}), + patch.object(doctor_mod, "has_valid_databricks_auth", return_value=False), + patch.object(doctor_mod, "run_databricks_login", side_effect=RuntimeError("nope")), + ): + check = _check_databricks_auth() + assert check.suggestion.apply() is False + + +class TestAnthropicEnvCollision: + def test_none_when_unset(self, monkeypatch): + for var in doctor_mod._CLAUDE_TOKEN_ENV_VARS: + monkeypatch.delenv(var, raising=False) + assert _check_anthropic_env_collision() is None + + def test_warns_when_set(self, monkeypatch): + for var in doctor_mod._CLAUDE_TOKEN_ENV_VARS: + monkeypatch.delenv(var, raising=False) + monkeypatch.setenv("ANTHROPIC_AUTH_TOKEN", "secret") + check = _check_anthropic_env_collision() + assert check.status == "warn" + assert "ANTHROPIC_AUTH_TOKEN" in check.detail + # Advisory only — no auto-fix for a parent shell's env. + assert check.suggestion is None + + def test_blank_value_is_ignored(self, monkeypatch): + for var in doctor_mod._CLAUDE_TOKEN_ENV_VARS: + monkeypatch.delenv(var, raising=False) + monkeypatch.setenv("ANTHROPIC_API_KEY", " ") + assert _check_anthropic_env_collision() is None + + +class TestTracingMlflowCheck: + def test_none_when_tracing_disabled(self): + with patch.object(doctor_mod, "tracing_config", return_value=None): + assert _check_tracing_mlflow() is None + + def test_ok_when_mlflow_present(self): + with ( + patch.object(doctor_mod, "tracing_config", return_value={"enabled": True}), + patch.object(doctor_mod, "tracing_mlflow_ok", return_value=True), + ): + check = _check_tracing_mlflow() + assert check.status == "ok" + assert check.suggestion is None + + def test_warn_and_install_suggestion_when_missing(self): + with ( + patch.object(doctor_mod, "tracing_config", return_value={"enabled": True}), + patch.object(doctor_mod, "tracing_mlflow_ok", return_value=False), + ): + check = _check_tracing_mlflow() + assert check.status == "warn" + assert check.suggestion is not None + + +class TestDoctorFlow: + def _only(self, checks: list[Check]): + """Run doctor() with a fixed set of checks and a stubbed prompter.""" + return patch.object(doctor_mod, "_gather_checks", return_value=checks) + + def test_applies_fix_when_user_accepts(self): + applied = [] + suggestion = Suggestion("Fix it?", lambda: applied.append(True) or True) + check = Check("thing", "warn", "broken", suggestion) + with ( + self._only([check]), + patch.object(doctor_mod, "prompt_yes_no_default", return_value=True), + ): + rc = doctor() + assert rc == 0 + assert applied == [True] + + def test_skips_fix_when_user_declines(self): + applied = [] + suggestion = Suggestion("Fix it?", lambda: applied.append(True) or True) + check = Check("thing", "warn", "broken", suggestion) + with ( + self._only([check]), + patch.object(doctor_mod, "prompt_yes_no_default", return_value=False), + ): + doctor() + assert applied == [] + + def test_reports_fix_failure_without_raising(self): + suggestion = Suggestion("Fix it?", lambda: False) + check = Check("thing", "error", "broken", suggestion) + with ( + self._only([check]), + patch.object(doctor_mod, "prompt_yes_no_default", return_value=True), + patch.object(doctor_mod, "print_warning") as warn, + ): + rc = doctor() + assert rc == 0 + warn.assert_called() + + def test_ok_check_is_never_prompted(self): + check = Check("thing", "ok", "healthy", None) + with ( + self._only([check]), + patch.object(doctor_mod, "prompt_yes_no_default") as prompt, + ): + doctor() + prompt.assert_not_called()