tests/settings_test.py

"""
Settings smoke tests for DemoPrep.

These tests verify that a user setting actually flows through the pipeline
and produces the expected output. Each test:
  1. Reads the current setting value (baseline)
  2. Writes the test value directly to Supabase
  3. Runs the pipeline with a fresh browser session (picks up new value)
  4. Verifies the output
  5. Resets the setting back (always — even on failure)

These are separate from e2e_quality.py which tests output quality.
These test that settings are wired correctly end-to-end.

Usage:
    source demoprep/bin/activate
    python tests/settings_test.py                   # run all
    python tests/settings_test.py --test fact_10k   # run one test
"""

import argparse
import json
import os
import sys
import time
from datetime import datetime
from pathlib import Path
from typing import Optional

import yaml
from dotenv import load_dotenv
from playwright.sync_api import Page, sync_playwright

sys.path.insert(0, str(Path(__file__).parent.parent))
load_dotenv(Path(__file__).parent.parent / ".env")

BASE_URL      = os.getenv("TEST_TARGET_URL")
if not BASE_URL:
    raise ValueError("TEST_TARGET_URL not set in .env")
TEST_USER     = os.getenv("TEST_USER")
TEST_PASSWORD = os.getenv("TEST_PASSWORD")

RESULTS_DIR = Path(__file__).parent / "quality_results"
RESULTS_DIR.mkdir(exist_ok=True)

# Fixed test cases (from quality_config.yaml)
FIXED_TESTS = [
    {
        "name": "Nike — Retail Sales",
        "company": "Nike",
        "company_url": "nike.com",
        "vertical": "Retail & Consumer Goods",
        "line": "Fashion/Apparel",
        "function": "Sales",
        "ts_environment": "secloud - primary",
    },
    {
        "name": "Wells Fargo — Banking Marketing",
        "company": "Wells Fargo",
        "company_url": "wellsfargo.com",
        "vertical": "Financial Services",
        "line": "Banking",
        "function": "Marketing",
        "ts_environment": "secloud - primary",
    },
]


# ---------------------------------------------------------------------------
# Supabase settings helpers
# ---------------------------------------------------------------------------
def read_setting(key: str) -> str:
    from supabase_client import SupabaseSettings
    s = SupabaseSettings()
    settings = s.load_all_settings(TEST_USER)
    return settings.get(key, "")


def write_setting(key: str, value: str):
    from supabase_client import SupabaseSettings
    s = SupabaseSettings()
    s.save_all_settings(TEST_USER, {key: value})


# ---------------------------------------------------------------------------
# Snowflake verification
# ---------------------------------------------------------------------------
def check_snowflake_rows(schema_name: str) -> dict:
    """Return {table: row_count} for every table in the schema."""
    try:
        from snowflake_auth import get_snowflake_connection, get_demo_database
        conn   = get_snowflake_connection()
        cursor = conn.cursor()
        database = get_demo_database()  # demos live in the rotating <base>_<YYYY_MM> DB
        cursor.execute(f'SHOW TABLES IN SCHEMA "{database}"."{schema_name}"')
        tables = [row[1] for row in cursor.fetchall()]
        counts = {}
        for t in tables:
            cursor.execute(f'SELECT COUNT(*) FROM "{database}"."{schema_name}"."{t}"')
            counts[t] = cursor.fetchone()[0]
        cursor.close()
        conn.close()
        return counts
    except Exception as e:
        return {"error": str(e)}


def get_schema_from_logs(start_time: float) -> Optional[str]:
    """Pull the schema name written to session_logs during this run."""
    try:
        from supabase_client import SupabaseSettings
        from datetime import datetime, timezone
        start_iso = datetime.fromtimestamp(start_time, tz=timezone.utc).isoformat()
        s = SupabaseSettings()
        result = (
            s.client.table("session_logs")
            .select("*")
            .eq("user_email", TEST_USER)
            .gte("ts", start_iso)
            .order("ts", desc=False)
            .execute()
        )
        for log in (result.data or []):
            meta = log.get("meta") or {}
            if "schema" in meta:
                return meta["schema"]
    except Exception:
        pass
    return None


