brain-of-reese/tests/e2e/test_admin_auth.py

"""Phase 16 E2E (Playwright): single-admin sign-in (A10 revised).

Story: ``.agents/user_stories/admin-auth.md``
Run in isolation (DB must be up: ``podman compose up -d db``):

    uv run pytest tests/e2e/test_admin_auth.py -v --no-cov

The E2E app server boots with ``BOR_ADMIN_PASSWORD``/``BOR_SESSION_SECRET``
set (``tests/e2e/conftest.py``); the shared ``tests/e2e/auth_helpers.py::login``
performs the real form login on /login.html.

Test → story mapping (Playwright Mapping Rule):
1. ``test_anonymous_chat_gated_no_tuning``
2. ``test_anonymous_sources_and_viewer_gated``
3. ``test_login_wrong_password_shows_error``
4. ``test_admin_login_unlocks_sources_and_tuning``
5. ``test_logout_returns_to_anonymous``
6. ``test_login_page_a11y``

Phase 79 (API tokens): the anonymous pins moved to the gated contract —
``POST /api/chat`` and ``GET /api/documents/content`` are
``require_user`` (401 ``authentication required`` for anonymous; the
phase-16 "the viewer stays open" soft rule is SUPERSEDED, shared chats
are the only open surface). The password sign-in / sign-out /
wrong-password assertions are UNCHANGED.

Phase 97 adaptation: the admin catalog pin (test 4) is re-pointed at
the DRILL-DOWN TREE — the top level lists the fixture's single
indexed-only source (``docs``), the flat 13-row tbody no longer
renders, so the total is re-asserted PER LEVEL (drill → count; the
stat cards carry the KB total). The asserted document behavior is
unchanged; navigation only.
"""
from __future__ import annotations

import asyncio
from pathlib import Path
from threading import Thread
from typing import Any

from playwright.sync_api import Page, expect
from sqlalchemy import text

from app.config import Settings
from app.db import SessionLocal
from app.rag.importer import ImportSummary, import_sources
from app.rag.llm import LLMClient
from e2e.auth_helpers import ADMIN_PASSWORD, login

REPO = Path(__file__).resolve().parents[2]
FIXTURES = REPO / "tests" / "fixtures" / "docs"
QUESTION = "How is my Kubernetes cluster set up?"
MOCK_ANSWER_MARKER = "Deterministic mock answer for E2E"
DOC_VIEWER_URL = "/document.html?source=docs&path=homelab%2Fkubernetes.md"


async def _import_fixtures(mock_port: int) -> ImportSummary:
    kwargs: dict[str, Any] = {"_env_file": None, "llm_base_url": f"http://127.0.0.1:{mock_port}/v1"}
    settings = Settings(**kwargs)  # pyright: ignore[reportCallIssue]
    return await import_sources([FIXTURES], LLMClient(settings))


def _run_in_thread(coro: Any) -> Any:
    """Run a coroutine on a worker thread (Playwright owns the test loop)."""
    box: dict[str, Any] = {}

    def runner() -> None:
        try:
            box["value"] = asyncio.run(coro)
        except BaseException as e:  # noqa: BLE001 — re-raised on the test thread
            box["error"] = e

    t = Thread(target=runner)
    t.start()
    t.join()
    if "error" in box:
        raise box["error"]
    return box["value"]


def _reset_db(mock_port: int, seed: bool) -> ImportSummary | None:
    """Truncate the KB (and query log + steering notes), optionally re-seed."""
    with SessionLocal() as db:
        db.execute(text("TRUNCATE chunks, documents, query_log, steering_notes"))
        db.commit()
    if not seed:
        return None
    return _run_in_thread(_import_fixtures(mock_port))


# ---------------------------------------------------------------------------
# Phase 97: the catalog is the drill-down tree the agent's `ls` walks —
# top level = the sources (the fixture's single indexed-only source,
# `docs`), then one folder link per path segment. Small per-suite
# drill helpers (the test_kb_tree house pattern).
# ---------------------------------------------------------------------------


def _drill(page: Page, *names: str) -> None:
    """Drill one level at a time: each name is the EXACT text of the
    source/folder link at the current level (no URL change; since phase
    167 the first open of a level the depth-1 payload only NAMES costs
    one /api/docs/tree/node fetch, a level already held costs none)."""
    for name in names:
        page.click(f'#folders-tbody a.folder-link:text-is("{name}")')


def _go_top(page: Page) -> None:
    """Back to the top level: the breadcrumb's top-level link (call
    between drills only — the breadcrumb is hidden at the top)."""
    page.locator("#kb-crumb a.kb-crumb-link").first.click()


