verify.py

r"""Verify the whole system works.

Run:  .\.venv\Scripts\python.exe verify.py
"""

from __future__ import annotations

import os
import subprocess
import sys
import time
from pathlib import Path

ROOT = Path(__file__).resolve().parent
PY = str(ROOT / ".venv" / "Scripts" / "python.exe")
LUMAN = str(ROOT / ".venv" / "Scripts" / "luman.exe")
if not Path(PY).exists():
    PY, LUMAN = sys.executable, sys.executable

# Keep third-party chatter off the console so the check list is readable.
CHILD_ENV = dict(os.environ)
CHILD_ENV.update({
    "PYTHONWARNINGS": "ignore",
    "HF_HUB_DISABLE_SYMLINKS_WARNING": "1",
    "HF_HUB_DISABLE_PROGRESS_BARS": "1",
    "TOKENIZERS_PARALLELISM": "false",
    "TRANSFORMERS_VERBOSITY": "error",
    "TQDM_DISABLE": "1",
})

results: list[tuple[bool, str, str]] = []


def run(label, args, expect, timeout=900, show=False):
    started = time.perf_counter()
    try:
        proc = subprocess.run(
            args, cwd=ROOT, capture_output=True, text=True,
            timeout=timeout, encoding="utf-8", errors="replace", env=CHILD_ENV,
        )
        out = (proc.stdout + proc.stderr).lower()
        ok = all(word.lower() in out for word in expect)
        note = ""
        if show:
            lines = [l for l in (proc.stdout or "").splitlines()
                     if l.strip() and "warning" not in l.lower()][:4]
            note = " | ".join(lines)[:150]
        seconds = time.perf_counter() - started
        results.append((ok, label, f"{note[:24]}{note and ' ' or ''}{seconds:.0f}s"))
    except subprocess.TimeoutExpired:
        results.append((False, label, "TIMEOUT"))
    except Exception as exc:
        results.append((False, label, f"{type(exc).__name__}: {exc}"))


def check(label, fn):
    try:
        detail = fn()
        results.append((True, label, detail or ""))
    except Exception as exc:
        results.append((False, label, f"{type(exc).__name__}: {exc}"))


def ollama_up():
    from app.generation.llm import OllamaClient
    return OllamaClient().is_available()


def package_installed():
    from importlib.metadata import version
    return f"luman {version('luman')}"


def models_cached():
    from app.config import EmbeddingConfig
    from app.embeddings.encoder import SentenceTransformerEmbedder
    e = SentenceTransformerEmbedder(EmbeddingConfig())
    return f"embeddings {e.dim}-dim ready"


print("Running checks...\n")

check("1. package installed", package_installed)
check("2. embedding model", models_cached)
check("3. ollama reachable", lambda: "yes" if ollama_up() else "NO - run 'ollama serve'")

run("4. unit tests", [PY, "-m", "pytest", "tests/", "-q", "-m", "not integration"],
    ["passed"], timeout=900)
run("5. index consistency", [PY, "-m", "app.cli", "verify"], ["consistent"], timeout=300)
run("6. hybrid search", [PY, "ask.py", "--search", "What is the value of maxUnavailable?"],
    ["deployment.md"], timeout=300)
run("7. dense search finds doc", [PY, "ask.py", "--search", "--dense", "access tokens"],
    ["api-auth.md"], timeout=300)
run("8. answers a question", [PY, "ask.py", "how long are access tokens valid?"],
    ["60 minutes"], timeout=300)
run("9. refuses unanswerable", [PY, "ask.py", "what is the office wifi password?"],
    ["could not find this in the documentation"], timeout=300)
run("10. abstains on absent pdf", [PY, "ask.py", "what is the penguins home address?"],
    ["could not find this in the documentation"], timeout=300)
run("11. luman command", [LUMAN, "--help"], ["luman chat"], timeout=120)
run("12. luman status", [LUMAN, "status"], ["documents"], timeout=300)
run("13. dataset valid", [PY, "-m", "app.cli", "check-dataset"], ["no problems"], timeout=120)

print("=" * 64)
passed = sum(1 for ok, _, _ in results if ok)
for ok, label, note in results:
    mark = "PASS" if ok else "FAIL"
    print(f"  [{mark}] {label:<28} {note[:35]}")
print("=" * 64)
print(f"  {passed}/{len(results)} checks passed")
print("=" * 64)

sys.exit(0 if passed == len(results) else 1)