# ---------------------------------------------------------------------------
# Browser helpers (shared with e2e_quality.py pattern)
# ---------------------------------------------------------------------------
def _do_login(page: Page):
    page.fill('input[type=text]', TEST_USER)
    page.fill('input[type=password]', TEST_PASSWORD)
    page.click('button:has-text("Login")')
    page.wait_for_selector('button[role=tab]', timeout=90000)
    page.wait_for_timeout(3000)


def select_gradio_dropdown(page: Page, label: str, value: str):
    inp = page.locator(f'input[aria-label="{label}"]').first
    inp.click(timeout=5000)
    page.wait_for_timeout(300)
    page.get_by_role('option', name=value, exact=True).click(timeout=5000)
    page.wait_for_timeout(300)


def open_settings_accordion(page: Page):
    """Open the ⚙️ Settings accordion in the right panel if not already open."""
    try:
        accordion = page.locator('button:has-text("⚙️ Settings")').first
        # Check if accordion is collapsed (aria-expanded=false)
        if accordion.get_attribute('aria-expanded') == 'false':
            accordion.click(timeout=5000)
            page.wait_for_timeout(500)
    except Exception:
        pass


def submit_and_wait(page: Page, test_case: dict, timeout_min: int = 60) -> dict:
    """
    Submit the form and wait for pipeline completion.
    Sets panel values directly — no Supabase write needed.
    Returns {"stages": {...}, "timed_out": bool, "error": str|None}
    """
    # Fresh navigation — ensures panel loads with current Supabase defaults
    page.goto(BASE_URL, timeout=90000)
    page.wait_for_selector('button[role=tab], input[type=password]', timeout=90000)
    if page.locator('input[type=password]').is_visible(timeout=2000):
        _do_login(page)

    page.wait_for_timeout(2000)
    page.get_by_role('tab', name='App', exact=True).click(timeout=10000)
    page.wait_for_timeout(1000)

    select_gradio_dropdown(page, "TS Environment", test_case.get("ts_environment", "secloud - primary"))
    select_gradio_dropdown(page, "Vertical",       test_case["vertical"])
    select_gradio_dropdown(page, "Line",           test_case["line"])
    select_gradio_dropdown(page, "Function",       test_case["function"])

    url_el = page.locator('input[aria-label="Company URL"], textarea[placeholder="e.g. Amazon.com"]').first
    url_el.click(click_count=3, timeout=5000)
    url_el.fill(test_case["company_url"])
    page.wait_for_timeout(300)

    # Open Settings accordion and set any overrides from test_case
    open_settings_accordion(page)

    if test_case.get("data_size"):
        select_gradio_dropdown(page, "Data Size", test_case["data_size"])

    # Liveboard name
    try:
        lb_el = page.locator('input[aria-label="Liveboard Name"], textarea[placeholder="Auto from company URL if blank"]').first
        lb_el.scroll_into_view_if_needed(timeout=3000)
        lb_el.click(click_count=3, timeout=3000)
        lb_el.fill(f"Settings Test — {test_case['company']}")
        page.wait_for_timeout(300)
    except Exception:
        pass

    page.click('button:has-text("→ GO")', timeout=10000)
    print(f"    ✅ Submitted: {test_case['name']}")

    # Poll for completion
    timeout_sec = timeout_min * 60
    start       = time.time()
    poll        = 15
    last_stages = {}

    while time.time() - start < timeout_sec:
        time.sleep(poll)
        text   = page.inner_text('body')
        stages = {}
        for key, label in [("research","Research"),("ddl","DDL"),("data","Data"),
                           ("thoughtspot","ThoughtSpot"),("complete","Complete")]:
            if f"✓ {label}" in text or f"✅ {label}" in text:
                stages[key] = "complete"
            elif f"▶ {label}" in text:
                stages[key] = "running"
            else:
                stages[key] = "unknown"

        if stages != last_stages:
            done    = [k for k, v in stages.items() if v == "complete"]
            running = [k for k, v in stages.items() if v == "running"]
            print(f"      ✓ {done}  ▶ {running}")
            last_stages = stages

        main = ("research", "ddl", "data", "thoughtspot")
        if stages.get("complete") == "complete" or all(stages.get(s) == "complete" for s in main):
            print("    ✅ Pipeline complete")
            return {"stages": stages, "timed_out": False, "error": None,
                    "start_time": start}
    else:
        print(f"    ⏰ Timed out after {timeout_min} min")
        return {"stages": last_stages, "timed_out": True, "error": None,
                "start_time": start}