def _ask(page: Page, question: str) -> None:
    """Send one turn and wait until the grounded answer has fully landed."""
    page.fill("#message-input", question)
    page.click("#send-btn")
    expect(page.locator(".msg.user .bubble").last).to_contain_text(question)
    expect(page.locator(".msg.brain .bubble").last).to_contain_text(
        MOCK_ANSWER_MARKER, timeout=30_000
    )
    expect(page.locator("#send-btn")).to_be_enabled()
    expect(page.locator("#send-label")).to_have_text("Send")


# ---------------------------------------------------------------------------
# 1. Anonymous: chat is GATED (phase 79), the tuning UI is gone, Sign in
#    is offered
# ---------------------------------------------------------------------------


def test_anonymous_chat_gated_no_tuning(page: Page, app_url: str, db_ready: None) -> None:
    page.set_default_timeout(30_000)
    page.goto(app_url)

    # Header: Sign in offered, Sign out not.
    expect(page.locator("#sign-in-link")).to_be_visible()
    # Phase 34 task 02: the shared header module rewrites the static
    # ?next= fallback to the CURRENT pathname ("return to where you
    # were") — on the chat page that is "/" (the markup keeps
    # ?next=/sources.html as the no-JS fallback only).
    expect(page.locator("#sign-in-link")).to_have_attribute("href", "/login.html?next=/")
    expect(page.locator("#sign-out-btn")).to_be_hidden()

    # Phase 79: the phase-16 "anonymous chat still streams" pin is
    # SUPERSEDED — POST /api/chat is require_user-gated and the
    # anonymous browser's own fetch gets the 401 contract (the in-app
    # token gate that locks this UI is task 05's surface; the API
    # contract is the stable half of the pin).
    anon_chat = page.evaluate(
        """async () => {
            const r = await fetch('/api/chat', {
                method: 'POST',
                headers: {'Content-Type': 'application/json'},
                body: JSON.stringify({message: 'hello?'}),
            });
            return {status: r.status, body: await r.json()};
        }"""
    )
    assert anon_chat["status"] == 401, anon_chat
    assert anon_chat["body"] == {"detail": "authentication required"}, anon_chat

    # …and the server agrees the visitor is anonymous.
    who = page.evaluate("() => fetch('/api/whoami').then((r) => r.json())")
    assert who == {"authenticated": False, "role": "anonymous"}

    # The tuning UI is completely gone: no Tune button (new or
    # restored), no Tuning toggle or panel in the DOM at all.
    expect(page.locator(".msg.brain .tune-btn")).to_have_count(0)
    expect(page.locator("#steering-toggle")).to_have_count(0)
    expect(page.locator("#steering-panel")).to_have_count(0)

    # A reload must not bring it back.
    page.reload()
    expect(page.locator(".msg.brain .tune-btn")).to_have_count(0)
    expect(page.locator("#steering-toggle")).to_have_count(0)
    expect(page.locator("#sign-in-link")).to_be_visible()


# ---------------------------------------------------------------------------
# 2. Anonymous: Sources gated AND the document viewer's DATA is gated
#    (phase 79 supersedes the phase-16 soft rule)
# ---------------------------------------------------------------------------


def test_anonymous_sources_and_viewer_gated(
    page: Page, app_url: str, mock_llm: int, db_ready: None
) -> None:
    _reset_db(mock_llm, seed=True)

    api_docs_calls: list[str] = []
    page.on(
        "request",
        lambda r: api_docs_calls.append(r.url) if "/api/docs" in r.url else None,
    )

    page.goto(f"{app_url}/sources.html")
    # The gate, with its sign-in link (≥44px) — not a redirect.
    gate = page.locator("#sources-gate")
    expect(gate).to_be_visible()
    expect(gate).to_contain_text("Sign in to view the full catalog")
    link = gate.locator("a[href='/login.html?next=/sources.html']")
    expect(link).to_have_count(1)
    box = link.bounding_box()
    assert box is not None and box["height"] >= 44

    # Stat cards + table hidden…
    expect(page.locator("#stat-cards")).to_be_hidden()
    expect(page.locator("#docs-table")).to_be_hidden()
    expect(page.locator("#sources-empty")).to_be_hidden()
    # …and NO /api/docs call was ever made.
    assert api_docs_calls == [], f"anonymous sources page called /api/docs: {api_docs_calls}"

    # Phase 79 (task 05): the phase-16 soft rule ("any seeded document
    # still opens by direct URL") is SUPERSEDED — the content endpoint
    # is require_user-gated. The page DOCUMENT still loads (anonymous
    # gets the HTML), but the GATED DATA does not: the API refuses
    # with 401, and the viewer shows the inline token gate
    # (#doc-auth-gate) instead of a content error — the content fetch
    # never runs, so no not-found card either.
    page.goto(app_url + DOC_VIEWER_URL)
    expect(page.locator("#doc-auth-gate")).to_be_visible(timeout=15_000)
    expect(page.locator("#doc-not-found")).to_be_hidden()
    expect(page.locator("#doc-title")).to_have_text("Loading…")
    anon_content = page.evaluate(
        """async () => {
            const r = await fetch(
                '/api/documents/content?source=docs&path=homelab%2Fkubernetes.md');
            return {status: r.status, body: await r.json()};
        }"""
    )
    assert anon_content["status"] == 401, anon_content
    assert anon_content["body"] == {"detail": "authentication required"}, anon_content


