"""Integration: steering notes (phase 15) — CRUD + system-prompt injection. Real Postgres (``podman compose up -d db``); the chat path reuses the deterministic fake LLM from ``test_chat_api`` (token-overlap embeddings), so the stored note's journey — API → Postgres → ```` section of the captured system prompt — is verified end-to-end without a network. Requires: podman compose up -d db """ from __future__ import annotations import asyncio import logging import uuid from collections.abc import Iterator from datetime import UTC, datetime, timedelta from pathlib import Path import pytest from fastapi.testclient import TestClient 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?" OFF_TOPIC = "How do I bake sourdough bread?" NOTE = "STEEER-MARKER be concise" @pytest.fixture(autouse=True) def clean_steering(db) -> Iterator[None]: """Steering notes + query log are global state: reset around every test.""" db.execute(text("TRUNCATE steering_notes, query_log")) db.commit() yield db.execute(text("TRUNCATE steering_notes, query_log")) db.commit() @pytest.fixture() def seeded_kb(db) -> Iterator[FakeRagLLM]: """Fresh Postgres with the fixture docs imported (real pipeline).""" db.execute(text("TRUNCATE chunks, documents, query_log, steering_notes")) db.commit() llm = FakeRagLLM() summary = asyncio.run(import_sources([FIXTURES], llm, session=db)) assert summary.added == 13 # A9 formats (phase 47 added quadlet+j2); .hidden/ skipped yield llm db.execute(text("TRUNCATE chunks, documents, query_log, steering_notes")) db.commit() def _turn_log_lines(caplog: pytest.LogCaptureFixture) -> list[str]: """The per-turn ``question=…`` log lines (PLAN §9) from this test.""" return [r.getMessage() for r in caplog.records if "question=" in r.getMessage()] # ---------- CRUD ---------- def test_create_note_returns_201_and_stores_trimmed(admin_client: TestClient, db) -> None: r = admin_client.post("/api/steering", json={"note": f" {NOTE} "}) assert r.status_code == 201 body = r.json() assert body["note"] == NOTE # trimmed before storage uuid.UUID(body["id"]) # valid UUID assert body["created_at"] rows = db.scalars(select(SteeringNote)).all() assert [row.note for row in rows] == [NOTE] def test_list_notes_empty(admin_client: TestClient) -> None: r = admin_client.get("/api/steering") assert r.status_code == 200 assert r.json() == {"notes": []} def test_list_notes_newest_first(admin_client: TestClient, db) -> None: base = datetime.now(UTC) db.add_all( [ SteeringNote(note="oldest", created_at=base), SteeringNote(note="newest", created_at=base + timedelta(hours=2)), SteeringNote(note="middle", created_at=base + timedelta(hours=1)), ] ) db.commit() r = admin_client.get("/api/steering") assert r.status_code == 200 body = r.json() assert [n["note"] for n in body["notes"]] == ["newest", "middle", "oldest"] for n in body["notes"]: assert set(n) == {"id", "note", "created_at"} uuid.UUID(n["id"]) def test_delete_note_returns_204_and_removes(admin_client: TestClient, db) -> None: created = admin_client.post("/api/steering", json={"note": NOTE}).json() assert admin_client.delete(f"/api/steering/{created['id']}").status_code == 204 assert admin_client.get("/api/steering").json() == {"notes": []} assert db.scalars(select(SteeringNote)).all() == [] def test_delete_unknown_note_returns_404(admin_client: TestClient) -> None: r = admin_client.delete(f"/api/steering/{uuid.uuid4()}") assert r.status_code == 404 assert "not found" in r.json()["detail"] def test_delete_invalid_id_returns_422(admin_client: TestClient) -> None: assert admin_client.delete("/api/steering/not-a-uuid").status_code == 422 def test_create_rejects_empty_and_blank_notes(admin_client: TestClient) -> None: assert admin_client.post("/api/steering", json={"note": ""}).status_code == 422 assert admin_client.post("/api/steering", json={"note": " \t\n "}).status_code == 422 assert admin_client.get("/api/steering").json() == {"notes": []} def test_create_enforces_2000_char_limit(admin_client: TestClient) -> None: assert admin_client.post("/api/steering", json={"note": "x" * 2001}).status_code == 422 r = admin_client.post("/api/steering", json={"note": "x" * 2000}) assert r.status_code == 201 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 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 ---------- def test_chat_turn_high_mode_receives_note_in_system_prompt( admin_client: TestClient, seeded_kb: FakeRagLLM, caplog: pytest.LogCaptureFixture ) -> None: admin_client.post("/api/steering", json={"note": NOTE}) caplog.set_level(logging.INFO, logger="app.chat") fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb try: _, _, frames = _stream_chat(admin_client, QUESTION) finally: fastapi_app.dependency_overrides.clear() assert frames[-1]["type"] == "done" assert frames[-1]["deflected"] is False # The LLM received the HIGH prompt with the section. (system, user) = seeded_kb.seen_messages[0][0], seeded_kb.seen_messages[0][1] assert user["content"] == QUESTION assert "HIGH" in system["content"] assert "" in system["content"] assert f"1. {NOTE}" in system["content"] assert "" in system["content"] # The section sits between the relevance marker and the documents. assert ( system["content"].index("HIGH") < system["content"].index("") < system["content"].index("") < system["content"].index("") ) # The per-turn log line records tuning=N (PLAN §9). lines = _turn_log_lines(caplog) assert lines and "tuning=1" in lines[-1] def test_chat_turn_low_mode_receives_note_in_system_prompt( admin_client: TestClient, seeded_kb: FakeRagLLM, caplog: pytest.LogCaptureFixture ) -> None: admin_client.post("/api/steering", json={"note": NOTE}) caplog.set_level(logging.INFO, logger="app.chat") fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb try: _, _, frames = _stream_chat(admin_client, OFF_TOPIC) finally: fastapi_app.dependency_overrides.clear() assert frames[-1]["deflected"] is True (system, _user) = seeded_kb.seen_messages[0][0], seeded_kb.seen_messages[0][1] assert "LOW" in system["content"] assert "DEFLECT_MODE" in system["content"] assert "" in system["content"] assert f"1. {NOTE}" in system["content"] lines = _turn_log_lines(caplog) assert lines and "tuning=1" in lines[-1] def test_chat_turn_without_notes_has_no_tuning_section( admin_client: TestClient, seeded_kb: FakeRagLLM, caplog: pytest.LogCaptureFixture ) -> None: caplog.set_level(logging.INFO, logger="app.chat") fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb try: _stream_chat(admin_client, QUESTION) finally: fastapi_app.dependency_overrides.clear() (system, _user) = seeded_kb.seen_messages[0][0], seeded_kb.seen_messages[0][1] assert "" not in system["content"] lines = _turn_log_lines(caplog) assert lines and "tuning=0" in lines[-1] def test_chat_turn_numbers_notes_oldest_first(admin_client: TestClient, db, seeded_kb) -> None: base = datetime.now(UTC) db.add_all( [ SteeringNote(note="older note", created_at=base), SteeringNote(note="newer note", created_at=base + timedelta(hours=1)), ] ) db.commit() fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb try: _stream_chat(admin_client, QUESTION) finally: fastapi_app.dependency_overrides.clear() (system, _user) = seeded_kb.seen_messages[0][0], seeded_kb.seen_messages[0][1] assert "" in system["content"] assert "1. older note" in system["content"] assert "2. newer note" in system["content"] assert system["content"].index("1. older note") < system["content"].index("2. newer note") def test_multiple_turns_keep_reading_notes(admin_client: TestClient, seeded_kb) -> None: """The note steers EVERY subsequent turn, not just the next one.""" admin_client.post("/api/steering", json={"note": NOTE}) fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb try: _stream_chat(admin_client, QUESTION) _stream_chat(admin_client, QUESTION) assert len(seeded_kb.seen_messages) == 2 for messages in seeded_kb.seen_messages: assert f"1. {NOTE}" in messages[0]["content"] # Delete → the following turn is clean again. note_id = admin_client.get("/api/steering").json()["notes"][0]["id"] assert admin_client.delete(f"/api/steering/{note_id}").status_code == 204 _stream_chat(admin_client, QUESTION) assert len(seeded_kb.seen_messages) == 3 assert "" not in seeded_kb.seen_messages[-1][0]["content"] finally: fastapi_app.dependency_overrides.clear()