scripts/smoke_backend_brain.py

#!/usr/bin/env python3
"""
DocDoe Backend Brain — Smoke Test Script

Runs a full end-to-end smoke test against a running DocDoe backend.

Usage:
    python scripts/smoke_backend_brain.py

Environment variables:
    DOCDOE_API_BASE_URL   Override base URL (default: http://localhost:8000)
    DOCDOE_AUTH_TOKEN     Bearer token (required if AUTH_ENABLED=true)
"""
from __future__ import annotations

import json
import os
import sys
import time
from typing import Any

try:
    import requests
except ImportError:
    print("ERROR: 'requests' library not found.")
    print("Install it with: pip install requests")
    sys.exit(1)


# ── Config ──────────────────────────────────────────────────────────────────

BASE_URL = os.environ.get("DOCDOE_API_BASE_URL", "http://localhost:8000").rstrip("/")
AUTH_TOKEN = os.environ.get("DOCDOE_AUTH_TOKEN", "")

HEADERS: dict[str, str] = {"Content-Type": "application/json"}
if AUTH_TOKEN:
    HEADERS["Authorization"] = f"Bearer {AUTH_TOKEN}"


# ── Helpers ──────────────────────────────────────────────────────────────────

PASS = "\033[92mPASS\033[0m"
FAIL = "\033[91mFAIL\033[0m"
INFO = "\033[94mINFO\033[0m"

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


def _print_step(label: str, passed: bool, detail: str = "") -> None:
    status = PASS if passed else FAIL
    print(f"  [{status}] {label}")
    if detail:
        print(f"         {detail}")


def run(
    label: str,
    method: str,
    path: str,
    body: dict[str, Any] | None = None,
    *,
    expect_status: int = 200,
    required_keys: list[str] | None = None,
) -> tuple[bool, dict[str, Any]]:
    """Run one API call, print result, return (passed, body)."""
    url = f"{BASE_URL}{path}"
    try:
        if method.upper() == "GET":
            resp = requests.get(url, headers=HEADERS, timeout=30)
        elif method.upper() == "POST":
            resp = requests.post(url, headers=HEADERS, json=body or {}, timeout=30)
        elif method.upper() == "PATCH":
            resp = requests.patch(url, headers=HEADERS, json=body or {}, timeout=30)
        elif method.upper() == "DELETE":
            resp = requests.delete(url, headers=HEADERS, timeout=30)
        else:
            raise ValueError(f"Unknown method: {method}")
    except requests.ConnectionError:
        print(f"\n  [{FAIL}] {label}")
        print()
        print("  ─────────────────────────────────────────────────────")
        print(f"  Backend is not running at {BASE_URL}")
        print()
        print("  Start it with:")
        print("    cd backend && uvicorn app.main:app --reload")
        print()
        if AUTH_TOKEN:
            print("  Auth token is set. If you get 401 errors, verify the token.")
        else:
            print("  If AUTH_ENABLED=true is set, export DOCDOE_AUTH_TOKEN=<your_token>")
        print("  ─────────────────────────────────────────────────────")
        print()
        sys.exit(1)
    except requests.Timeout:
        _print_step(label, False, "Request timed out after 30 seconds.")
        _results.append((label, False, "timeout"))
        return False, {}
    except Exception as exc:  # noqa: BLE001
        _print_step(label, False, str(exc))
        _results.append((label, False, str(exc)))
        return False, {}

    passed = resp.status_code == expect_status

    try:
        data: dict[str, Any] = resp.json()
    except Exception:  # noqa: BLE001
        data = {}

    if passed and required_keys:
        for key in required_keys:
            if key not in data:
                passed = False
                _print_step(label, False, f"Missing key '{key}' in response")
                _results.append((label, False, f"missing key: {key}"))
                return False, data

    detail = ""
    if not passed:
        detail = f"status={resp.status_code} expected={expect_status} body={str(data)[:120]}"

    _print_step(label, passed, detail)
    _results.append((label, passed, detail))
    return passed, data


# ── Smoke Steps ──────────────────────────────────────────────────────────────

