From 1a0c14e3225865763bbcbf5a87d7b11d7e9eb9b0 Mon Sep 17 00:00:00 2001 From: Smart_Chen Date: Fri, 17 Jul 2026 14:46:13 +0800 Subject: [PATCH] feat(cli): add `contextseek doctor` config & connectivity self-check (#46) Add a `doctor` subcommand that loads ContextSeekSettings, reports the resolved storage/embedding/LLM configuration, and runs a lightweight liveness check on each component: - Storage: constructs the backend and calls initialize() to verify connectivity (supports memory, file, sqlite, seekdb, oceanbase) - Embedding: builds the embedder via factory and probes with embedder("test") - LLM: builds the LLM via factory and probes with llm.invoke([HumanMessage]) Output uses clear PASS/FAIL/SKIP/WARN lines with actionable hints pointing at .env.example sections. Never prints secrets (api keys, passwords, tokens are sanitised from error messages). Exits non-zero if any required component fails; provider=none is treated as SKIP (not failure). Includes 39 unit tests covering all backends, provider states, error classifications, cross-dependency warnings, secret sanitisation, and output format. Closes #46 --- src/contextseek/cli/doctor_cmd.py | 581 ++++++++++++++++++++++++++++++ src/contextseek/cli/main.py | 11 + tests/unit_tests/test_cli.py | 561 +++++++++++++++++++++++++++++ 3 files changed, 1153 insertions(+) create mode 100644 src/contextseek/cli/doctor_cmd.py diff --git a/src/contextseek/cli/doctor_cmd.py b/src/contextseek/cli/doctor_cmd.py new file mode 100644 index 0000000..2e7aeb0 --- /dev/null +++ b/src/contextseek/cli/doctor_cmd.py @@ -0,0 +1,581 @@ +"""``contextseek doctor`` — config & connectivity self-check. + +Loads ``ContextSeekSettings``, reports which backend/embedder/LLM are resolved, +runs a lightweight liveness check on each (e.g. a tiny embedding call, a storage +ping), and prints clear PASS/FAIL/SKIP/WARN lines with actionable hints pointing +at ``.env.example``. + +Never prints secrets. Exits non-zero if any required component fails. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from typing import Any + +from contextseek.config.factory import ( + build_embedder, + build_llm, + resolve_embedding_dims, +) +from contextseek.config.settings import ContextSeekSettings + + +# --------------------------------------------------------------------------- +# Data structures +# --------------------------------------------------------------------------- + +PASS = "PASS" +FAIL = "FAIL" +SKIP = "SKIP" +WARN = "WARN" + + +@dataclass +class CheckResult: + """Outcome of a single doctor check.""" + + status: str + component: str + message: str + hint: str = "" + env_section: str = "" + + +# --------------------------------------------------------------------------- +# Error message sanitisation — never leak secrets +# --------------------------------------------------------------------------- + +_SECRET_PATTERNS: list[tuple[re.Pattern[str], str]] = [ + (re.compile(r"sk-[A-Za-z0-9\-_]{10,}"), "sk-***"), + (re.compile(r"(?i)(password\s*=\s*)\S+"), r"\1***"), + (re.compile(r"(?i)(api[_-]?key\s*=\s*)\S+"), r"\1***"), + (re.compile(r"(?i)(token\s*=\s*)\S+"), r"\1***"), +] + + +def _sanitize_error_message(msg: str, *, max_len: int = 200) -> str: + """Remove secrets from an error message and cap its length.""" + sanitized = msg + for pattern, replacement in _SECRET_PATTERNS: + sanitized = pattern.sub(replacement, sanitized) + if len(sanitized) > max_len: + sanitized = sanitized[:max_len] + "…" + return sanitized + + +# --------------------------------------------------------------------------- +# Error classification +# --------------------------------------------------------------------------- + +def _classify_exception(exc: Exception, component: str) -> tuple[str, str, str]: + """Return (message, hint, env_section) for a caught exception. + + The message is sanitised before being returned. + """ + raw = str(exc) + exc_type = type(exc).__name__ + msg = _sanitize_error_message(raw) + + # Build a short type prefix for the message + short = f"{exc_type}: {msg}" if msg else exc_type + + # --- Import / missing dependency --- + if isinstance(exc, ImportError): + if "pyseekdb" in raw.lower(): + return ( + short, + "Install pyseekdb: pip install pyseekdb", + "section 1.0", + ) + if "pyobvector" in raw.lower() or "sqlalchemy" in raw.lower(): + return ( + short, + "Install OceanBase deps: pip install 'contextseek[oceanbase]'", + "section 1.1", + ) + if "langchain" in raw.lower(): + pkg_hint = "pip install 'contextseek[langchain]'" + if "openai" in raw.lower(): + pkg_hint += " && pip install 'contextseek[openai]'" + elif "dashscope" in raw.lower(): + pkg_hint += " && pip install 'contextseek[dashscope]'" + elif "ollama" in raw.lower(): + pkg_hint += " && pip install 'contextseek[ollama]'" + elif "huggingface" in raw.lower(): + pkg_hint += " && pip install 'contextseek[huggingface]'" + return ( + short, + f"Missing LangChain package: {pkg_hint}", + f"section {'2' if component == 'embedding' else '3'}", + ) + return (short, "Install the missing dependency", f"section {'2' if component == 'embedding' else '3'}") + + # --- Authentication --- + raw_lower = raw.lower() + if ( + "auth" in raw_lower + or "api key" in raw_lower + or "api_key" in raw_lower + or "unauthorized" in raw_lower + or "401" in raw_lower + or "forbidden" in raw_lower + or "403" in raw_lower + ): + if component == "embedding": + return (short, "Set the correct API key (e.g. OPENAI_API_KEY) in .env", "section 2") + return (short, "Set the correct API key (e.g. OPENAI_API_KEY) in .env", "section 3") + + # --- Connection / network --- + if ( + isinstance(exc, (ConnectionError, TimeoutError, OSError)) + or "connection refused" in raw_lower + or "timed out" in raw_lower + or "timeout" in raw_lower + or "unreachable" in raw_lower + or "network" in raw_lower + ): + if component == "storage": + return (short, "Check that the storage service is running and host/port are correct", "section 1 / 1.0 / 1.1") + if component == "embedding": + return (short, "Check EMBEDDING_BASE_URL and network connectivity", "section 2") + return (short, "Check LLM_BASE_URL and network connectivity", "section 3") + + # --- Configuration / value errors --- + if isinstance(exc, ValueError): + if "dims" in raw_lower or "embedding_dims" in raw_lower: + return (short, "Set EMBEDDING_DIMS when using oceanbase backend", "section 1.1 / 2") + if "unknown" in raw_lower and "provider" in raw_lower: + if component == "embedding": + return (short, "Check EMBEDDING_PROVIDER spelling; see supported providers", "section 2") + return (short, "Check LLM_PROVIDER spelling; see supported providers", "section 3") + return (short, "Check configuration values in .env", "section 1 / 2 / 3") + + # --- Generic fallback --- + return (short, "See .env.example for configuration reference", "section 1 / 2 / 3") + + +# --------------------------------------------------------------------------- +# Individual checks +# --------------------------------------------------------------------------- + +def _check_storage(settings: ContextSeekSettings) -> CheckResult: + """Build and initialise the storage backend to verify connectivity.""" + storage = settings.storage + backend_name = storage.backend + + # --- oceanbase requires EMBEDDING_DIMS --- + if backend_name == "oceanbase": + vector_dims = settings.embedding.dims + if not vector_dims: + return CheckResult( + FAIL, + "storage", + "EMBEDDING_DIMS must be set when STORAGE_BACKEND=oceanbase", + "Set EMBEDDING_DIMS (e.g. 1536 for OpenAI text-embedding-3-small)", + "section 1.1 / 2", + ) + + # --- Construct the backend (mirrors client/contextseek.py:261-318) --- + backend: Any + try: + if backend_name == "oceanbase": + from contextseek.storage.ob_backend import OceanBaseBackend + + ob = settings.ob + geo = getattr(settings, "geo", None) + if geo is not None and getattr(geo, "enabled", False): + from contextseek.storage.ob_geo_backend import OceanBaseGeoBackend + + backend = OceanBaseGeoBackend( + table_name=ob.table_name, + vector_dims=vector_dims, + host=ob.host, + port=ob.port, + user=ob.user, + password=ob.password, + db_name=ob.db_name, + geo_table_name=geo.geo_table_name, + distance_decay_km=geo.distance_decay_km, + route_sample_interval_km=geo.route_sample_interval_km, + ) + else: + backend = OceanBaseBackend( + table_name=ob.table_name, + vector_dims=vector_dims, + host=ob.host, + port=ob.port, + user=ob.user, + password=ob.password, + db_name=ob.db_name, + ) + elif backend_name == "sqlite": + from contextseek.storage.sqlite_backend import SQLiteBackend + + backend = SQLiteBackend(path=settings.sqlite.path) + elif backend_name == "seekdb": + from contextseek.storage.seekdb_backend import SeekDBBackend + + seekdb = settings.seekdb + backend = SeekDBBackend( + path=seekdb.path, + database=seekdb.database, + host=seekdb.host, + port=seekdb.port, + ) + elif backend_name == "file": + from contextseek.storage.file_backend import FileBackend + + backend = FileBackend(root_dir=storage.path) + elif backend_name == "memory": + from contextseek.storage.in_memory_backend import InMemoryBackend + + backend = InMemoryBackend() + else: + return CheckResult( + FAIL, + "storage", + f"Unknown storage backend: {backend_name!r}", + "Set STORAGE_BACKEND to one of: memory, file, sqlite, seekdb, oceanbase", + "section 1", + ) + except Exception as exc: + msg, hint, section = _classify_exception(exc, "storage") + return CheckResult(FAIL, "storage", msg, hint, section) + + # --- Initialise (this is the actual connectivity check) --- + try: + backend.initialize() + except Exception as exc: + msg, hint, section = _classify_exception(exc, "storage") + return CheckResult(FAIL, "storage", msg, hint, section) + + # --- Clean up --- + close = getattr(backend, "close", None) + if callable(close): + try: + close() + except Exception: + pass + + return CheckResult(PASS, "storage", f"{backend_name} backend initialized") + + +def _check_embedding(settings: ContextSeekSettings) -> CheckResult: + """Build embedder and do a tiny probe call.""" + emb_settings = settings.embedding + provider = emb_settings.provider.strip().lower() if emb_settings.provider else "none" + + if provider in {"", "none"}: + return CheckResult( + SKIP, + "embedding", + "provider=none — vector search disabled", + "Set EMBEDDING_PROVIDER to enable vector retrieval", + "section 2", + ) + + # langchain provider requires class_path + if provider == "langchain" and not emb_settings.class_path: + return CheckResult( + SKIP, + "embedding", + "provider=langchain but EMBEDDING_CLASS_PATH not set", + "Set EMBEDDING_CLASS_PATH to your custom LangChain embeddings class", + "section 2", + ) + + # Build the embedder + try: + embedder = build_embedder(emb_settings) + except Exception as exc: + msg, hint, section = _classify_exception(exc, "embedding") + return CheckResult(FAIL, "embedding", msg, hint, section) + + if embedder is None: + return CheckResult( + SKIP, + "embedding", + f"provider={provider} resolved to None — vector search disabled", + "Check EMBEDDING_PROVIDER / EMBEDDING_CLASS_PATH configuration", + "section 2", + ) + + # Probe call + try: + vec = embedder("test") + except Exception as exc: + msg, hint, section = _classify_exception(exc, "embedding") + return CheckResult(FAIL, "embedding", msg, hint, section) + + if not vec or not isinstance(vec, list): + return CheckResult( + FAIL, + "embedding", + f"embedder returned empty/invalid result for probe: {type(vec).__name__}", + "Check EMBEDDING_MODEL and provider configuration", + "section 2", + ) + + actual_dims = len(vec) + configured_dims = resolve_embedding_dims(emb_settings) + if configured_dims and configured_dims != actual_dims: + return CheckResult( + WARN, + "embedding", + f"embedder returned {actual_dims}-dim vector (configured dims={configured_dims})", + f"Update EMBEDDING_DIMS to {actual_dims} or verify EMBEDDING_MODEL", + "section 2", + ) + + return CheckResult( + PASS, + "embedding", + f"embedder returned {actual_dims}-dim vector for probe \"test\"", + ) + + +def _check_llm(settings: ContextSeekSettings) -> CheckResult: + """Build LLM and do a tiny invoke call.""" + llm_settings = settings.llm + provider = llm_settings.provider.strip().lower() if llm_settings.provider else "none" + + if provider in {"", "none"}: + return CheckResult( + SKIP, + "llm", + "provider=none — rerank/summarize/evolution LLM disabled", + "Set LLM_PROVIDER to enable LLM-powered features", + "section 3", + ) + + # langchain provider requires class_path + if provider == "langchain" and not llm_settings.class_path: + return CheckResult( + SKIP, + "llm", + "provider=langchain but LLM_CLASS_PATH not set", + "Set LLM_CLASS_PATH to your custom LangChain chat model class", + "section 3", + ) + + # Build the LLM + try: + llm = build_llm(llm_settings) + except Exception as exc: + msg, hint, section = _classify_exception(exc, "llm") + return CheckResult(FAIL, "llm", msg, hint, section) + + if llm is None: + return CheckResult( + SKIP, + "llm", + f"provider={provider} resolved to None — LLM disabled", + "Check LLM_PROVIDER / LLM_CLASS_PATH configuration", + "section 3", + ) + + # Probe call — directly invoke to get precise error reporting + try: + from langchain_core.messages import HumanMessage + + resp = llm.invoke([HumanMessage(content="hello")]) + except Exception as exc: + msg, hint, section = _classify_exception(exc, "llm") + return CheckResult(FAIL, "llm", msg, hint, section) + + if resp is None: + return CheckResult( + FAIL, + "llm", + "LLM invoke returned None", + "Check LLM_MODEL and provider configuration", + "section 3", + ) + + return CheckResult(PASS, "llm", f"LLM responded to probe (provider={provider})") + + +def _check_cross( + settings: ContextSeekSettings, + results: list[CheckResult], +) -> list[CheckResult]: + """Cross-dependency checks that produce WARN (never FAIL).""" + warnings: list[CheckResult] = [] + + backend_name = settings.storage.backend + emb_provider = settings.embedding.provider.strip().lower() if settings.embedding.provider else "none" + llm_provider = settings.llm.provider.strip().lower() if settings.llm.provider else "none" + + # seekdb + no external embedder + if backend_name == "seekdb" and emb_provider in {"", "none"}: + warnings.append( + CheckResult( + WARN, + "cross", + "seekdb backend with no external embedder — using built-in 384-dim embedding", + "Set EMBEDDING_PROVIDER for higher-quality vectors", + "section 2", + ) + ) + + # LLM configured but embedding not — retrieval will lack vector route + if llm_provider not in {"", "none"} and emb_provider in {"", "none"}: + warnings.append( + CheckResult( + WARN, + "cross", + "LLM configured but embedding is not — retrieval recall routes will not include vector search", + "Set EMBEDDING_PROVIDER to enable vector retrieval", + "section 2", + ) + ) + + return warnings + + +# --------------------------------------------------------------------------- +# Configuration report (no side effects) +# --------------------------------------------------------------------------- + +def _describe_storage(settings: ContextSeekSettings) -> str: + """Human-readable description of the resolved storage config (no secrets).""" + s = settings.storage + backend = s.backend + if backend == "sqlite": + return f"sqlite (path={settings.sqlite.path})" + if backend == "seekdb": + seekdb = settings.seekdb + if seekdb.host: + return f"seekdb (host={seekdb.host}:{seekdb.port}, database={seekdb.database})" + return f"seekdb embedded (path={seekdb.path}, database={seekdb.database})" + if backend == "oceanbase": + ob = settings.ob + return f"oceanbase (host={ob.host}:{ob.port}, user={ob.user}, db={ob.db_name}, table={ob.table_name})" + if backend == "file": + return f"file (path={s.path})" + if backend == "memory": + return "memory" + return f"{backend} (unknown)" + + +def _describe_embedding(settings: ContextSeekSettings) -> str: + """Human-readable description of the resolved embedding config (no secrets).""" + emb = settings.embedding + provider = emb.provider or "none" + if provider in {"", "none"}: + return "none" + parts = [f"provider={provider}"] + if emb.model: + parts.append(f"model={emb.model}") + dims = resolve_embedding_dims(emb) + if dims: + parts.append(f"dims={dims}") + if emb.base_url: + parts.append(f"base_url={emb.base_url}") + return ", ".join(parts) + + +def _describe_llm(settings: ContextSeekSettings) -> str: + """Human-readable description of the resolved LLM config (no secrets).""" + llm = settings.llm + provider = llm.provider or "none" + if provider in {"", "none"}: + return "none" + parts = [f"provider={provider}"] + if llm.model: + parts.append(f"model={llm.model}") + if llm.base_url: + parts.append(f"base_url={llm.base_url}") + return ", ".join(parts) + + +# --------------------------------------------------------------------------- +# Output rendering +# --------------------------------------------------------------------------- + +_STATUS_LABELS = { + PASS: "[PASS]", + FAIL: "[FAIL]", + SKIP: "[SKIP]", + WARN: "[WARN]", +} + + +def _print(text: str = "") -> None: + print(text) + + +def _render_results(results: list[CheckResult]) -> None: + """Print check results in human-readable format.""" + _print() + _print("Checks:") + for r in results: + label = _STATUS_LABELS.get(r.status, f"[{r.status}]") + _print(f" {label} {r.component:<10} {r.message}") + if r.hint: + section = f"; see .env.example ({r.env_section})" if r.env_section else "" + indent = " ↳ fix: " if r.status == FAIL else " ↳ hint: " + _print(f"{indent}{r.hint}{section}") + _print() + + +def _render_summary(results: list[CheckResult]) -> int: + """Print summary line and return exit code.""" + counts = {PASS: 0, FAIL: 0, SKIP: 0, WARN: 0} + for r in results: + counts[r.status] = counts.get(r.status, 0) + 1 + + parts = [] + if counts[PASS]: + parts.append(f"{counts[PASS]} PASS") + if counts[FAIL]: + parts.append(f"{counts[FAIL]} FAIL") + if counts[SKIP]: + parts.append(f"{counts[SKIP]} SKIP") + if counts[WARN]: + parts.append(f"{counts[WARN]} WARN") + + exit_code = 1 if counts[FAIL] > 0 else 0 + suffix = f" — exit {exit_code}" if exit_code else "" + _print(f"Result: {', '.join(parts) if parts else 'no checks'}{suffix}") + return exit_code + + +# --------------------------------------------------------------------------- +# Main entry point +# --------------------------------------------------------------------------- + +def run_doctor(settings: ContextSeekSettings) -> int: + """Run the doctor diagnostics and return an exit code. + + Exit code is 1 if any check FAILs, 0 otherwise. + """ + # --- Configuration report (no side effects) --- + _print() + _print("ContextSeek doctor — configuration diagnostics") + _print() + _print("Configuration resolved:") + _print(f" storage : {_describe_storage(settings)}") + _print(f" embedding : {_describe_embedding(settings)}") + _print(f" llm : {_describe_llm(settings)}") + + # --- Run checks --- + results: list[CheckResult] = [] + + # Storage check — wrap in suppress_backend_noise to avoid pyseekdb etc. + from contextseek.cli.ui import suppress_backend_noise + + with suppress_backend_noise(): + results.append(_check_storage(settings)) + + results.append(_check_embedding(settings)) + results.append(_check_llm(settings)) + + # Cross-dependency warnings + results.extend(_check_cross(settings, results)) + + # --- Render --- + _render_results(results) + return _render_summary(results) \ No newline at end of file diff --git a/src/contextseek/cli/main.py b/src/contextseek/cli/main.py index 340f74f..3205949 100644 --- a/src/contextseek/cli/main.py +++ b/src/contextseek/cli/main.py @@ -311,6 +311,12 @@ def build_parser() -> argparse.ArgumentParser: help="initialize ~/.contextseek/: generate config.env, mcp.json, register system service", ) + # doctor — config & connectivity self-check + subparsers.add_parser( + "doctor", + help="diagnose storage / embedding / LLM configuration and connectivity", + ) + # daemon daemon_parser = subparsers.add_parser("daemon", help="manage the background daemon") daemon_sub = daemon_parser.add_subparsers(dest="daemon_command", required=True) @@ -500,6 +506,11 @@ def run_cli( run_init(pathlib.Path.home() / ".contextseek") return 0 + if args.command == "doctor": + from contextseek.cli.doctor_cmd import run_doctor + + return run_doctor(settings) + if args.command == "daemon" and args.daemon_command in {"stop", "status"}: from contextseek.daemon.process import DaemonProcess import pathlib diff --git a/tests/unit_tests/test_cli.py b/tests/unit_tests/test_cli.py index 11fa121..eee17bd 100644 --- a/tests/unit_tests/test_cli.py +++ b/tests/unit_tests/test_cli.py @@ -182,3 +182,564 @@ def test_expand_reports_missing_ids(self) -> None: assert code == 0 assert [it["id"] for it in payload["items"]] == [item.id] assert payload["missing_ids"] == ["missing-id"] + + +class TestDoctor: + """Tests for the ``contextseek doctor`` subcommand.""" + + # ------------------------------------------------------------------ + # Parser registration + # ------------------------------------------------------------------ + + def test_doctor_parser_registered(self) -> None: + """``doctor`` subcommand is registered and parseable.""" + parser = build_parser() + args = parser.parse_args(["doctor"]) + assert args.command == "doctor" + + def test_doctor_in_help(self) -> None: + """``doctor`` appears in the main parser help text.""" + parser = build_parser() + help_text = parser.format_help() + assert "doctor" in help_text + assert "diagnose" in help_text + + # ------------------------------------------------------------------ + # Storage checks + # ------------------------------------------------------------------ + + def test_doctor_all_none_passes(self, capsys) -> None: + """Default isolated env (memory/none/none) → exit 0 with PASS + SKIPs.""" + code = run_cli(["doctor"]) + out = capsys.readouterr().out + + assert code == 0 + assert "[PASS] storage" in out + assert "[SKIP] embedding" in out + assert "[SKIP] llm" in out + assert "[FAIL]" not in out + assert "exit 0" not in out # no failure, no exit suffix + # Should mention .env.example in the SKIP hints + assert ".env.example" in out + + def test_doctor_storage_memory_pass(self, monkeypatch, capsys) -> None: + """Memory backend → PASS.""" + monkeypatch.setenv("STORAGE_BACKEND", "memory") + code = run_cli(["doctor"]) + out = capsys.readouterr().out + + assert code == 0 + assert "[PASS] storage" in out + assert "memory backend initialized" in out + + def test_doctor_storage_sqlite_pass(self, monkeypatch, tmp_path, capsys) -> None: + """SQLite backend with a temp path → PASS.""" + db_path = str(tmp_path / "test.sqlite3") + monkeypatch.setenv("STORAGE_BACKEND", "sqlite") + monkeypatch.setenv("SQLITE_PATH", db_path) + code = run_cli(["doctor"]) + out = capsys.readouterr().out + + assert code == 0 + assert "[PASS] storage" in out + assert "sqlite backend initialized" in out + + def test_doctor_storage_file_pass(self, monkeypatch, tmp_path, capsys) -> None: + """File backend with a temp path → PASS.""" + file_path = str(tmp_path / "store") + monkeypatch.setenv("STORAGE_BACKEND", "file") + monkeypatch.setenv("STORAGE_PATH", file_path) + code = run_cli(["doctor"]) + out = capsys.readouterr().out + + assert code == 0 + assert "[PASS] storage" in out + assert "file backend initialized" in out + + def test_doctor_storage_oceanbase_missing_dims_fails( + self, monkeypatch, capsys + ) -> None: + """OceanBase backend without EMBEDDING_DIMS → storage FAIL, exit 1.""" + monkeypatch.setenv("STORAGE_BACKEND", "oceanbase") + monkeypatch.setenv("EMBEDDING_PROVIDER", "none") + monkeypatch.setenv("EMBEDDING_DIMS", "0") + code = run_cli(["doctor"]) + out = capsys.readouterr().out + + assert code == 1 + assert "[FAIL] storage" in out + assert "EMBEDDING_DIMS" in out + assert ".env.example" in out + + def test_doctor_storage_unknown_backend_fails(self, monkeypatch, capsys) -> None: + """Unknown storage backend → FAIL.""" + monkeypatch.setenv("STORAGE_BACKEND", "unknown_backend") + code = run_cli(["doctor"]) + out = capsys.readouterr().out + + assert code == 1 + assert "[FAIL] storage" in out + assert "Unknown storage backend" in out + assert "memory, file, sqlite, seekdb, oceanbase" in out + + def test_doctor_storage_empty_backend_fails(self, monkeypatch, capsys) -> None: + """Empty storage backend string → FAIL (unknown).""" + monkeypatch.setenv("STORAGE_BACKEND", "") + code = run_cli(["doctor"]) + out = capsys.readouterr().out + + assert code == 1 + assert "[FAIL] storage" in out + + # ------------------------------------------------------------------ + # Embedding checks + # ------------------------------------------------------------------ + + def test_doctor_embedding_none_skips(self, capsys) -> None: + """Embedding provider=none → SKIP.""" + code = run_cli(["doctor"]) + out = capsys.readouterr().out + + assert code == 0 + assert "[SKIP] embedding" in out + assert "vector search disabled" in out + + def test_doctor_embedding_langchain_no_class_path_skips( + self, monkeypatch, capsys + ) -> None: + """Embedding provider=langchain without class_path → SKIP.""" + monkeypatch.setenv("EMBEDDING_PROVIDER", "langchain") + code = run_cli(["doctor"]) + out = capsys.readouterr().out + + assert code == 0 + assert "[SKIP] embedding" in out + assert "EMBEDDING_CLASS_PATH" in out + + def test_doctor_embedding_build_failure(self, monkeypatch, capsys) -> None: + """Embedding provider configured but build_embedder raises → FAIL.""" + monkeypatch.setenv("EMBEDDING_PROVIDER", "openai") + monkeypatch.setenv("EMBEDDING_MODEL", "text-embedding-3-small") + + def _fake_build_embedder(settings): + raise PermissionError("AuthenticationError: Incorrect API key provided") + + monkeypatch.setattr( + "contextseek.cli.doctor_cmd.build_embedder", _fake_build_embedder + ) + code = run_cli(["doctor"]) + out = capsys.readouterr().out + + assert code == 1 + assert "[FAIL] embedding" in out + assert "API key" in out + assert ".env.example" in out + + def test_doctor_embedding_probe_success(self, monkeypatch, capsys) -> None: + """Embedder successfully returns a vector → PASS.""" + monkeypatch.setenv("EMBEDDING_PROVIDER", "openai") + monkeypatch.setenv("EMBEDDING_MODEL", "text-embedding-3-small") + monkeypatch.setenv("EMBEDDING_DIMS", "4") + + def _fake_embedder(text): + return [0.1, 0.2, 0.3, 0.4] + + monkeypatch.setattr( + "contextseek.cli.doctor_cmd.build_embedder", lambda s: _fake_embedder + ) + code = run_cli(["doctor"]) + out = capsys.readouterr().out + + assert code == 0 + assert "[PASS] embedding" in out + assert "4-dim" in out + + def test_doctor_embedding_probe_returns_empty_fails( + self, monkeypatch, capsys + ) -> None: + """Embedder returns empty list → FAIL.""" + monkeypatch.setenv("EMBEDDING_PROVIDER", "openai") + + def _fake_embedder(text): + return [] + + monkeypatch.setattr( + "contextseek.cli.doctor_cmd.build_embedder", lambda s: _fake_embedder + ) + code = run_cli(["doctor"]) + out = capsys.readouterr().out + + assert code == 1 + assert "[FAIL] embedding" in out + + def test_doctor_embedding_probe_raises_fails(self, monkeypatch, capsys) -> None: + """Embedder probe call raises → FAIL.""" + monkeypatch.setenv("EMBEDDING_PROVIDER", "openai") + + def _fake_embedder(text): + raise ConnectionError("Connection refused to embedding server") + + monkeypatch.setattr( + "contextseek.cli.doctor_cmd.build_embedder", lambda s: _fake_embedder + ) + code = run_cli(["doctor"]) + out = capsys.readouterr().out + + assert code == 1 + assert "[FAIL] embedding" in out + assert "Connection" in out or "connection" in out + + def test_doctor_embedding_dims_mismatch_warns(self, monkeypatch, capsys) -> None: + """Embedder returns dims different from configured → WARN.""" + monkeypatch.setenv("EMBEDDING_PROVIDER", "openai") + monkeypatch.setenv("EMBEDDING_DIMS", "1536") + + def _fake_embedder(text): + return [0.1] * 768 # returns 768 but configured 1536 + + monkeypatch.setattr( + "contextseek.cli.doctor_cmd.build_embedder", lambda s: _fake_embedder + ) + code = run_cli(["doctor"]) + out = capsys.readouterr().out + + # WARN does not cause failure + assert code == 0 + assert "[WARN] embedding" in out + assert "768-dim" in out + assert "1536" in out + + def test_doctor_embedding_probe_returns_non_list_fails( + self, monkeypatch, capsys + ) -> None: + """Embedder returns non-list (e.g. None) → FAIL.""" + monkeypatch.setenv("EMBEDDING_PROVIDER", "openai") + + monkeypatch.setattr( + "contextseek.cli.doctor_cmd.build_embedder", lambda s: (lambda t: None) + ) + code = run_cli(["doctor"]) + out = capsys.readouterr().out + + assert code == 1 + assert "[FAIL] embedding" in out + + # ------------------------------------------------------------------ + # LLM checks + # ------------------------------------------------------------------ + + def test_doctor_llm_none_skips(self, capsys) -> None: + """LLM provider=none → SKIP.""" + code = run_cli(["doctor"]) + out = capsys.readouterr().out + + assert code == 0 + assert "[SKIP] llm" in out + assert "rerank/summarize/evolution" in out + + def test_doctor_llm_langchain_no_class_path_skips( + self, monkeypatch, capsys + ) -> None: + """LLM provider=langchain without class_path → SKIP.""" + monkeypatch.setenv("LLM_PROVIDER", "langchain") + code = run_cli(["doctor"]) + out = capsys.readouterr().out + + assert code == 0 + assert "[SKIP] llm" in out + assert "LLM_CLASS_PATH" in out + + def test_doctor_llm_build_failure(self, monkeypatch, capsys) -> None: + """LLM build_llm raises → FAIL.""" + monkeypatch.setenv("LLM_PROVIDER", "openai") + + def _fake_build_llm(settings): + raise ImportError("No module named 'langchain_openai'") + + monkeypatch.setattr("contextseek.cli.doctor_cmd.build_llm", _fake_build_llm) + code = run_cli(["doctor"]) + out = capsys.readouterr().out + + assert code == 1 + assert "[FAIL] llm" in out + assert "langchain" in out.lower() + + def test_doctor_llm_invoke_failure(self, monkeypatch, capsys) -> None: + """LLM provider configured but invoke raises → FAIL.""" + monkeypatch.setenv("LLM_PROVIDER", "openai") + monkeypatch.setenv("LLM_MODEL", "gpt-4o-mini") + + class _FakeLLM: + def invoke(self, messages): + raise ConnectionError("Connection refused to api.openai.com") + + monkeypatch.setattr( + "contextseek.cli.doctor_cmd.build_llm", lambda s: _FakeLLM() + ) + code = run_cli(["doctor"]) + out = capsys.readouterr().out + + assert code == 1 + assert "[FAIL] llm" in out + assert ".env.example" in out + + def test_doctor_llm_invoke_success(self, monkeypatch, capsys) -> None: + """LLM responds to probe → PASS.""" + monkeypatch.setenv("LLM_PROVIDER", "openai") + monkeypatch.setenv("LLM_MODEL", "gpt-4o-mini") + + class _FakeLLM: + def invoke(self, messages): + return "hello back" + + monkeypatch.setattr( + "contextseek.cli.doctor_cmd.build_llm", lambda s: _FakeLLM() + ) + code = run_cli(["doctor"]) + out = capsys.readouterr().out + + assert code == 0 + assert "[PASS] llm" in out + + def test_doctor_llm_invoke_returns_none_fails(self, monkeypatch, capsys) -> None: + """LLM invoke returns None → FAIL.""" + monkeypatch.setenv("LLM_PROVIDER", "openai") + + class _FakeLLM: + def invoke(self, messages): + return None + + monkeypatch.setattr( + "contextseek.cli.doctor_cmd.build_llm", lambda s: _FakeLLM() + ) + code = run_cli(["doctor"]) + out = capsys.readouterr().out + + assert code == 1 + assert "[FAIL] llm" in out + assert "None" in out + + # ------------------------------------------------------------------ + # Cross-dependency checks + # ------------------------------------------------------------------ + + def test_doctor_cross_dependency_seekdb_warn(self, monkeypatch, capsys) -> None: + """seekdb + embedding=none → WARN about built-in embedding.""" + monkeypatch.setenv("STORAGE_BACKEND", "seekdb") + monkeypatch.setenv("EMBEDDING_PROVIDER", "none") + + from contextseek.cli.doctor_cmd import CheckResult, PASS + + def _fake_check_storage(settings): + return CheckResult(PASS, "storage", "seekdb backend initialized") + + monkeypatch.setattr( + "contextseek.cli.doctor_cmd._check_storage", _fake_check_storage + ) + code = run_cli(["doctor"]) + out = capsys.readouterr().out + + assert code == 0 + assert "[WARN] cross" in out + assert "seekdb" in out.lower() + + def test_doctor_cross_dependency_llm_without_embedding_warn( + self, monkeypatch, capsys + ) -> None: + """LLM configured but embedding=none → WARN about missing vector search.""" + monkeypatch.setenv("EMBEDDING_PROVIDER", "none") + monkeypatch.setenv("LLM_PROVIDER", "openai") + + from contextseek.cli.doctor_cmd import CheckResult, PASS + + def _fake_check_llm(settings): + return CheckResult(PASS, "llm", "LLM responded") + + monkeypatch.setattr( + "contextseek.cli.doctor_cmd._check_llm", _fake_check_llm + ) + code = run_cli(["doctor"]) + out = capsys.readouterr().out + + assert code == 0 + assert "[WARN] cross" in out + assert "vector search" in out.lower() + + def test_doctor_no_cross_warning_when_both_configured( + self, monkeypatch, capsys + ) -> None: + """Both embedding and LLM configured → no cross WARN.""" + monkeypatch.setenv("EMBEDDING_PROVIDER", "openai") + monkeypatch.setenv("LLM_PROVIDER", "openai") + + monkeypatch.setattr( + "contextseek.cli.doctor_cmd.build_embedder", + lambda s: (lambda t: [0.1, 0.2]), + ) + + class _FakeLLM: + def invoke(self, messages): + return "ok" + + monkeypatch.setattr( + "contextseek.cli.doctor_cmd.build_llm", lambda s: _FakeLLM() + ) + code = run_cli(["doctor"]) + out = capsys.readouterr().out + + assert code == 0 + assert "[WARN] cross" not in out + + # ------------------------------------------------------------------ + # Secret sanitisation + # ------------------------------------------------------------------ + + def test_doctor_no_secrets_leaked_sk_key(self, monkeypatch, capsys) -> None: + """Doctor output must never contain raw sk- keys from error messages.""" + fake_secret = "sk-FAKESECRET1234567890abcdef" + + def _fake_build_embedder(settings): + raise PermissionError( + f"AuthenticationError: Incorrect API key provided: {fake_secret}" + ) + + monkeypatch.setenv("EMBEDDING_PROVIDER", "openai") + monkeypatch.setattr( + "contextseek.cli.doctor_cmd.build_embedder", _fake_build_embedder + ) + code = run_cli(["doctor"]) + out = capsys.readouterr().out + + assert code == 1 + assert fake_secret not in out + assert "sk-***" in out + + def test_doctor_no_secrets_leaked_password(self, monkeypatch, capsys) -> None: + """Doctor output must not leak password= values from error messages.""" + fake_password = "SuperSecret123!" + + def _fake_build_embedder(settings): + raise ValueError(f"Connection failed: password={fake_password} rejected") + + monkeypatch.setenv("EMBEDDING_PROVIDER", "openai") + monkeypatch.setattr( + "contextseek.cli.doctor_cmd.build_embedder", _fake_build_embedder + ) + code = run_cli(["doctor"]) + out = capsys.readouterr().out + + assert code == 1 + assert fake_password not in out + assert "password=***" in out + + def test_doctor_no_secrets_leaked_token(self, monkeypatch, capsys) -> None: + """Doctor output must not leak token= values from error messages.""" + fake_token = "tok_abcDEF123456" + + def _fake_build_embedder(settings): + raise RuntimeError(f"Auth failed: token={fake_token} is invalid") + + monkeypatch.setenv("EMBEDDING_PROVIDER", "openai") + monkeypatch.setattr( + "contextseek.cli.doctor_cmd.build_embedder", _fake_build_embedder + ) + code = run_cli(["doctor"]) + out = capsys.readouterr().out + + assert code == 1 + assert fake_token not in out + assert "token=***" in out + + def test_doctor_config_report_no_password(self, monkeypatch, capsys) -> None: + """OceanBase config report must not include the password field.""" + monkeypatch.setenv("STORAGE_BACKEND", "oceanbase") + monkeypatch.setenv("OB_PASSWORD", "should_not_appear_in_output") + monkeypatch.setenv("EMBEDDING_DIMS", "1536") + + # Patch storage check to avoid real OB connection + from contextseek.cli.doctor_cmd import CheckResult, PASS + + monkeypatch.setattr( + "contextseek.cli.doctor_cmd._check_storage", + lambda s: CheckResult(PASS, "storage", "ok"), + ) + run_cli(["doctor"]) + out = capsys.readouterr().out + + assert "should_not_appear_in_output" not in out + + # ------------------------------------------------------------------ + # Sanitisation unit tests + # ------------------------------------------------------------------ + + def test_sanitize_truncates_long_messages(self) -> None: + """_sanitize_error_message truncates messages over 200 chars.""" + from contextseek.cli.doctor_cmd import _sanitize_error_message + + long_msg = "x" * 300 + result = _sanitize_error_message(long_msg) + assert len(result) <= 201 # 200 + ellipsis + assert result.endswith("…") + + def test_sanitize_preserves_short_messages(self) -> None: + """_sanitize_error_message preserves short messages unchanged.""" + from contextseek.cli.doctor_cmd import _sanitize_error_message + + assert _sanitize_error_message("short error") == "short error" + + def test_sanitize_sk_key_pattern(self) -> None: + """_sanitize_error_message masks sk- prefixed keys.""" + from contextseek.cli.doctor_cmd import _sanitize_error_message + + result = _sanitize_error_message("error: sk-ABCDEFGHIJKLMNOPQRSTUVWXYZ1234") + assert "sk-***" in result + assert "ABCDEFGHIJKLMNOPQRSTUVWXYZ1234" not in result + + # ------------------------------------------------------------------ + # Output format structure + # ------------------------------------------------------------------ + + def test_doctor_output_has_header(self, capsys) -> None: + """Doctor output starts with the diagnostic header.""" + run_cli(["doctor"]) + out = capsys.readouterr().out + + assert "ContextSeek doctor" in out + assert "configuration diagnostics" in out.lower() + + def test_doctor_output_has_config_resolved_section(self, capsys) -> None: + """Doctor output includes a 'Configuration resolved:' section.""" + run_cli(["doctor"]) + out = capsys.readouterr().out + + assert "Configuration resolved:" in out + assert "storage" in out + assert "embedding" in out + assert "llm" in out + + def test_doctor_output_has_checks_section(self, capsys) -> None: + """Doctor output includes a 'Checks:' section.""" + run_cli(["doctor"]) + out = capsys.readouterr().out + + assert "Checks:" in out + + def test_doctor_output_has_result_summary(self, capsys) -> None: + """Doctor output ends with a 'Result:' summary line.""" + run_cli(["doctor"]) + out = capsys.readouterr().out + + assert "Result:" in out + assert "PASS" in out + assert "SKIP" in out + + def test_doctor_exit_code_nonzero_on_fail(self, monkeypatch, capsys) -> None: + """Exit code is 1 when any check FAILs.""" + monkeypatch.setenv("STORAGE_BACKEND", "bad_backend") + code = run_cli(["doctor"]) + assert code == 1 + + def test_doctor_exit_code_zero_on_all_pass_skip(self, capsys) -> None: + """Exit code is 0 when no checks FAIL.""" + code = run_cli(["doctor"]) + assert code == 0