# ---------------------------------------------------------------------------
# 3. Wrong password → role=alert error, no redirect, still anonymous
# ---------------------------------------------------------------------------


def test_login_wrong_password_shows_error(page: Page, app_url: str, db_ready: None) -> None:
    _reset_db(mock_port=0, seed=False)
    page.set_default_timeout(30_000)

    login(page, app_url, password="definitely-not-the-password")

    error = page.locator("#login-error")
    expect(error).to_be_visible()
    assert error.get_attribute("role") == "alert"
    expect(error).not_to_be_empty()
    # No redirect happened…
    expect(page).to_have_url(app_url + "/login.html")
    # …and the server agrees: still anonymous, no session cookie set.
    who = page.evaluate("() => fetch('/api/whoami').then((r) => r.json())")
    assert who == {"authenticated": False, "role": "anonymous"}

    # The form stays usable: the correct password now succeeds.
    page.fill("#login-password", ADMIN_PASSWORD)
    page.click("#login-form button[type=submit]")
    expect(page).to_have_url(app_url + "/sources.html", timeout=30_000)


# ---------------------------------------------------------------------------
# 4. Correct password → Sources + tuning unlocked, Sign out offered
# ---------------------------------------------------------------------------


def test_admin_login_unlocks_sources_and_tuning(
    page: Page, app_url: str, mock_llm: int, db_ready: None
) -> None:
    _reset_db(mock_llm, seed=True)
    page.set_default_timeout(30_000)

    # Real form login (default password + next) lands on the catalog.
    login(page, app_url)
    expect(page).to_have_url(app_url + "/sources.html")
    expect(page.locator("#sources-gate")).to_be_hidden()
    expect(page.locator("#stat-docs")).to_have_text("13")  # phase 47: +quadlet/j2
    expect(page.locator("#stat-chunks")).not_to_have_text("–")
    # Phase 97: the catalog is the drill-down tree — the top level
    # lists the source (the catalog-rendered signal), the file table
    # is hidden there; the flat 13-row total is re-asserted PER LEVEL
    # (drill → count; the sum is 3 + 1 + 2 + 1 + 3 + 1 + 1 + 1 = 13).
    page.locator("#folders-tbody .folder-link").first.wait_for(state="visible")
    expect(page.locator("#docs-table")).to_be_hidden()
    _drill(page, "docs")
    expect(page.locator("#folders-tbody tr")).to_have_count(2)
    expect(page.locator("#docs-tbody tr")).to_have_count(0)  # no root-level files
    _drill(page, "homelab")
    expect(page.locator("#docs-table")).to_be_visible()
    expect(page.locator("#docs-tbody tr")).to_have_count(3)
    _go_top(page)
    _drill(page, "docs", "deployments")
    expect(page.locator("#docs-tbody tr")).to_have_count(1)
    for folder, count in (
        ("container_gitlab", 2),
        ("networking", 1),
        ("quadlet", 3),
        ("scripts", 1),
        ("ssh", 1),
        ("templates", 1),
    ):
        _go_top(page)
        _drill(page, "docs", "homelab", folder)
        expect(page.locator("#docs-tbody tr")).to_have_count(count)

    # Chat: the tuning UI is back — Sign out instead of Sign in, Tune
    # under the answer. The header toggle is NOT back: removed from the
    # navbar at owner request (2026-08-28), the panel still ships
    # hidden (note management lives on /tuning.html).
    page.goto(app_url)
    expect(page.locator("#sign-out-btn")).to_be_visible()
    expect(page.locator("#sign-in-link")).to_be_hidden()
    assert page.locator("#steering-toggle").count() == 0, (
        "the steering toggle was removed from the navbar (2026-08-28)"
    )
    expect(page.locator("#steering-panel")).to_be_hidden()

    _ask(page, QUESTION)
    tune = page.locator(".msg.brain .tune-btn").last
    expect(tune).to_be_visible()
    box = tune.bounding_box()
    assert box is not None and box["height"] >= 44

    # The API agrees: admin, and the gated endpoints answer now.
    who = page.evaluate("() => fetch('/api/whoami').then((r) => r.json())")
    assert who == {"authenticated": True, "role": "admin"}
    docs_status = page.evaluate("() => fetch('/api/docs').then((r) => r.status)")
    assert docs_status == 200