def main() -> int:
    print()
    print("  ==========================================================")
    print("  DocDoe Backend Brain -- Smoke Test")
    print(f"  Target: {BASE_URL}")
    if AUTH_TOKEN:
        print(f"  Auth: Bearer token present ({len(AUTH_TOKEN)} chars)")
    else:
        print("  Auth: No token -- assuming AUTH_ENABLED=false (demo user)")
    print("  ==========================================================")
    print()

    source_id: str | None = None

    # 1. Health
    print("── 1. Health ──")
    ok, body = run("Health check", "GET", "/health", required_keys=["status"])
    if not ok:
        print(f"  {FAIL} Health check failed — backend may not be running.")
        sys.exit(1)
    print(f"       version={body.get('version', 'unknown')}")
    print()

    # 2. Analyze request
    print("── 2. Study Request Analyzer ──")
    ok, body = run(
        "Analyze: 'Kerala +2 Physics EMI exam tomorrow I need A+'",
        "POST",
        "/study/analyze-request",
        {
            "raw_text": "Kerala +2 Physics electromagnetic induction exam tomorrow, I need A+",
        },
        required_keys=["exam", "subject", "confidence"],
    )
    if ok:
        print(f"       exam={body.get('exam')} subject={body.get('subject')} "
              f"topic={body.get('topic')} confidence={body.get('confidence')}")
    print()

    # 3. Create study profile
    print("── 3. Study Profile ──")
    ok, profile = run(
        "Create/replace study profile",
        "POST",
        "/study-profile",
        {
            "exam": "Kerala +2",
            "board": "Kerala HSE",
            "subject": "Physics",
            "topic": "Electromagnetic Induction",
            "goal": "A+",
            "time_left": "5_hours",
            "level": "intermediate",
            "language_preference": "Malayalam + English",
            "primary_need": "notes",
        },
        expect_status=201,
        required_keys=["id", "subject", "topic"],
    )
    run("Get study profile (me)", "GET", "/study-profile/me", required_keys=["id"])
    run("Patch study profile", "PATCH", "/study-profile/me", {"level": "advanced"},
        required_keys=["level"])
    print()

    # 4. Create syllabus text source
    print("── 4. Sources ──")
    ok, source = run(
        "Create text source (syllabus)",
        "POST",
        "/sources/text",
        {
            "title": "Kerala +2 Physics Chapter 6 — Electromagnetic Induction",
            "text": (
                "Electromagnetic induction is the process of inducing EMF by changing magnetic flux. "
                "Faraday's law: e = -N dΦ/dt. "
                "Lenz's law states the induced current opposes the cause. "
                "Self-inductance L: e = -L dI/dt. "
                "Mutual inductance M: EMF in coil 2 due to change in coil 1. "
                "Transformer: Vs/Vp = Ns/Np. Step-up: Ns > Np. Step-down: Ns < Np."
            ),
            "source_type": "syllabus_text",
            "subject": "Physics",
            "chapter": "Electromagnetic Induction",
            "syllabus": "Kerala HSE",
        },
        expect_status=201,
        required_keys=["id", "status", "chunk_count"],
    )
    if ok:
        source_id = source.get("id")
        print(f"       source_id={source_id} chunks={source.get('chunk_count')}")
    print()

    # 5. List sources
    print("── 5. List Sources ──")
    ok, lst = run("List sources", "GET", "/sources", required_keys=["sources"])
    if ok:
        print(f"       total sources={len(lst.get('sources', []))}")
    print()

    # 6. Retrieve from source
    print("── 6. Source Retrieval ──")
    if source_id:
        run(
            "Retrieve chunks from source",
            "POST",
            f"/sources/{source_id}/retrieve",
            {"query": "Faraday's law of induction", "limit": 3},
            required_keys=["chunks"],
        )
    else:
        print(f"  [{INFO}] Skipping retrieval — no source_id available")
    print()

    # 7. Generate study path
    print("── 7. Study Path ──")
    sp_payload: dict[str, Any] = {
        "raw_text": "Kerala +2 Physics electromagnetic induction exam tomorrow",
        "use_my_profile": True,
    }
    if source_id:
        sp_payload["source_id"] = source_id
    ok, sp = run(
        "Generate study path",
        "POST",
        "/study-path/generate",
        sp_payload,
        required_keys=["title", "study_timeline", "readiness_score", "trust_notes"],
    )
    if ok:
        print(f"       readiness={sp.get('readiness_score')} "
              f"steps={len(sp.get('study_timeline', []))} "
              f"trust_notes={len(sp.get('trust_notes', []))}")
    print()

    # 8. Ask DocDoe without source
    print("── 8. Ask DocDoe ──")
    run(
        "Ask without source (generic trust note expected)",
        "POST",
        "/ask",
        {"question": "Explain Faraday's law in simple terms", "mode": "explain_simple"},
        required_keys=["answer", "trust_note"],
    )
    if source_id:
        run(
            "Ask with source",
            "POST",
            "/ask",
            {
                "question": "What is Lenz's law?",
                "mode": "exam_answer",
                "source_id": source_id,
            },
            required_keys=["answer", "sections"],
        )
    print()

    # 9-13. Studio endpoints
    print("── 9. Studio Generation ──")
    studio_payload = {
        "topic": "Electromagnetic Induction",
        "subject": "Physics",
        "language_preference": "English",
        "level": "intermediate",
        "goal": "A+",
    }
    if source_id:
        studio_payload["source_id"] = source_id

    run("Generate notes", "POST", "/generate/notes",
        {**studio_payload, "options": {"mode": "smart_notes"}},
        required_keys=["type", "output", "model_used"])

    run("Generate quiz", "POST", "/generate/quiz",
        {**studio_payload, "options": {"difficulty": "exam", "question_count": 5}},
        required_keys=["type", "output"])

    run("Generate flashcards", "POST", "/generate/flashcards",
        {**studio_payload, "options": {"card_count": 8}},
        required_keys=["type", "output"])

    run("Generate exam answer", "POST", "/generate/exam-answer",
        studio_payload, required_keys=["type", "output"])

    run("Generate last-night plan", "POST", "/generate/last-night-plan",
        {**studio_payload, "time_left": "5_hours"},
        required_keys=["type", "output"])
    print()

    # 14. PYQ analyze
    print("── 10. PYQ Analysis ──")
    ok, pyq = run(
        "PYQ analyze (expect unavailable)",
        "POST",
        "/pyq/analyze",
        {"subject": "Physics", "board": "Kerala HSE", "exam": "Kerala +2",
         "topic": "Electromagnetic Induction"},
        required_keys=["available", "trust_note"],
    )
    if ok:
        avail = pyq.get("available", True)
        trust = pyq.get("trust_note", "")
        if avail is False:
            _print_step("PYQ available=false confirmed", True)
        else:
            _print_step("PYQ available=false confirmed", False, "WARNING: available=true without real PYQ data!")
        if trust:
            print(f"       trust_note: {trust[:100]}")
    print()

    # 15-17. Video plans
    print("── 11. Video Plans ──")
    ok, qc = run(
        "Video plan: quick_concept",
        "POST",
        "/video-generator/plan",
        {"topic": "Electromagnetic Induction", "video_mode": "quick_concept",
         "teaching_style": "normal_teacher", "language_preference": "English"},
        required_keys=["target_duration_range", "render_supported", "planning_only", "scenes"],
    )
    if ok:
        print(f"       duration_range={qc.get('target_duration_range')} "
              f"render_supported={qc.get('render_supported')}")

    ok, ef = run(
        "Video plan: exam_focus",
        "POST",
        "/video-generator/plan",
        {"topic": "Electromagnetic Induction", "video_mode": "exam_focus",
         "teaching_style": "normal_teacher", "language_preference": "English"},
        required_keys=["target_duration_range", "render_supported"],
    )
    if ok:
        print(f"       duration_range={ef.get('target_duration_range')}")

    ok, dm = run(
        "Video plan: deep_masterclass (planning_only=true expected)",
        "POST",
        "/video-generator/plan",
        {"topic": "Electromagnetic Induction", "video_mode": "deep_masterclass",
         "teaching_style": "visual_tutor", "language_preference": "English"},
        required_keys=["planning_only", "render_supported"],
    )
    if ok:
        if dm.get("planning_only") is True and dm.get("render_supported") is False:
            _print_step("planning_only=true confirmed", True)
        else:
            _print_step("planning_only=true confirmed", False,
                        f"planning_only={dm.get('planning_only')} render_supported={dm.get('render_supported')}")
    print()

    # 18-20. Billing
    print("── 12. Billing ──")
    ok, plan = run("Get billing plan", "GET", "/billing/me",
                   required_keys=["selected_plan", "monthly_video_limit", "coming_soon"])
    if ok:
        print(f"       plan={plan.get('selected_plan')} "
              f"videos={plan.get('monthly_video_used')}/{plan.get('monthly_video_limit')} "
              f"gen={plan.get('monthly_generation_used')}/{plan.get('monthly_generation_limit')}")

    ok, sel = run("Select popular_299", "POST", "/billing/select-plan", {"plan": "popular_299"},
                  required_keys=["selected_plan", "monthly_video_limit"])
    if ok:
        assert sel.get("selected_plan") == "popular_299", "Plan mismatch"
        assert sel.get("monthly_video_limit") == 20, "Limit mismatch"

    ok, trial = run("Start trial", "POST", "/billing/start-trial",
                    required_keys=["status", "trial_ends_at"])
    if ok:
        if trial.get("status") == "trialing":
            _print_step("Trial status=trialing confirmed", True)
        else:
            _print_step("Trial status=trialing confirmed", False,
                        f"status={trial.get('status')}")
    print()

    # ── Summary ───────────────────────────────────────────────────────────────

    total = len(_results)
    passed_count = sum(1 for _, ok, _ in _results if ok)
    failed_count = total - passed_count

    print("  ==========================================================")
    print(f"  Results: {passed_count}/{total} passed")
    if failed_count:
        print(f"  {FAIL}  {failed_count} step(s) failed:")
        for label, ok, detail in _results:
            if not ok:
                print(f"    [X] {label}")
                if detail:
                    print(f"      {detail}")
    else:
        print(f"  {PASS}  All checks passed!")
    print("  ==========================================================")
    print()

    return 0 if failed_count == 0 else 1


if __name__ == "__main__":
    sys.exit(main())