# ---------------------------------------------------------------------------
# Test: fact_table_size = 10,000
# ---------------------------------------------------------------------------
def test_fact_10k():
    """
    Select Data Size = Large (10k rows) directly in the panel, run both fixed
    test cases, and verify ~10k rows in the fact table via Snowflake.
    No Supabase write needed — values come from the panel at run time.
    """
    print("\n" + "="*62)
    print("  Settings Test: Data Size = Large (10,000 rows)")
    print(f"  Target: {BASE_URL}")
    print("="*62)

    # Inject data_size override into each test case
    test_cases = [{**tc, "data_size": "Large"} for tc in FIXED_TESTS]

    results = []

    with sync_playwright() as p:
        for tc in test_cases:
            print(f"─── {tc['name']} ───")
            browser = p.chromium.launch()
            ctx     = browser.new_context(viewport={"width": 1280, "height": 900})
            page    = ctx.new_page()

            print(f"  🔐 Logging in...")
            page.goto(BASE_URL, timeout=90000)
            page.wait_for_selector('input[type=password]', timeout=90000)
            _do_login(page)
            print("  ✅ Logged in")

            run = submit_and_wait(page, tc, timeout_min=90)
            ctx.close()
            browser.close()

            # Verify row counts
            schema = get_schema_from_logs(run["start_time"])
            result = {
                "name":      tc["name"],
                "timed_out": run["timed_out"],
                "schema":    schema,
                "pass":      False,
                "note":      "",
            }

            if run["timed_out"]:
                result["note"] = "timed out"
            elif not schema:
                result["note"] = "schema not found in session_logs"
            else:
                counts = check_snowflake_rows(schema)
                result["row_counts"] = counts
                if "error" in counts:
                    result["note"] = f"Snowflake error: {counts['error']}"
                else:
                    fact_rows = max(counts.values()) if counts else 0
                    expected  = 10000
                    tolerance = 0.10  # within 10%
                    passed    = fact_rows >= expected * (1 - tolerance)
                    result["fact_rows"]  = fact_rows
                    result["pass"]       = passed
                    result["note"]       = (
                        f"fact table: {fact_rows} rows "
                        f"({'✅ PASS' if passed else f'❌ FAIL — expected ~{expected}'})"
                    )
                    print(f"  📦 {schema}")
                    for t, c in sorted(counts.items()):
                        print(f"      {t}: {c} rows")

            results.append(result)
            icon = "✅" if result["pass"] else "❌"
            print(f"  {icon} {result['name']}: {result['note']}\n")

    # Summary
    print("="*62)
    passed = sum(1 for r in results if r["pass"])
    print(f"  RESULT: {passed}/{len(results)} passed")
    for r in results:
        icon = "✅" if r["pass"] else "❌"
        print(f"  {icon} {r['name']}: {r['note']}")
    print("="*62 + "\n")

    # Save results
    ts   = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
    path = RESULTS_DIR / f"{ts}_settings_fact10k.json"
    with open(path, "w") as f:
        json.dump({"timestamp": datetime.now().isoformat(),
                   "test": "fact_table_size_10k",
                   "target": BASE_URL,
                   "results": results}, f, indent=2, default=str)
    print(f"💾 Results: {path}")
    return all(r["pass"] for r in results)


# ---------------------------------------------------------------------------
# Registry + CLI
# ---------------------------------------------------------------------------
TESTS = {
    "fact_10k": test_fact_10k,
}


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--test", choices=list(TESTS.keys()),
                        help="Run a specific test (default: all)")
    args = parser.parse_args()

    if not TEST_USER or not TEST_PASSWORD:
        raise RuntimeError("TEST_USER and TEST_PASSWORD must be set in .env")

    to_run = [args.test] if args.test else list(TESTS.keys())
    passed = 0
    for name in to_run:
        ok = TESTS[name]()
        if ok:
            passed += 1
    print(f"\nTotal: {passed}/{len(to_run)} passed")
    sys.exit(0 if passed == len(to_run) else 1)


if __name__ == "__main__":
    main()