feat(rag): global tuning manager — /tuning.html + PUT /api/steering/{id}: create, edit, list, delete steering notes without a chat

This commit is contained in:
2026-08-25 13:46:32 -04:00
parent fcde1fd37b
commit 589e26dbe9
16 changed files with 1697 additions and 21 deletions
+550
View File
@@ -0,0 +1,550 @@
"""Phase 27 E2E (Playwright): global tuning manager — steering notes without
a chat.
The admin manages steering notes on a dedicated page (``/tuning.html``):
create, edit inline, list, and delete — **without any chat conversation**.
Notes are stored in Postgres (``steering_notes``) and read into the system
prompt of every future turn as the ``<tuning>`` section. The mock LLM
echoes the first tuning note into its answer (`` (tuning: <first note
line>)``), so prompt injection — created OR edited on this page — is
observable in the chat UI deterministically. The page's header carries the
admin-only "Tuning" nav link (``#nav-tuning`` — is-active on this page):
like the Sources link it ships hidden and ``header.js`` reveals it once
whoami says admin, so an anonymous visitor never sees it.
Story: ``.agent/user_stories/global-tuning.md``
Run in isolation (DB must be up: ``podman compose up -d db``):
uv run pytest tests/e2e/test_global_tuning.py -v --no-cov
Test → story mapping (Playwright Mapping Rule):
1. ``test_create_note_without_chat``
2. ``test_edit_note_inline``
3. ``test_delete_note``
4. ``test_edit_note_steers_answer``
5. ``test_tuning_page_a11y_and_no_cdn``
6. ``test_anonymous_cannot_manage``
"""
from __future__ import annotations
import asyncio
import re
from pathlib import Path
from threading import Thread
from typing import Any
from playwright.sync_api import Page, expect
from sqlalchemy import select, text
from app.config import Settings
from app.db import SessionLocal
from app.models import SteeringNote
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"
TUNING_URL = "/tuning.html"
QUESTION = "How is my Kubernetes cluster set up?"
MOCK_ANSWER_MARKER = "Deterministic mock answer for E2E"
NOTE = "TUNE-MARKER be concise"
EDITED_NOTE = "EDIT-MARKER assume I'm on NixOS"
SEED_NOTE = "SEED-MARKER an existing note"
# ---------------------------------------------------------------------------
# KB seeding + DB reset (same harness pattern as test_steering.py /
# test_document_viewer.py)
# ---------------------------------------------------------------------------
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), re-import fixtures."""
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 _open_tuning(page: Page, app_url: str) -> None:
"""Form-login and land on /tuning.html, admin state fully wired.
The page's own admin-only "Tuning" nav link is the sync point: it
ships hidden and ``header.js`` reveals it once whoami says admin (the
exact Sources-link contract), so a visible ``#nav-tuning`` proves the
shared-header wiring on this page.
"""
login(page, app_url, next=TUNING_URL)
expect(page.locator("#nav-tuning")).to_be_visible(timeout=15_000)
expect(page.locator("#nav-sources")).to_be_visible()
expect(page.locator("#sign-out-btn")).to_be_visible()
expect(page.locator("#sign-in-link")).to_be_hidden()
# The link marks the current page.
expect(page.locator("#nav-tuning")).to_have_attribute("aria-current", "page")
expect(page.locator("#nav-tuning")).to_have_class(re.compile(r"\bis-active\b"))
def _create_note(page: Page, note: str) -> None:
"""Type *note* into the page's create form and submit it."""
page.fill("#tune-note", note)
page.click("#tune-save")
expect(page.locator("#tune-list .tuning-note-text")).to_have_text(note, timeout=15_000)
def _edit_note_inline(page: Page, new_note: str) -> None:
"""Open the row's inline edit form, replace the text, and Save."""
page.locator(".tuning-edit").click()
form = page.locator(".tuning-edit-form")
expect(form).to_be_visible()
form.locator(".tuning-edit-input").fill(new_note)
form.locator(".tune-save").click()
expect(page.locator("#tune-list .tuning-note-text")).to_have_text(new_note, timeout=15_000)
expect(page.locator(".tuning-edit-form")).to_have_count(0)
def _ask(page: Page, question: str) -> None:
"""Send one chat 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")
# ---------------------------------------------------------------------------
# :focus-visible Tab walk (same pattern as test_responsive_polish.py) and
# WCAG 2.1 contrast (same helpers as test_dark_tech_theme.py)
# ---------------------------------------------------------------------------
def _tab_outline_walk(page: Page, max_tabs: int = 60) -> list[dict[str, str]]:
"""Real keyboard Tab walk; returns each focused element's outline."""
first_key: str | None = None
seen: list[dict[str, str]] = []
for _ in range(max_tabs):
page.keyboard.press("Tab")
info = page.evaluate(
"""() => {
const el = document.activeElement;
const cs = getComputedStyle(el);
const cls = String(el.className).split(" ")[0];
const label = (el.getAttribute("aria-label")
|| el.textContent || "").trim().slice(0, 24);
return {
key: el.tagName + "#" + (el.id || "") + "." + cls + ":" + label,
outline_style: cs.outlineStyle,
outline_width: cs.outlineWidth,
};
}"""
)
if info["key"].startswith("BODY"):
continue # focus has not entered the document yet
if first_key is None:
first_key = info["key"]
seen.append(info)
if len(seen) > 1 and info["key"] == first_key:
break # wrapped back to the first focusable
return seen
def _rgb(value: str) -> tuple[int, int, int]:
value = value.strip()
hex_match = re.match(r"^#([0-9a-f]{6})$", value, re.IGNORECASE)
if hex_match:
h = hex_match.group(1)
return int(h[0:2], 16), int(h[2:4], 16), int(h[4:6], 16)
match = re.match(r"^rgba?\(\s*(\d+)\s*,\s*(\d+)\s*,\s*(\d+)", value)
assert match, f"unparsable color: {value!r}"
return int(match.group(1)), int(match.group(2)), int(match.group(3))
def _rel_luminance(rgb: tuple[int, int, int]) -> float:
def chan(c: int) -> float:
s = c / 255
return s / 12.92 if s <= 0.04045 else ((s + 0.055) / 1.055) ** 2.4
r, g, b = (chan(c) for c in rgb)
return 0.2126 * r + 0.7152 * g + 0.0722 * b
def contrast_ratio(fg: str, bg: str) -> float:
l1, l2 = _rel_luminance(_rgb(fg)), _rel_luminance(_rgb(bg))
if l1 < l2:
l1, l2 = l2, l1
return (l1 + 0.05) / (l2 + 0.05)
def _assert_aa(pair: Any, label: str) -> None:
fg, bg = str(pair[0]), str(pair[1])
ratio = contrast_ratio(fg, bg)
assert ratio >= 4.5, f"contrast {label}: {fg} on {bg} = {ratio:.2f}:1 (< 4.5:1)"
# ---------------------------------------------------------------------------
# 1. Create a note without any chat
# ---------------------------------------------------------------------------
def test_create_note_without_chat(
page: Page, app_url: str, mock_llm: int, db_ready: None
) -> None:
_reset_db(mock_port=0, seed=False)
page.set_default_timeout(30_000)
_open_tuning(page, app_url)
# Fresh page: the empty state, no notes — and no chat anywhere in
# sight (the story's point: a note can exist before any conversation).
expect(page.locator("#tune-list .tuning-note")).to_have_count(0)
expect(page.locator("#tune-empty")).to_be_visible()
assert page.locator(".msg").count() == 0, "the tuning page carries no chat"
# Create: type + submit. The row lands in the list and the empty
# state steps aside…
_create_note(page, NOTE)
expect(page.locator("#tune-list .tuning-note")).to_have_count(1)
expect(page.locator("#tune-empty")).to_be_hidden()
# …the live region announces it, and the form is cleared (201).
expect(page.locator("#tune-announcer")).to_contain_text("Tuning note added.")
expect(page.locator("#tune-note")).to_have_value("")
# Persisted in Postgres — with zero chat turns in this whole test.
with SessionLocal() as db:
rows = db.scalars(select(SteeringNote)).all()
assert [r.note for r in rows] == [NOTE]
# ---------------------------------------------------------------------------
# 2. Edit a note inline (Cancel reverts, Save persists)
# ---------------------------------------------------------------------------
def test_edit_note_inline(
page: Page, app_url: str, mock_llm: int, db_ready: None
) -> None:
_reset_db(mock_port=0, seed=False)
page.set_default_timeout(30_000)
_open_tuning(page, app_url)
_create_note(page, NOTE)
# Open the inline edit form: the row swaps to a prefilled textarea
# with a (visually-hidden) label, while the row's text span and its
# Edit/Delete buttons step aside.
page.locator(".tuning-edit").click()
form = page.locator(".tuning-edit-form")
expect(form).to_be_visible()
edit_input = form.locator(".tuning-edit-input")
expect(edit_input).to_have_value(NOTE)
expect(form.locator("label.visually-hidden")).to_have_count(1)
expect(page.locator(".tuning-note-text")).to_be_hidden()
expect(page.locator(".tuning-note > .tuning-edit")).to_be_hidden()
expect(page.locator(".tuning-note > .tuning-delete")).to_be_hidden()
# Cancel reverts to the text span — the note is untouched.
form.locator(".tune-cancel").click()
expect(page.locator(".tuning-edit-form")).to_have_count(0)
expect(page.locator("#tune-list .tuning-note-text")).to_have_text(NOTE)
# Now for real: open, change the text, Save.
page.locator(".tuning-edit").click()
form = page.locator(".tuning-edit-form")
expect(form).to_be_visible()
form.locator(".tuning-edit-input").fill(EDITED_NOTE)
form.locator(".tune-save").click()
# The list shows the NEW text, the edit form is gone, and a "Saved"
# pill + the live region confirm the update.
expect(page.locator("#tune-list .tuning-note-text")).to_have_text(
EDITED_NOTE, timeout=15_000
)
expect(page.locator(".tuning-edit-form")).to_have_count(0)
expect(page.locator(".tuning-note-text")).to_be_visible()
expect(page.locator(".tuning-saved")).to_contain_text("Saved")
expect(page.locator("#tune-announcer")).to_contain_text("Tuning note updated.")
# Postgres: still exactly one note, with the updated text (edited in
# place — no duplicate row).
with SessionLocal() as db:
rows = db.scalars(select(SteeringNote)).all()
assert [r.note for r in rows] == [EDITED_NOTE]
# ---------------------------------------------------------------------------
# 3. Delete a note → row removed, empty state back
# ---------------------------------------------------------------------------
def test_delete_note(
page: Page, app_url: str, mock_llm: int, db_ready: None
) -> None:
_reset_db(mock_port=0, seed=False)
page.set_default_timeout(30_000)
_open_tuning(page, app_url)
_create_note(page, NOTE)
expect(page.locator("#tune-empty")).to_be_hidden()
# Delete: the row leaves the DOM and the empty state comes back.
page.locator(".tuning-delete").click()
expect(page.locator("#tune-list .tuning-note")).to_have_count(0, timeout=15_000)
expect(page.locator("#tune-empty")).to_be_visible()
expect(page.locator("#tune-announcer")).to_contain_text("Tuning note deleted.")
# Gone from Postgres.
with SessionLocal() as db:
assert db.scalars(select(SteeringNote)).all() == []
# ---------------------------------------------------------------------------
# 4. A note created + edited on /tuning.html steers the next chat answer
# ---------------------------------------------------------------------------
def test_edit_note_steers_answer(
page: Page, app_url: str, mock_llm: int, db_ready: None
) -> None:
summary = _reset_db(mock_llm, seed=True)
assert summary is not None and summary.added == 8 # A9 formats
page.set_default_timeout(30_000)
_open_tuning(page, app_url)
# Create the note without a chat…
_create_note(page, NOTE)
# …and edit it to the final wording (inline — still no chat).
_edit_note_inline(page, EDITED_NOTE)
expect(page.locator("#tune-announcer")).to_contain_text("Tuning note updated.")
# Now to the chat via the header's "Chat" nav link (a real navigation):
# the next answer must carry the EDITED note. The mock echoes the
# first <tuning> line, so the marker proves the edited note reached
# the system prompt.
page.click('a.nav-link[href="/"]')
expect(page.locator("#message-input")).to_be_visible(timeout=15_000)
_ask(page, QUESTION)
bubble = page.locator(".msg.brain .bubble").last
expect(bubble).to_contain_text(MOCK_ANSWER_MARKER)
expect(bubble).to_contain_text(f"(tuning: {EDITED_NOTE})")
# ---------------------------------------------------------------------------
# 5. Page accessibility + no-CDN
# ---------------------------------------------------------------------------
def test_tuning_page_a11y_and_no_cdn(
page: Page, app_url: str, mock_llm: int, db_ready: None
) -> None:
_reset_db(mock_port=0, seed=False)
# One note in the DB before load so the page renders a real row (the
# row's Edit/Delete buttons are JS-built — they need a note).
with SessionLocal() as db:
db.add(SteeringNote(note=SEED_NOTE))
db.commit()
page.set_default_timeout(30_000)
_open_tuning(page, app_url)
expect(page.locator("#tune-list .tuning-note")).to_have_count(1)
page.wait_for_load_state("networkidle")
# Landmarks (PLAN §7.2).
assert page.locator("header.app-header").count() == 1, "header missing"
assert page.locator("nav[aria-label]").count() == 1, "labeled nav missing"
assert page.locator("main#main").count() == 1, "main#main missing"
assert page.locator("footer.app-footer").count() == 1, "footer missing"
# Skip link: present, first Tab lands on it, Enter moves focus to #main.
skip = page.locator('a.skip-link[href="#main"]')
assert skip.count() == 1, "skip link missing"
page.keyboard.press("Tab")
assert page.evaluate("() => document.activeElement.className") == "skip-link", (
"first Tab must land on the skip link"
)
page.keyboard.press("Enter")
assert page.evaluate("() => document.activeElement.id") == "main", (
"skip link must move focus to #main"
)
# Labeled controls: the create textarea has its (visually-hidden)
# label, every button has an accessible name, and the per-row Delete
# is labeled with the note text (Edit carries visible text).
expect(page.get_by_label("Add a global tuning note")).to_have_count(1)
unnamed = page.evaluate(
"""() => [...document.querySelectorAll("button")]
.filter((b) => !(b.getAttribute("aria-label") || b.textContent.trim()))
.length"""
)
assert unnamed == 0, f"{unnamed} button(s) without an accessible name"
delete_btn = page.locator("#tune-list .tuning-delete")
assert (delete_btn.get_attribute("aria-label") or "").startswith(
"Delete tuning note:"
), "row Delete must be labeled with the note text"
expect(page.locator("#tune-list .tuning-edit")).to_have_text("Edit")
# The live region that announces create / edit / delete.
announcer = page.locator("#tune-announcer")
assert announcer.get_attribute("role") == "status"
assert announcer.get_attribute("aria-live") == "polite"
# ≥44px touch targets on every interactive control.
for selector in (
"#tune-save",
"#tune-note",
"#nav-tuning",
".new-chat-btn",
"#sign-out-btn",
".tuning-edit",
".tuning-delete",
):
box = page.locator(selector).first.bounding_box()
assert box is not None and box["height"] >= 44, f"target too small: {selector} {box}"
# :focus-visible — a real keyboard Tab walk: every focused element
# shows a visible outline (solid, ≥2px; the design uses 3px).
page.evaluate(
"() => { if (document.activeElement instanceof HTMLElement)"
" document.activeElement.blur(); }"
)
seen = _tab_outline_walk(page)
assert len(seen) >= 6, f"expected several focusables, tabbed {len(seen)}"
for info in seen:
width_px = float(info["outline_width"].replace("px", ""))
assert info["outline_style"] == "solid" and width_px >= 2, (
f"no visible focus outline on {info['key']} "
f"({info['outline_style']} {info['outline_width']})"
)
# Dark theme: the page canvas is the phase-08 dark bg (body stays
# transparent for the background layers).
bg = page.evaluate("() => getComputedStyle(document.documentElement).backgroundColor")
assert bg == "rgb(10, 14, 23)", f"expected the dark page bg, got {bg}"
# Contrast: the live text/background pairs compute ≥ 4.5:1 (AA).
pairs = page.evaluate(
"""() => {
const cs = (sel, prop) => getComputedStyle(document.querySelector(sel))[prop];
return {
save_btn: [cs("#tune-save", "color"), cs("#tune-save", "backgroundColor")],
active_nav: [
cs(".nav-link.is-active", "color"),
cs(".nav-link.is-active", "backgroundColor"),
],
note_text: [
cs(".tuning-note-text", "color"),
cs(".tuning-panel", "backgroundColor"),
],
empty_state: [
cs("#tune-empty", "color"),
cs(".tuning-panel", "backgroundColor"),
],
};
}"""
)
_assert_aa(pairs["save_btn"], "dark ink on brand (Add note)")
_assert_aa(pairs["active_nav"], "dark ink on brand (active nav)")
_assert_aa(pairs["note_text"], "ink on surface (note text)")
_assert_aa(pairs["empty_state"], "ink-soft on surface (empty state)")
# No CDN: every script/link reference is same-origin or a data: URI.
refs = page.evaluate(
"""() => [...document.querySelectorAll("script[src], link[href]")]
.map((el) => el.src || el.href)"""
)
assert refs, "expected local asset references"
for ref in refs:
assert ref.startswith(app_url) or ref.startswith("data:"), (
f"non-local asset reference: {ref}"
)
# ---------------------------------------------------------------------------
# 6. Anonymous: no list, no PUT, no create
# ---------------------------------------------------------------------------
def test_anonymous_cannot_manage(
page: Page, app_url: str, mock_llm: int, db_ready: None
) -> None:
_reset_db(mock_port=0, seed=False)
# A note exists in the DB (the admin saved it at some point) — an
# anonymous visitor must never see it or touch it.
note = SteeringNote(note=SEED_NOTE)
with SessionLocal() as db:
db.add(note)
db.commit()
db.refresh(note)
note_id = note.id
page.set_default_timeout(30_000)
page.goto(app_url + TUNING_URL)
# Anonymous header state: Sign in visible, and the admin-only nav
# links — Sources AND this page's own Tuning link — stay hidden.
expect(page.locator("#sign-in-link")).to_be_visible()
expect(page.locator("#sign-out-btn")).to_be_hidden()
expect(page.locator("#nav-sources")).to_be_hidden()
expect(page.locator("#nav-tuning")).to_be_hidden()
# The list stays on its empty state even though a note exists…
expect(page.locator("#tune-list .tuning-note")).to_have_count(0)
expect(page.locator("#tune-empty")).to_be_visible()
# …and the API is gated: GET and a PUT on a REAL note id are 403.
get_status = page.evaluate("async () => (await fetch('/api/steering')).status")
assert get_status == 403, f"anonymous GET /api/steering → {get_status}"
put_status = page.evaluate(
"""async (id) => (await fetch('/api/steering/' + id, {
method: 'PUT',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify({note: 'anonymous overwrite attempt'}),
})).status""",
str(note_id),
)
assert put_status == 403, f"anonymous PUT /api/steering/{note_id} → {put_status}"
with SessionLocal() as db:
row = db.get(SteeringNote, note_id)
assert row is not None and row.note == SEED_NOTE, "the note must stay untouched"
# The create form 403s gracefully: an inline error (role=alert) with
# the API detail, and the typed instruction survives in the textarea.
page.fill("#tune-note", "an anonymous attempt")
page.click("#tune-save")
error = page.locator("#tune-form .tuning-error")
expect(error).to_have_attribute("role", "alert")
expect(error).to_be_visible(timeout=15_000)
assert "admin only" in (error.inner_text() or "").lower()
expect(page.locator("#tune-note")).to_have_value("an anonymous attempt")
# Nothing was created: the DB still holds only the seeded note.
with SessionLocal() as db:
assert [r.note for r in db.scalars(select(SteeringNote)).all()] == [SEED_NOTE]
+3
View File
@@ -57,6 +57,7 @@ def test_suggestions_honors_bor_suggestions_env_override(monkeypatch) -> None:
("/sources.html", "Knowledge base"),
("/document.html", "Brain of Reese"), # phase 10: viewer page
("/login.html", "Sign in"), # phase 16: admin sign-in page
("/tuning.html", "Global Tuning"), # phase 27: global tuning page
],
)
def test_html_pages_served_locally_no_cdn(client, path: str, marker: str) -> None:
@@ -78,6 +79,7 @@ def test_styles_and_js_served(client) -> None:
assert client.get("/assets/document.js").status_code == 200 # phase 10: viewer page
assert client.get("/assets/login.js").status_code == 200 # phase 16: login page
assert client.get("/assets/document-modal.js").status_code == 200 # phase 26: modal module
assert client.get("/assets/tuning.js").status_code == 200 # phase 27: tuning page
# Emoji code points banned from UI chrome (phase 08): the pictograph
@@ -108,6 +110,7 @@ def _find_emoji(text: str) -> list[str]:
"/sources.html",
"/document.html",
"/login.html", # phase 16
"/tuning.html", # phase 27
"/assets/app.js",
"/assets/sources.js",
"/assets/markdown.js",
@@ -22,9 +22,11 @@ from sqlalchemy import select, text
from test_chat_api import FakeRagLLM, _stream_chat
from app.api import chat as chat_api
from app.api.steering import load_steering_notes
from app.main import app as fastapi_app
from app.models import SteeringNote
from app.rag.importer import import_sources
from app.rag.prompts import build_steering_section
FIXTURES = Path(__file__).resolve().parents[1] / "fixtures" / "docs"
QUESTION = "How is my Kubernetes cluster set up?"
@@ -131,6 +133,92 @@ def test_create_enforces_2000_char_limit(admin_client: TestClient) -> None:
assert len(r.json()["note"]) == 2000
# ---------- update (PUT) — phase 27: edit a note in place ----------
def test_update_note_returns_200_and_replaces_text(admin_client: TestClient, db) -> None:
created = admin_client.post("/api/steering", json={"note": NOTE}).json()
r = admin_client.put(f"/api/steering/{created['id']}", json={"note": f" {NOTE}-v2 "})
assert r.status_code == 200
body = r.json()
assert body["id"] == created["id"]
assert body["note"] == f"{NOTE}-v2" # trimmed, full replacement
# Editing does not redate the note.
assert datetime.fromisoformat(body["created_at"]) == datetime.fromisoformat(
created["created_at"]
)
row = db.get(SteeringNote, uuid.UUID(created["id"]))
assert row is not None
assert row.note == f"{NOTE}-v2"
def test_update_is_reflected_in_list_load_and_prompt(admin_client: TestClient, db) -> None:
base = datetime.now(UTC)
db.add_all(
[
SteeringNote(note="oldest note", created_at=base),
SteeringNote(note="newest note", created_at=base + timedelta(hours=1)),
]
)
db.commit()
oldest = db.scalars(select(SteeringNote).order_by(SteeringNote.created_at.asc())).first()
assert oldest is not None
updated = "oldest note — updated"
assert admin_client.put(f"/api/steering/{oldest.id}", json={"note": updated}).status_code == 200
# The GET list keeps its newest-first order and carries the new text.
body = admin_client.get("/api/steering").json()
assert [n["note"] for n in body["notes"]] == ["newest note", updated]
# The chat path reads the updated note back, oldest first… (expire the
# test session's identity map so the fresh DB values are loaded).
db.expire_all()
notes = load_steering_notes(db)
assert notes == [updated, "newest note"]
# …and it appears, in order, in the <tuning> prompt section.
section = build_steering_section(notes)
assert f"1. {updated}" in section
assert "2. newest note" in section
assert section.index("1. ") < section.index("2. ")
def test_update_unknown_note_returns_404(admin_client: TestClient) -> None:
r = admin_client.put(f"/api/steering/{uuid.uuid4()}", json={"note": "x"})
assert r.status_code == 404
assert r.json() == {"detail": "steering note not found"}
def test_update_rejects_invalid_bodies(admin_client: TestClient) -> None:
created = admin_client.post("/api/steering", json={"note": NOTE}).json()
note_id = created["id"]
assert admin_client.put(f"/api/steering/{note_id}", json={"note": ""}).status_code == 422
assert admin_client.put(f"/api/steering/{note_id}", json={"note": " "}).status_code == 422
assert (
admin_client.put(f"/api/steering/{note_id}", json={"note": "x" * 2001}).status_code == 422
)
# The 2000-char boundary passes and is stored as-is.
r = admin_client.put(f"/api/steering/{note_id}", json={"note": "y" * 2000})
assert r.status_code == 200
assert len(r.json()["note"]) == 2000
def test_update_anonymous_returns_403(admin_client: TestClient, db) -> None:
created = admin_client.post("/api/steering", json={"note": NOTE}).json()
anon = TestClient(fastapi_app) # fresh jar: truly anonymous
r = anon.put(f"/api/steering/{created['id']}", json={"note": "anonymous edit"})
assert r.status_code == 403
assert r.json() == {"detail": "admin only"}
# The text is untouched, in both the API and the chat path.
body = admin_client.get("/api/steering").json()
assert [n["note"] for n in body["notes"]] == [NOTE]
assert load_steering_notes(db) == [NOTE]
# ---------- chat turn: note reaches the system prompt ----------
+29 -8
View File
@@ -20,12 +20,14 @@ APP_JS = ASSETS / "app.js"
SOURCES_JS = ASSETS / "sources.js"
DOCUMENT_JS = ASSETS / "document.js"
LOGIN_JS = ASSETS / "login.js"
TUNING_JS = ASSETS / "tuning.js"
STYLES_CSS = ASSETS / "styles.css"
INDEX_HTML = FRONTEND / "index.html"
SOURCES_HTML = FRONTEND / "sources.html"
DOCUMENT_HTML = FRONTEND / "document.html"
LOGIN_HTML = FRONTEND / "login.html"
TUNING_HTML = FRONTEND / "tuning.html"
def _text(path: Path) -> str:
@@ -74,7 +76,7 @@ def test_init_shared_header_toggles_only_elements_that_exist() -> None:
assert fn != -1
body = js[fn : js.find("\n}", fn)]
assert "await fetchIsAdmin()" in body
for selector in ('#sign-in-link', '#sign-out-btn', '#nav-sources'):
for selector in ("#sign-in-link", "#sign-out-btn", "#nav-sources", "#nav-tuning"):
assert f'querySelector("{selector}")' in body
assert "return admin" in body, "callers may reuse the flag"
@@ -111,13 +113,32 @@ def test_nav_sources_ships_hidden_on_every_nav_page() -> None:
nav link is hidden for anonymous — so it SHIPS with the hidden
attribute (anonymous-safe default) on every page that has a nav
(chat, sources, login)."""
for html in (INDEX_HTML, SOURCES_HTML, LOGIN_HTML):
for html in (INDEX_HTML, SOURCES_HTML, LOGIN_HTML, TUNING_HTML):
text = _text(html)
assert re.search(r'id="nav-sources"[^>]*\bhidden\b', text), (
f"{html.name}: #nav-sources must ship hidden"
)
def test_nav_tuning_ships_hidden_on_the_tuning_page() -> None:
"""Phase 27: the Global Tuning page reuses the shared header — the
"Tuning" nav link is admin-only, so it SHIPS hidden (revealed by
initSharedHeader once whoami says admin), is the page's active link
(is-active + aria-current), and the page loads markdown.js (classic)
+ the tuning.js module with NO direct header.js <script> tag
(single-evaluation design)."""
text = _text(TUNING_HTML)
tag = re.search(r'<a[^>]*id="nav-tuning"[^>]*>', text)
assert tag, "tuning.html must carry the #nav-tuning nav link"
assert 'class="nav-link is-active"' in tag.group(0), "the Tuning link is the active one"
assert 'aria-current="page"' in tag.group(0)
assert "hidden" in tag.group(0), "#nav-tuning must ship hidden (admin-only)"
srcs = _script_srcs(TUNING_HTML)
assert [s for s in srcs if "header.js" in s] == [], "no direct header.js <script> tag"
assert [s for s in srcs if "markdown.js" in s]
assert [s for s in srcs if "tuning.js" in s]
def test_nav_sources_is_absent_from_the_viewer() -> None:
"""The document viewer has no nav — no #nav-sources element there (the
module's missing-element no-op keeps it out)."""
@@ -229,12 +250,12 @@ def test_login_js_uses_the_shared_fetch_is_admin() -> None:
def test_non_chat_pages_bind_new_chat_to_the_chat_page() -> None:
"""On sources and the viewer, New Chat means "go to the chat,
fresh": the binding clears the phase-14 key (clearChatStorage) and
navigates to "/" — and both pages run initSharedHeader() at boot
on the shared cached whoami. Phase 23: the import is relative
(`./header.js`)."""
for js_file in (SOURCES_JS, DOCUMENT_JS):
"""On sources, the viewer, and the tuning page, New Chat means "go to
the chat, fresh": the binding clears the phase-14 key
(clearChatStorage) and navigates to "/" — and each page runs
initSharedHeader() at boot on the shared cached whoami. Phase 23:
the import is relative (`./header.js`)."""
for js_file in (SOURCES_JS, DOCUMENT_JS, TUNING_JS):
js = _text(js_file)
assert 'from "./header.js"' in js
assert "initSharedHeader()" in js
+139
View File
@@ -0,0 +1,139 @@
"""Unit: steering notes PUT endpoint (phase 27 — Global Tuning).
``PUT /api/steering/{note_id}`` is driven **via the router** with a
stubbed session (FastAPI dependency override — no Postgres required):
200 with the updated note (trimmed, ``created_at`` preserved), 404 for an
unknown id, 422 for empty/whitespace/over-2000 bodies, and the
router-level 403 for anonymous callers (the real phase-16
``require_admin`` gate — no auth surface added by the PUT route itself).
"""
from __future__ import annotations
import uuid
from datetime import UTC, datetime
from fastapi.testclient import TestClient
from app.config import get_settings
from app.db import get_db
from app.main import create_app
from app.models import SteeringNote
CREATED_AT = datetime(2026, 8, 24, 12, 0, 0, tzinfo=UTC)
class _FakeSession:
"""Just enough of a SQLAlchemy session for the PUT route."""
def __init__(self, rows: dict[uuid.UUID, SteeringNote]) -> None:
self.rows = rows
self.commits = 0
def get(self, _model: object, pk: object) -> SteeringNote | None:
if isinstance(pk, uuid.UUID):
return self.rows.get(pk)
return None
def commit(self) -> None:
self.commits += 1
def refresh(self, _row: object) -> None:
pass
def _note() -> SteeringNote:
return SteeringNote(id=uuid.uuid4(), note="be concise", created_at=CREATED_AT)
def _put_client(rows: dict[uuid.UUID, SteeringNote]) -> tuple[TestClient, _FakeSession]:
"""A fresh app whose ``get_db`` is a stub holding ``rows``."""
session = _FakeSession(rows)
app = create_app()
app.dependency_overrides[get_db] = lambda: session
return TestClient(app), session
def _sign_in(client: TestClient) -> None:
"""Real login route (no DB): sets the signed admin session cookie."""
r = client.post("/api/login", json={"password": get_settings().admin_password})
assert r.status_code == 204, f"admin login failed: {r.status_code} {r.text}"
# ---------- 200: update in place ----------
def test_update_note_returns_200_with_new_text() -> None:
note = _note()
client, session = _put_client({note.id: note})
_sign_in(client)
r = client.put(f"/api/steering/{note.id}", json={"note": " be MORE concise "})
assert r.status_code == 200
body = r.json()
assert body["note"] == "be MORE concise" # trimmed before storage
assert body["id"] == str(note.id)
assert datetime.fromisoformat(body["created_at"]) == CREATED_AT # preserved, not redated
assert note.note == "be MORE concise" # updated in place
assert session.commits == 1
def test_update_accepts_full_length_note() -> None:
note = _note()
client, _session = _put_client({note.id: note})
_sign_in(client)
r = client.put(f"/api/steering/{note.id}", json={"note": "x" * 2000})
assert r.status_code == 200
assert len(r.json()["note"]) == 2000
# ---------- 404 / 422 ----------
def test_update_unknown_id_returns_404() -> None:
note = _note()
client, _session = _put_client({note.id: note})
_sign_in(client)
r = client.put(f"/api/steering/{uuid.uuid4()}", json={"note": "anything"})
assert r.status_code == 404
assert r.json() == {"detail": "steering note not found"}
assert note.note == "be concise" # untouched
def test_update_invalid_id_returns_422() -> None:
note = _note()
client, _session = _put_client({note.id: note})
_sign_in(client)
assert client.put("/api/steering/not-a-uuid", json={"note": "x"}).status_code == 422
def test_update_rejects_empty_blank_and_overlong_bodies() -> None:
note = _note()
client, session = _put_client({note.id: note})
_sign_in(client)
assert client.put(f"/api/steering/{note.id}", json={"note": ""}).status_code == 422
assert client.put(f"/api/steering/{note.id}", json={"note": " \t\n "}).status_code == 422
assert client.put(f"/api/steering/{note.id}", json={"note": "x" * 2001}).status_code == 422
assert note.note == "be concise" # rejected bodies never touch the row
assert session.commits == 0
# ---------- 403: router-level require_admin (phase 16) ----------
def test_update_anonymous_returns_403_admin_only() -> None:
note = _note()
client, session = _put_client({note.id: note})
r = client.put(f"/api/steering/{note.id}", json={"note": "sneaky"})
assert r.status_code == 403
assert r.json() == {"detail": "admin only"} # the require_admin message
assert note.note == "be concise" # anonymous callers never mutate
assert session.commits == 0