Experimental/brain-of-reese
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
70ba8710f36c4b40cac5b9533887a18b214b4234
brain-of-reese/tests/e2e/test_share_chat.py
T

528 lines
22 KiB
Python
Raw Blame History

"""Phase 51 E2E (Playwright): share a chat by link — anonymous view.
TODO.md L6 (owner 2026-08-29): "Need a way to share a chat with a link
so others can see it anonymously."
Run in isolation (DB must be up: ``podman compose up -d db``):
uv run pytest tests/e2e/test_share_chat.py -v --no-cov
The owner-locked loop under test (2026-08-29, roadmap confirmation;
phase 55 replaced the Save pill with auto-save — by the time the answer
settles the conversation is already saved, so the chat page's Share
click is the idempotent share of the linked row):
* **Share from the chat page** — the Share pill (admin-only, ships
hidden) on the AUTO-SAVED conversation shares the linked row via
``POST /api/chats/<id>/share`` — the row appears in ``GET
/api/chats`` with a non-null ``share_url`` of the shape
``/shared/<uuid4>`` (the unlinked save-then-share one-action wire
path — ``POST /api/chats`` with ``share: true`` — is pinned by the
integration suite, guest-reachable since phase 55 task 01); the
absolute URL is copied to the clipboard, with the inline-link
fallback on a non-secure origin (the assertion branches on
``navigator.clipboard`` availability);
* **Anonymous view** — a FRESH browser context (a separate session, no
cookies) opening ``/shared/<token>`` sees the full conversation
read-only through the same record shape: title = the auto-title,
user + brain bubbles (the same deterministic answer text the admin
session saw), the thinking block RESTORED COLLAPSED, the source
chips as PLAIN TEXT (zero ``a.source-chip`` — guests cannot open
documents, the documents API is admin-only), and ZERO interactive
controls anywhere (no composer, no Save/Share pills, no Tune/Retry,
no button chips); the nav's admin-only links stay hidden for a
guest;
* **Share from History + unshare** — the History row's Share column:
"Create link" → Copy + Unshare; Unshare is the inline two-step
(no native dialog); after Yes the cell returns to "Create link",
the SAME URL now shows the "invalid or was revoked" state in a
fresh anonymous context, and ``GET /api/chats/<id>`` no longer
carries ``share_url`` (the omission rule — the key is absent, not
null);
* **Bad token** — a well-formed but unknown token renders the invalid
state with no JS crash and the guest header (the page route serves
the HTML for any well-formed token; the client's 404 read drives
the invalid card).
DB isolation: the shared e2e Postgres keeps ``saved_chats`` rows
across suites, so every test here uses a DISTINCTIVE question text
(its auto-title is therefore unique), never asserts on absolute row
counts, and deletes the rows it creates in a ``finally`` (admin
cookie). The KB tables are truncated + re-seeded the house way
(deterministic mock embeddings); ``saved_chats`` is never touched by
the reset.
"""
from __future__ import annotations
import asyncio
import re
import time
from pathlib import Path
from threading import Thread
from typing import Any
import httpx
from playwright.sync_api import Browser, BrowserContext, 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 login
REPO = Path(__file__).resolve().parents[2]
FIXTURES = REPO / "tests" / "fixtures" / "docs"
MOCK_ANSWER_MARKER = "Deterministic mock answer for E2E"
SHARE_URL_RE = re.compile(r"^/shared/[0-9a-f-]{36}$")
#: The invalid/revoked card's line (frontend/shared.html, task 03).
INVALID_TEXT = "This share link is invalid or was revoked."
BAD_TOKEN = "00000000-0000-4000-8000-000000000000"
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's sync API keeps an asyncio loop running on the test
thread, so ``asyncio.run`` cannot be called directly from a test
body.
"""
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 — deterministic
mock answers), then optionally re-import fixtures. ``saved_chats``
is deliberately NOT touched: rows persist across suites and every
test here cleans up after itself."""
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 (the ``done`` event restored the Send button)."""
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")
def _admin_cookies(page: Page) -> dict[str, str]:
"""The signed session cookies the browser holds after a form login —
used to call the admin API with plain httpx (the test's API side
sees exactly what the signed-in browser sees)."""
return {
c["name"]: c["value"]
for c in page.context.cookies()
if "name" in c and "value" in c
}
def _chats(app_url: str, cookies: dict[str, str]) -> list[dict[str, Any]]:
r = httpx.get(f"{app_url}/api/chats", timeout=10, cookies=cookies)
assert r.status_code == 200
return r.json()["chats"]
def _auto_title(question: str) -> str:
"""The phase-50 auto-title convention: the first question,
whitespace-collapsed, capped at 120 chars."""
return " ".join(question.split())[:120]
def _find_row(
rows: list[dict[str, Any]], title: str
) -> dict[str, Any] | None:
return next((c for c in rows if c["title"] == title), None)
def _delete_chat(app_url: str, cookies: dict[str, str], chat_id: str) -> None:
"""Best-effort row cleanup (a 404 — already deleted — is fine)."""
httpx.delete(f"{app_url}/api/chats/{chat_id}", timeout=10, cookies=cookies)
def _wait_saved_row(
app_url: str,
cookies: dict[str, str],
title: str,
messages: int = 2,
) -> dict[str, Any]:
"""Wait for the auto-saved row (phase 55: auto-saves are SILENT —
A2 — so there is no status line to wait on). The upsert is
fire-and-forget from the UI's point of view, so poll the admin
list until the row with the conversation's auto-title appears with
the expected message count."""
deadline = time.monotonic() + 15
last: dict[str, Any] | None = None
while time.monotonic() < deadline:
last = _find_row(_chats(app_url, cookies), title)
if last is not None and last["message_count"] >= messages:
return last
time.sleep(0.2)
raise AssertionError(f"no auto-saved row for {title!r} (last: {last!r})")
def _grant_clipboard(page: Page, app_url: str) -> None:
"""Grant the async-clipboard permissions on the admin context.
``http://127.0.0.1`` is a secure context, so ``navigator.clipboard``
exists — but headless Chromium still requires the permission grant
before ``writeText`` resolves (without it the owner-locked
inline-link fallback fires). The assertion below branches on the
API's availability, so a non-secure origin still passes through
the fallback branch deterministically.
"""
page.context.grant_permissions(
["clipboard-read", "clipboard-write"], origin=app_url
)
def _click_share_and_assert_status(page: Page, app_url: str) -> None:
"""Press the chat page's Share pill and pin the owner-locked
outcome on the live region: "Share link copied." when
``navigator.clipboard`` is available in the context, else the
inline fallback link field carrying the ``/shared/<uuid>`` URL.
The branch is on the API's availability (task spec)."""
page.locator("#share-chat-btn").click()
if page.evaluate("() => !!navigator.clipboard"):
expect(page.locator("#send-status")).to_have_text(
"Share link copied.", timeout=15_000
)
expect(page.locator(".share-link-fallback")).to_have_count(0)
else:
expect(page.locator("#send-status")).to_have_text(
"Share link ready — copy it from the field.", timeout=15_000
)
field = page.locator(".share-link-fallback")
expect(field).to_be_visible()
href = field.get_attribute("href")
assert href is not None
assert href.startswith(app_url)
assert SHARE_URL_RE.fullmatch(href.removeprefix(app_url))
# ---------------------------------------------------------------------------
# 1. Share from the chat page: the AUTO-SAVED conversation is shared
# (the idempotent share of the linked row — phase 55); the API row
# carries the /shared/<uuid> link
# ---------------------------------------------------------------------------
def test_share_from_chat_page(
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="/")
expect(page).to_have_url(app_url + "/", timeout=30_000)
q = "How is my Kubernetes cluster set up? (share-chat)"
_ask(page, q)
# Admin: the Share pill is revealed (ships hidden, whoami reveals
# it). Phase 55: the conversation is AUTO-SAVED by the time the
# answer settles (the Save pill is gone) — the row exists under the
# auto-title, and the Share click below shares the linked row.
share = page.locator("#share-chat-btn")
expect(share).to_be_visible()
expect(share).to_have_attribute("aria-label", "Share chat")
cookies = _admin_cookies(page)
row = _wait_saved_row(app_url, cookies, _auto_title(q))
assert row["message_count"] == 2, "auto-saved with both messages before the Share click"
_grant_clipboard(page, app_url)
_click_share_and_assert_status(page, app_url)
created: str | None = None
try:
# The API agrees: the Share click shared the auto-saved row —
# it carries a non-null share_url of the token shape. (The
# unlinked save-then-share one-action path is pinned by the
# integration suite — from the chat page the conversation is
# always linked by the time there is anything to share.)
row = _find_row(_chats(app_url, cookies), _auto_title(q))
assert row is not None, "the auto-saved row must exist"
assert row["message_count"] == 2, "the saved conversation holds both messages"
share_url = row.get("share_url")
assert share_url is not None, "the share_url must be present (non-null)"
assert SHARE_URL_RE.fullmatch(share_url), f"bad share_url shape: {share_url}"
created = row["id"]
finally:
if created is not None:
_delete_chat(app_url, cookies, created)
# ---------------------------------------------------------------------------
# 2. The anonymous shared view: a FRESH context (no cookies) sees the
# full conversation read-only — thinking collapsed, plain-text chips,
# zero controls, the guest header
# ---------------------------------------------------------------------------
def test_anonymous_shared_view(
page: Page,
browser: Browser,
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="/")
expect(page).to_have_url(app_url + "/", timeout=30_000)
# "think out loud" → the mock streams reasoning first, so the saved
# record (and the shared view) carries a thinking block.
q = "think out loud — how is my kubernetes cluster set up? (share-view)"
_ask(page, q)
# The answer text the admin session saw — the shared view must
# render the SAME deterministic text (same record, same renderer).
answer = page.locator(".msg.brain .bubble").first.inner_text()
_grant_clipboard(page, app_url)
_click_share_and_assert_status(page, app_url)
cookies = _admin_cookies(page)
row = _find_row(_chats(app_url, cookies), _auto_title(q))
assert row is not None and row.get("share_url")
share_url: str = row["share_url"]
created: str | None = row["id"]
anon_ctx: BrowserContext | None = None
try:
# A FRESH context: a separate session with no cookies — the
# guest's only credential is the token in the URL.
anon_ctx = browser.new_context()
anon = anon_ctx.new_page()
anon.set_default_timeout(30_000)
anon.goto(app_url + share_url)
# Title = the auto-title (the shared chat's h1).
expect(anon.locator("#shared-title")).to_have_text(_auto_title(q))
# The conversation rendered: the user question bubble + the
# brain answer with the SAME deterministic text the admin saw.
expect(anon.locator(".msg.user .bubble")).to_have_count(1)
expect(anon.locator(".msg.user .bubble")).to_contain_text(q)
bubble = anon.locator(".msg.brain .bubble").first
expect(bubble).to_contain_text(MOCK_ANSWER_MARKER, timeout=30_000)
assert bubble.inner_text() == answer, "the shared answer must match the admin's"
# The thinking block exists and is RESTORED COLLAPSED (the
# phase-17 restore convention — a closed <details> has no
# `open` attribute).
think = anon.locator("details.thinking")
expect(think).to_have_count(1)
expect(think.first).not_to_have_attribute("open")
# Source chips are PLAIN TEXT: the on-topic turn carries its
# source chips (top-2 docs — the hybrid retrieval), but every
# one as a <span>: zero <a.source-chip> anywhere (a guest
# cannot open documents; the documents API is admin-only).
assert (
anon.locator(".msg.brain .source-chip").count() >= 1
), "the grounded turn must carry its source chips"
expect(
anon.locator(".msg.brain .source-chip", has_text="kubernetes.md")
).to_have_count(1)
expect(anon.locator("a.source-chip")).to_have_count(0)
# ZERO interactive controls anywhere in the conversation: no
# composer, no Save/Share pills, no Tune/Retry, and (were the
# turn deflected) the "Maybe try" chips would be spans — never
# buttons.
expect(anon.locator("#composer")).to_have_count(0)
expect(anon.locator("#save-chat-btn, #share-chat-btn")).to_have_count(0)
expect(anon.locator(".tune-btn")).to_have_count(0)
expect(anon.locator(".retry-btn")).to_have_count(0)
expect(anon.locator("button.suggestion-chip")).to_have_count(0)
# The nav's admin-only links stay hidden for a guest (and the
# Sign in link is what the guest gets instead).
for admin_link in ("#nav-sources", "#nav-git-sources", "#nav-tuning", "#nav-history"):
expect(anon.locator(admin_link)).to_be_hidden()
expect(anon.locator("#sign-in-link")).to_be_visible(timeout=15_000)
finally:
if anon_ctx is not None:
anon_ctx.close()
if created is not None:
_delete_chat(app_url, cookies, created)
# ---------------------------------------------------------------------------
# 3. Share from the History row + unshare: Create link → Copy/Unshare →
# the two-step confirm revokes — the same URL goes invalid and the
# API drops share_url
# ---------------------------------------------------------------------------
def test_share_from_history_and_unshare(
page: Page,
browser: Browser,
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="/")
expect(page).to_have_url(app_url + "/", timeout=30_000)
q = "How is my Kubernetes cluster set up? (share-history)"
_ask(page, q)
# Phase 55: the conversation is already AUTO-SAVED by the time the
# answer settles (no Save pill) — wait for the row via the admin
# list (this test drives the History column).
cookies = _admin_cookies(page)
_grant_clipboard(page, app_url)
row = _wait_saved_row(app_url, cookies, _auto_title(q))
chat_id: str = row["id"]
anon_ctx: BrowserContext | None = None
try:
# History: the row's Share cell ships in the unshared state —
# the single "Create link" button.
page.goto(app_url + "/history.html")
tr = page.locator(
"#history-tbody tr", has=page.locator(f"a[href='/?chat={chat_id}']")
)
create = tr.locator("button.history-share-create")
expect(create).to_be_visible(timeout=15_000)
expect(create).to_have_text("Create link")
# Create link → the cell re-renders to the shared state
# (Copy + Unshare) and the outcome lands on the live region
# (clipboard or the inline field — either success line).
create.click()
expect(tr.locator("button.history-share-copy")).to_be_visible(timeout=15_000)
expect(tr.locator("button.history-unshare")).to_be_visible()
expect(page.locator("#history-status")).to_contain_text("Share link")
# The API agrees: the row now carries the share link…
r = httpx.get(f"{app_url}/api/chats/{chat_id}", timeout=10, cookies=cookies)
assert r.status_code == 200
share_url = r.json().get("share_url")
assert share_url is not None
assert SHARE_URL_RE.fullmatch(share_url)
# …and the FRESH anonymous context renders the conversation at
# the URL.
anon_ctx = browser.new_context()
anon = anon_ctx.new_page()
anon.set_default_timeout(30_000)
anon.goto(app_url + share_url)
expect(anon.locator("#shared-title")).to_have_text(_auto_title(q))
expect(anon.locator(".msg.user .bubble")).to_contain_text(q)
expect(anon.locator(".msg.brain .bubble").first).to_contain_text(
MOCK_ANSWER_MARKER, timeout=30_000
)
anon_ctx.close()
anon_ctx = None
# Unshare: the inline two-step (the phase-50 confirm pattern —
# no native dialog; the pair appearing is the pinned contract).
tr.locator("button.history-unshare").click()
expect(tr.locator(".history-confirm-text")).to_have_text("Unshare?")
expect(tr.locator(".history-confirm-yes")).to_be_visible()
expect(tr.locator(".history-confirm-no")).to_be_visible()
tr.locator(".history-confirm-yes").click()
# The cell returns to the unshared state + the live region.
expect(tr.locator("button.history-share-create")).to_be_visible(timeout=15_000)
expect(tr.locator("button.history-share-copy")).to_have_count(0)
expect(page.locator("#history-status")).to_have_text(f'Unshared "{q}".')
# The SAME URL is revoked now: a fresh anonymous context sees
# the invalid state (no data rendered).
anon_ctx = browser.new_context()
anon = anon_ctx.new_page()
anon.set_default_timeout(30_000)
anon.goto(app_url + share_url)
expect(anon.locator("#shared-invalid")).to_be_visible(timeout=15_000)
expect(anon.locator("#shared-invalid")).to_contain_text(INVALID_TEXT)
expect(anon.locator(".msg")).to_have_count(0)
anon_ctx.close()
anon_ctx = None
# The API agrees: the row is unshared — share_url is ABSENT
# (the omission rule: no null in the wire shape).
r = httpx.get(f"{app_url}/api/chats/{chat_id}", timeout=10, cookies=cookies)
assert r.status_code == 200
assert "share_url" not in r.json(), "unshared → share_url must be absent"
finally:
if anon_ctx is not None:
anon_ctx.close()
_delete_chat(app_url, cookies, chat_id)
# ---------------------------------------------------------------------------
# 4. A well-formed but unknown token: the invalid state, no JS crash,
# the guest header renders
# ---------------------------------------------------------------------------
def test_bad_token_invalid_state(
page: Page, app_url: str, mock_llm: int, db_ready: None
) -> None:
# No DB reset / no login: the test is anonymous by construction
# (a fresh context from the page fixture) and the page route serves
# the HTML for any well-formed token — the 404 comes from the
# public API read, which needs the DB up (db_ready).
page.set_default_timeout(30_000)
js_errors: list[str] = []
page.on("pageerror", lambda e: js_errors.append(str(e)))
# The page route serves the page (200 HTML) for the well-formed
# token — the invalid state is rendered by the client after its
# 404 API read, not a server error page.
doc = page.goto(app_url + f"/shared/{BAD_TOKEN}")
assert doc is not None
assert doc.status == 200
expect(page.locator("#shared-invalid")).to_be_visible(timeout=15_000)
expect(page.locator("#shared-invalid")).to_contain_text(INVALID_TEXT)
# The title keeps its static fallback and nothing rendered.
expect(page.locator("#shared-title")).to_have_text("Shared conversation")
expect(page.locator(".msg")).to_have_count(0)
# The guest header rendered: Sign in visible, the admin-only nav
# links hidden.
expect(page.locator("#sign-in-link")).to_be_visible(timeout=15_000)
for admin_link in ("#nav-sources", "#nav-git-sources", "#nav-tuning", "#nav-history"):
expect(page.locator(admin_link)).to_be_hidden()
# No crash: no uncaught page errors.
assert not js_errors, f"the invalid-state render must not throw: {js_errors}"
Reference in New Issue View Git Blame Copy Permalink