# ---------------------------------------------------------------------------
# 5. Sign out → anonymous again (gate back, tuning gone, restore untunable)
# ---------------------------------------------------------------------------


def test_logout_returns_to_anonymous(
    page: Page, app_url: str, mock_llm: int, db_ready: None
) -> None:
    _reset_db(mock_llm, seed=True)
    page.set_default_timeout(30_000)

    login(page, app_url, next="/")  # straight into the chat
    expect(page).to_have_url(app_url + "/")
    expect(page.locator("#sign-out-btn")).to_be_visible()
    # The steering toggle was removed from the navbar (2026-08-28) —
    # absent for the admin too; the panel still ships hidden.
    assert page.locator("#steering-toggle").count() == 0
    expect(page.locator("#steering-panel")).to_be_hidden()

    # One grounded turn as admin (persisted to localStorage by phase 14).
    _ask(page, QUESTION)
    expect(page.locator(".msg.brain .tune-btn").last).to_be_visible()

    # Sign out: POST /api/logout + reload → anonymous again.
    page.click("#sign-out-btn")
    expect(page.locator("#sign-in-link")).to_be_visible(timeout=30_000)
    expect(page.locator("#sign-out-btn")).to_be_hidden()
    expect(page.locator("#steering-toggle")).to_have_count(0)
    expect(page.locator("#steering-panel")).to_have_count(0)

    # The restored conversation came back… without any Tune button.
    expect(page.locator(".msg.brain .bubble").last).to_contain_text(MOCK_ANSWER_MARKER)
    expect(page.locator(".msg.brain .tune-btn")).to_have_count(0)

    # The server agrees, and Sources is gated again.
    who = page.evaluate("() => fetch('/api/whoami').then((r) => r.json())")
    assert who == {"authenticated": False, "role": "anonymous"}
    page.goto(f"{app_url}/sources.html")
    expect(page.locator("#sources-gate")).to_be_visible()
    expect(page.locator("#docs-table")).to_be_hidden()


# ---------------------------------------------------------------------------
# 6. Login page accessibility (WCAG 2.1 AA basics)
# ---------------------------------------------------------------------------


def test_login_page_a11y(page: Page, app_url: str, db_ready: None) -> None:
    _reset_db(mock_port=0, seed=False)
    page.set_default_timeout(30_000)

    page.goto(f"{app_url}/login.html")

    # Standard app frame: landmarks + skip link, no CDN tags.
    expect(page.locator("header.app-header")).to_have_count(1)
    expect(page.locator("nav[aria-label='Primary']")).to_have_count(1)
    expect(page.locator("main#main")).to_have_count(1)
    expect(page.locator("footer.app-footer")).to_have_count(1)
    expect(page.locator(".skip-link")).to_have_count(1)
    html = page.content()
    assert 'src="https://' not in html and 'href="https://' not in html

    # The password field is labeled (visually-hidden <label for=…>).
    pw = page.get_by_label("Admin password")
    expect(pw).to_have_count(1)
    expect(pw.first).to_have_attribute("type", "password")
    expect(pw.first).to_have_attribute("autocomplete", "current-password")

    # Touch targets ≥44px (field + submit).
    for el in (pw.first, page.locator("#login-form button[type=submit]")):
        box = el.bounding_box()
        assert box is not None and box["height"] >= 44, f"target too small: {box}"

    # Keyboard focus draws the 3px focus-visible outline.
    page.focus("#login-password")
    outline = page.evaluate(
        "() => getComputedStyle(document.querySelector('#login-password')).outlineWidth"
    )
    assert outline == "3px", f"focus-visible outline missing: {outline!r}"

    # Errors are announced through the role=alert region.
    error = page.locator("#login-error")
    assert error.get_attribute("role") == "alert"
    expect(error).to_be_hidden()
    page.fill("#login-password", "wrong")
    page.click("#login-form button[type=submit]")
    expect(error).to_be_visible(timeout=15_000)

    # A signed-in visit to /login.html?next=/ redirects immediately.
    page.fill("#login-password", ADMIN_PASSWORD)
    page.click("#login-form button[type=submit]")
    expect(page).to_have_url(app_url + "/sources.html", timeout=30_000)
    page.goto(f"{app_url}/login.html?next=/")
    expect(page).to_have_url(app_url + "/", timeout=30_000)