ducoterra
dbf2af26c6
Standardize on the .agents/ directory (shared with project skills): phases/, user_stories/, reports/, screenshots/, validate.sh, and phase-sessions/ + pipeline.log all move to .agents/ (git mv preserves history; runtime artifacts move alongside). Updates every reference in AGENTS.md, README.md, .gitignore, app docstrings, and test story headers. Historical KB content in data/ and the runtime pipeline.log transcript are left untouched.
335 lines
14 KiB
Python
335 lines
14 KiB
Python
"""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_without_tuning``
|
||
2. ``test_anonymous_sources_gated_viewer_open``
|
||
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``
|
||
"""
|
||
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_TITLE = "Kubernetes Homelab Cluster"
|
||
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))
|
||
|
||
|
||
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 works, the tuning UI is gone, Sign in is offered
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_anonymous_chat_without_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)
|
||
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()
|
||
|
||
# Chat still streams a grounded answer (with source chips) for
|
||
# anonymous visitors…
|
||
_ask(page, QUESTION)
|
||
expect(page.locator(".msg.brain .source-chip", has_text="kubernetes.md")).to_have_count(1)
|
||
|
||
# …but 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 (the phase-14 restore path) must not bring it back.
|
||
page.reload()
|
||
expect(page.locator(".msg.brain .bubble").last).to_contain_text(MOCK_ANSWER_MARKER)
|
||
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, the document viewer stays open (soft rule)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_anonymous_sources_gated_viewer_open(
|
||
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}"
|
||
|
||
# The soft rule: any seeded document still opens by direct URL.
|
||
page.goto(app_url + DOC_VIEWER_URL)
|
||
expect(page.locator("#doc-title")).to_have_text(DOC_TITLE, timeout=15_000)
|
||
expect(page.locator("#doc-content")).not_to_be_empty()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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("–")
|
||
expect(page.locator("#docs-table")).to_be_visible()
|
||
expect(page.locator("#docs-tbody tr")).to_have_count(13)
|
||
|
||
# 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)
|