"""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 (client-side — no fetch, no URL change).""" 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