Files
brain-of-reese/tests/integration/test_admin_settings_api.py
ducoterra b34b8ca672
Build and Push Containers / build-and-push-app (push) Successful in 1m46s
Build and Push Containers / build-and-push-db (push) Successful in 12s
phase: 182_verbatim_show_text
**Phase 182 — final verification pass (all 8 tasks already complete; this was verification + docs + one defect fix).**

- Verified: `app/rag/verbatim.py` (locator grammar, `splitlines()` slice, line-boundary cap, `resolve_answer_blocks`), `SHOW_TEXT_TOOL` + `shown_texts`, `SHOW_TEXT_CLAUSE`, `done.answer_text` (9 fields, key dropped when empty) + `ChatMessageRecord.answer_text` (≤50, 600 000-char cap counted in the 8 M budget), `verbatim.js` + 4 `{verbatim:true}` renderers, `Containerfile` minify step, `.env.example` 3 vars.
- `uv run python -c "…answer_text, show_text_max_chars, show_text_max_blocks"` → `True 12000 5`; each negative refuses startup naming its field; `0` accepted as inert ✅
- `bash .agents/validate.sh` → **validation OK**: 4320 unit+integration tests green, `TOTAL 6849 17 → 99%` (gate app/ >90% OK), `ruff check .` "All checks passed!", `pyright` 0 errors (re-run after my edits, 08:28).
- E2E, each in isolation `--no-cov`: `test_verbatim_show_text.py` 13 passed (panel bytes == `documents.content`, server caption, clipboard == server string, `/document.html?source=…&path=…`, honest unresolved state, reload+share, `show_image` regression, flag-off app) ✅; green: `test_answer_images` `test_harness_aligned_tools` `test_agent_document_tools` `test_markdown_tables` `test_markdown_inline_links_images` `test_svg_document_preview` `test_smoke` `test_document_viewer` `test_markdown_blockquotes` `test_chat_persistence` `test_thinking_timeline` `test_chat_rag` `test_settings_mobile_value_width`.
- **Defect found & fixed (pre-existing, not phase 182):** `tests/e2e/test_admin_settings_panel.py::test_admin_sees_groups_and_source_badges` failed — its whole-page `assert "reese" not in page.content()` tripped on the shell's onboarding chips, which follow the newest saved chats in the one shared E2E DB (a leftover `reeselink.json` question from a tool suite). Scoped the strict sweep to `.settings-shell` and kept a whole-page `reese:reese` credential needle → 12 passed with the leftovers still present.
- Docs: README fixed 2 inconsistencies — Quick Start omitted the **required** `BOR_ADMIN_PASSWORD`/`BOR_SESSION_SECRET` (the app fail-loud refuses boot) and the broken anchor `#chat--` → `#chat-`; full re-read of README (1816 lines) + AGENTS confirmed the phase-182 feature/settings/log/UI-contract text matches the code, and every `.env.example` var + every `Settings` field is documented.
- Notable: kept `resolve_answer_blocks` synchronous on the SSE terminal path (bounded indexed SELECTs, matching the existing `query_log` precedent and the phase's documented fence-lookup bound); no live-infra changes made.
- Next pending phase: none — `todo/` holds only 182 (its 8 task files are in `complete/`; the harness commits + moves `00_phase.md`).
2026-10-15 08:33:12 -04:00

997 lines
46 KiB
Python

"""Integration: the admin settings panel's read/write surface (phase 168,
task 04) — ``/api/admin/settings`` GET / PUT ``/{key}`` / DELETE
``/{key}`` / DELETE all, driven through the real app (TestClient keeps
the cookie jar — the house ``test_ui_settings_api`` /
``test_tokens_api`` pattern).
The pins (00_phase, Testing & Quality):
* the auth matrix — anonymous AND a signed-in token USER get 403 on ALL
four routes (the router-wide ``require_admin`` closes the surface on
every method), the admin 200s on all four;
* the shape — ``items`` length == ``len(Settings.model_fields)`` (one per field),
groups in ``GROUPS`` order (the stable order the E2E pins against),
``totals.count`` / ``overridden`` correct, ``overlay_enabled`` true;
on an empty table a field whose ``BOR_*`` var the test env sets
reports ``source="env"``, an untouched one ``source="default"`` and
none reports ``"db"``;
* write → live — a saved ``relevance_threshold`` is in
``get_settings()`` with NO restart (the cache clear + task-03 hooks
are the whole job), the row is in the table with a non-null
``updated_at``, and the DB value reaches the ANONYMOUS
``/api/config`` endpoint through the same overlay;
* errors — unknown key 404; the two LOCKED A3 keys 422 naming the env
var; a bad shape 422 naming the field with NOTHING stored; a
cross-field pair that only breaks in combination (the second PUT 422s
and the first row is unchanged);
* masking (LOCKED A4) — a stored secret reports ``masked``/``has_value``
only, the plaintext appears NOWHERE in the raw response text, a blank
secret PUT 422s, and a DELETE clears it;
* reset — a DELETE ``/{key}`` restores the env value (idempotent);
DELETE ``""`` clears everything (``{"cleared": 3}``);
* the kill switch (D6) — ``BOR_SETTINGS_DB=0``: ``overlay_enabled``
false, a PUT still stores the row, ``get_settings()`` still reports
the env value — the switch disables the mechanism, not the panel;
* the byte budget — the whole catalogue stays under the phase-167 INFO
bar (64 KB), so it never floods the log;
* fail-soft boot — a malformed row inserted DIRECTLY into the table
(bypassing the API's 422 gate) never bricks: a fresh ``create_app()``
runs with it in the table, the row is skipped with a logged reason,
the rest of the table still applies, and ``/api/health`` is 200.
The module's teardown leaves NO row behind (the same discipline
``test_admin_theme_tab.py`` applies to ``ui_settings`` — a surviving
row would retune every later suite through the overlay).
Real Postgres (``podman compose up -d db``); no LLM involved.
Requires: podman compose up -d db
"""
from __future__ import annotations
import logging
from collections.abc import Iterator
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import text
from sqlalchemy.orm import Session
from app.api import chat as chat_api
from app.config import Settings, get_settings, invalidate_settings
from app.core import settings_overlay
from app.core.settings_catalog import CATALOG, GROUPS
from app.core.settings_overlay import register_invalidator
from app.main import app as fastapi_app
from app.main import create_app
from tests.conftest import ADMIN_PASSWORD
API = "/api/admin/settings"
@pytest.fixture(autouse=True)
def clean_state(
db: Session, monkeypatch: pytest.MonkeyPatch
) -> Iterator[None]:
"""A clean ``app_settings`` table + the overlay switch ON + a fresh
overlay/hook slate (the ``test_admin_settings_live`` idiom: reset
the process-wide hook list and re-wire what ``create_app()``
wires, so a save in this module runs the chat runtime reset).
Teardown deletes the rows + invalidates + clears the cache — NO row
may survive this module (a surviving row would retune every later
suite through the overlay).
"""
settings_overlay.reset_state_for_tests()
register_invalidator(chat_api.reset_runtime_caches)
db.execute(text("DELETE FROM app_settings"))
db.commit()
monkeypatch.setenv("BOR_SETTINGS_DB", "1")
invalidate_settings()
yield
db.execute(text("DELETE FROM app_settings"))
db.commit()
invalidate_settings()
get_settings.cache_clear()
def _admin_client() -> TestClient:
"""A fresh client signed in as the admin (the ``_admin_client``
pattern from ``test_ui_settings_api``)."""
c = TestClient(fastapi_app)
r = c.post("/api/login", json={"password": ADMIN_PASSWORD})
assert r.status_code == 204, f"admin login failed: {r.status_code} {r.text}"
return c
def _token_user_client(admin: TestClient, anon: TestClient) -> TestClient:
"""A signed-in token USER (role ``"user"`` — issue the token the way
``test_tokens_api`` does): the only non-admin identity, 403 on this
surface on every method (the phase-79 token matrix contract)."""
r = admin.post("/api/tokens", json={"label": "alice"})
assert r.status_code == 201, r.text
token = r.json()["token"]
assert anon.post("/api/token-auth", json={"token": token}).status_code == 204
return anon
def _item(body: dict, key: str) -> dict:
"""One catalogue item by key (the GET body's ``items`` list)."""
for item in body["items"]:
if item["key"] == key:
return item
raise AssertionError(f"catalogue item missing: {key}")
def test_auth_matrix_all_four_routes(client: TestClient, db: Session) -> None:
"""Anonymous AND token user: 403 ``admin only`` on ALL four routes
(the router-wide dependency closes the surface on every method);
the admin passes on all four."""
admin = _admin_client()
token_user = _token_user_client(admin, client)
for caller in (client, token_user):
r = caller.get(API)
assert r.status_code == 403 and r.json() == {"detail": "admin only"}
r = caller.put(f"{API}/relevance_threshold", json={"value": "0.5"})
assert r.status_code == 403 and r.json() == {"detail": "admin only"}
r = caller.delete(f"{API}/relevance_threshold")
assert r.status_code == 403 and r.json() == {"detail": "admin only"}
r = caller.delete(API)
assert r.status_code == 403 and r.json() == {"detail": "admin only"}
r = admin.get(API)
assert r.status_code == 200
r = admin.put(f"{API}/top_n_docs", json={"value": "3"})
assert r.status_code == 200, r.text
r = admin.delete(f"{API}/top_n_docs")
assert r.status_code == 200, r.text
r = admin.delete(API)
assert r.status_code == 200, r.text
assert r.json() == {"cleared": 0}
assert db.execute(text("SELECT count(*) FROM app_settings")).scalar() == 0
def test_catalogue_shape_and_sources_on_empty_table(admin_client) -> None:
"""The whole catalogue, grouped and stable: one item per field (exactly
``Settings.model_fields``), groups in ``GROUPS`` order (each group a
contiguous block, the block order = ``GROUPS``), ``totals`` correct,
``overlay_enabled`` true. On an empty table: the env-var-set field
reports ``"env"``, the untouched one ``"default"`` and NOTHING
reports ``"db"``."""
r = admin_client.get(API)
assert r.status_code == 200, r.text
body = r.json()
items = body["items"]
assert len(items) == len(Settings.model_fields)
assert body["totals"] == {
"count": len(Settings.model_fields),
"overridden": 0,
}
assert body["overlay_enabled"] is True
# The stable order: each group a contiguous block, block order GROUPS.
groups_seen: list[str] = []
for item in items:
assert item["group"] in GROUPS
if not groups_seen or groups_seen[-1] != item["group"]:
groups_seen.append(item["group"])
assert groups_seen == [g for g in GROUPS if g in groups_seen]
assert len(groups_seen) == len(set(groups_seen))
# Every item carries the full contract (the panel generates from it).
for item in items:
assert set(item) == {
"key", "env_var", "group", "label", "type", "secret",
"restart_required", "overridable", "source", "value",
"masked", "has_value", "has_override", "db_value", "default",
}
assert item["env_var"] == "BOR_" + item["key"].upper()
assert item["source"] in ("db", "env", "default")
# Empty table: env / default / never db.
sources = {item["key"]: item["source"] for item in items}
assert "db" not in sources.values()
assert sources["input_placeholder"] == "env" # BOR_… set in tests/conftest.py
assert sources["docs_work_dir"] == "default" # no BOR_… in the test env
assert sources["relevance_threshold"] == "env" # pinned 0.30 in conftest
def test_write_is_live_without_restart(admin_client, db: Session) -> None:
"""Write → live (no restart): the saved value is in ``get_settings()``
in the SAME process, the row is in the table with a non-null
``updated_at``, the item reports ``source="db"`` and
``totals.overridden == 1`` — and the DB value reaches the ANONYMOUS
``/api/config`` endpoint through the same overlay (the DB-over-env
rule is deployment-wide, not panel-local)."""
assert get_settings().relevance_threshold == 0.30 # the conftest env value
r = admin_client.put(f"{API}/relevance_threshold", json={"value": "0.5"})
assert r.status_code == 200, r.text
body = r.json()
assert body["key"] == "relevance_threshold"
assert body["source"] == "db"
assert body["has_override"] is True
assert body["db_value"] == "0.5"
assert get_settings().relevance_threshold == 0.5 # NO restart (task 03)
row = db.execute(
text("SELECT value, updated_at FROM app_settings WHERE key = 'relevance_threshold'")
).first()
assert row is not None
assert row[0] == "0.5"
assert row[1] is not None # the server-side now()
r = admin_client.get(API)
assert r.json()["totals"]["overridden"] == 1
# Visible through the public config endpoint (no auth) — the same
# overlay retunes every consumer at once (D2).
anon = TestClient(fastapi_app)
r = anon.get("/api/config")
assert r.status_code == 200
assert r.json()["input_placeholder"] == "Ask me anything…"
r = admin_client.put(f"{API}/input_placeholder", json={"value": "Type your question…"})
assert r.status_code == 200, r.text
r = anon.get("/api/config")
assert r.status_code == 200
assert r.json()["input_placeholder"] == "Type your question…"
def test_put_errors_404_and_422(admin_client, db: Session) -> None:
"""The write-path guards (LOCKED A5, A3, A4): unknown key 404; the
two A3 keys 422 naming the env var; a bad shape 422 naming the FIELD
with NOTHING stored; a cross-field pair that only breaks in
combination — the second PUT 422s and the first row is unchanged."""
r = admin_client.put(f"{API}/bogus_key", json={"value": "1"})
assert r.status_code == 404
assert r.json() == {"detail": "Unknown setting bogus_key"}
r = admin_client.put(f"{API}/database_url", json={"value": "postgresql://x"})
assert r.status_code == 422
assert "BOR_DATABASE_URL is not settable from the database" in r.json()["detail"]
r = admin_client.put(f"{API}/embedding_dim", json={"value": "384"})
assert r.status_code == 422
assert "BOR_EMBEDDING_DIM is not settable from the database" in r.json()["detail"]
# A bad shape: names the field, stores nothing.
r = admin_client.put(f"{API}/relevance_threshold", json={"value": "abc"})
assert r.status_code == 422
assert "relevance_threshold" in r.json()["detail"]
assert db.execute(text("SELECT count(*) FROM app_settings")).scalar() == 0
# The one list[str] field (K2): a malformed JSON array is a 422
# naming the field (the _coerce ValueError branch), nothing stored.
r = admin_client.put(f"{API}/suggestions", json={"value": "[not json"})
assert r.status_code == 422
assert "suggestions" in r.json()["detail"]
assert db.execute(text("SELECT count(*) FROM app_settings")).scalar() == 0
# A well-formed JSON array stores fine (the JSON round-trip works).
r = admin_client.put(
f"{API}/suggestions", json={"value": '["How do I deploy gitea?"]'}
)
assert r.status_code == 200, r.text
assert r.json()["db_value"] == '["How do I deploy gitea?"]'
assert db.execute(
text("SELECT value FROM app_settings WHERE key = 'suggestions'")
).scalar() == '["How do I deploy gitea?"]'
# (Clean up the row — the cross-field pin below counts rows.)
assert admin_client.delete(f"{API}/suggestions").status_code == 200
# The cross-field pair (only breaks in combination): store
# relevance_threshold=0.2 (valid alone), then lexical_support_floor=0.9
# — above the DB-overridden threshold: the SECOND PUT 422s and the
# first row is unchanged.
r = admin_client.put(f"{API}/relevance_threshold", json={"value": "0.2"})
assert r.status_code == 200, r.text
r = admin_client.put(f"{API}/lexical_support_floor", json={"value": "0.9"})
assert r.status_code == 422
assert "lexical_support_floor" in r.json()["detail"]
value = db.execute(
text("SELECT value FROM app_settings WHERE key = 'relevance_threshold'")
).scalar()
assert value == "0.2" # the first row is unchanged
assert db.execute(text("SELECT count(*) FROM app_settings")).scalar() == 1
# The failed write did not disturb the live view.
assert get_settings().relevance_threshold == 0.2
def test_secrets_are_masked_and_write_only(admin_client, db: Session) -> None:
"""LOCKED A4: a stored secret reports ``masked``/``has_value`` only —
``value``/``db_value`` are ``None`` and the plaintext appears NOWHERE
in the raw response text (asserted on ``response.text``, not the
parsed model); a blank secret PUT 422s (a self-lockout — DELETE is
the explicit clear); a DELETE clears it."""
db.execute(
text("INSERT INTO app_settings (key, value) VALUES ('llm_api_key', :v)"),
{"v": "sk-secret-12345"},
)
db.commit()
invalidate_settings()
r = admin_client.get(API)
assert r.status_code == 200
item = _item(r.json(), "llm_api_key")
assert item["masked"] is True
assert item["value"] is None
assert item["db_value"] is None
assert item["has_value"] is True # a non-empty value is in effect
assert item["has_override"] is True
assert item["source"] == "db"
assert item["default"] == "" # masked: the code default is never shown
assert "sk-secret-12345" not in r.text # NOWHERE in the raw body
# The other write-only fields are masked too (set in the test env).
for key in ("admin_password", "session_secret", "database_url"):
other = _item(r.json(), key)
assert other["masked"] is True and other["value"] is None
# A blank secret PUT is a 422 (an empty API key / session secret
# would self-lockout — DELETE is the explicit clear).
r = admin_client.put(f"{API}/llm_api_key", json={"value": " "})
assert r.status_code == 422
assert "BOR_LLM_API_KEY must not be blank" in r.json()["detail"]
assert db.execute(
text("SELECT value FROM app_settings WHERE key = 'llm_api_key'")
).scalar() == "sk-secret-12345" # unchanged
# A DELETE clears the row — the effective value falls back to the
# env value (whatever this deployment's env/.env provides — the
# test only pins that it is NOT the stored secret and that the
# item reports the fallback state).
r = admin_client.delete(f"{API}/llm_api_key")
assert r.status_code == 200
assert r.json()["source"] == "default" # no BOR_LLM_API_KEY in the PROCESS env
assert r.json()["has_override"] is False
assert r.json()["value"] is None
assert r.json()["has_value"] == bool(get_settings().llm_api_key)
assert get_settings().llm_api_key != "sk-secret-12345" # the override is gone
assert db.execute(text("SELECT count(*) FROM app_settings")).scalar() == 0
def test_reset_to_env_and_clear_all(admin_client, db: Session) -> None:
"""Reset semantics: DELETE ``/{key}`` deletes the row (idempotent — a
double Reset is a success, not a 404) and the env value wins again;
DELETE ``""`` clears every row (``{"cleared": 3}``) and
``totals.overridden`` is back to 0."""
r = admin_client.put(f"{API}/relevance_threshold", json={"value": "0.5"})
assert r.status_code == 200, r.text
assert get_settings().relevance_threshold == 0.5
r = admin_client.delete(f"{API}/relevance_threshold")
assert r.status_code == 200, r.text
assert r.json()["source"] == "env" # BOR_RELEVANCE_THRESHOLD is set (0.30)
assert r.json()["has_override"] is False
assert get_settings().relevance_threshold == 0.30 # the env value wins again
# Idempotent: the double Reset is a success, not a 404.
r = admin_client.delete(f"{API}/relevance_threshold")
assert r.status_code == 200, r.text
# Three rows, then "Clear all overrides".
for key, value in (("relevance_threshold", "0.5"), ("llm_retries", "7"), ("top_n_docs", "3")):
r = admin_client.put(f"{API}/{key}", json={"value": value})
assert r.status_code == 200, r.text
assert db.execute(text("SELECT count(*) FROM app_settings")).scalar() == 3
r = admin_client.delete(API)
assert r.status_code == 200
assert r.json() == {"cleared": 3}
assert db.execute(text("SELECT count(*) FROM app_settings")).scalar() == 0
assert get_settings().relevance_threshold == 0.30 # env values win again
r = admin_client.get(API)
assert r.json()["totals"]["overridden"] == 0
def test_kill_switch_disables_the_mechanism_not_the_panel(
admin_client, db: Session, monkeypatch: pytest.MonkeyPatch
) -> None:
"""D6: with ``BOR_SETTINGS_DB=0`` the catalogue reports
``overlay_enabled`` false (the panel shows a banner instead of
pretending a save took effect), a PUT STILL stores the row, and
``get_settings()`` still reports the ENV value — the switch disables
the mechanism, not the panel."""
r = admin_client.put(f"{API}/relevance_threshold", json={"value": "0.5"})
assert r.status_code == 200, r.text
assert get_settings().relevance_threshold == 0.5 # the switch was ON
# Deterministic env base for the pin (machine-independent):
monkeypatch.setenv("BOR_LLM_RETRIES", "3")
invalidate_settings()
monkeypatch.setenv("BOR_SETTINGS_DB", "0")
invalidate_settings()
r = admin_client.get(API)
assert r.status_code == 200
assert r.json()["overlay_enabled"] is False
assert r.json()["totals"]["overridden"] == 0 # no row is IN EFFECT
item = _item(r.json(), "relevance_threshold")
assert item["source"] == "env" # the row exists but is not in effect
assert item["has_override"] is True
assert item["db_value"] == "0.5" # the stored string is visible (non-secret)
assert get_settings().relevance_threshold == 0.30 # the env value stands
# The panel stays writable: the PUT stores the row …
r = admin_client.put(f"{API}/llm_retries", json={"value": "7"})
assert r.status_code == 200, r.text
assert db.execute(text("SELECT count(*) FROM app_settings")).scalar() == 2
# … but the env value stays in effect while the switch is off.
assert get_settings().llm_retries == 3 # the env value (pinned 3 above)
def test_catalogue_stays_under_the_byte_budget(admin_client) -> None:
"""The whole catalogue stays under the phase-167 INFO bar (64 KB —
``BOR_LOG_LARGE_RESPONSE_BYTES``), so ``GET /api/admin/settings``
never floods the log (the whole catalogue ≈ 20 KB; the audit script re-verifies
the 256 KB budget in task 06)."""
r = admin_client.get(API)
assert r.status_code == 200
assert len(r.content) < 65_536
def test_unknown_delete_is_a_404(admin_client) -> None:
"""A DELETE of an UNKNOWN SETTING is a 404 like the PUT (a typo in
the key is a client bug; the idempotency pin is about a known key
whose row is already gone — the double Reset in
``test_reset_to_env_and_clear_all``)."""
r = admin_client.delete(f"{API}/bogus_key")
assert r.status_code == 404
assert r.json() == {"detail": "Unknown setting bogus_key"}
def test_write_log_lines_carry_never_the_value(
admin_client, caplog: pytest.LogCaptureFixture
) -> None:
"""AGENTS rule 10 / A4: one line per write (``settings action=…`` on
the ``app.api.admin_settings`` logger) carrying the key and the
action — NEVER the value (a secret write must not leak the
plaintext into the log)."""
caplog.set_level(logging.INFO, logger="app.api.admin_settings")
r = admin_client.put(f"{API}/llm_api_key", json={"value": "sk-log-secret-xyz"})
assert r.status_code == 200, r.text
set_lines = [rec.message for rec in caplog.records
if rec.name == "app.api.admin_settings"]
assert "settings action=set key=llm_api_key status=200" in set_lines
assert "sk-log-secret-xyz" not in "\n".join(set_lines)
caplog.clear()
r = admin_client.delete(f"{API}/llm_api_key")
assert r.status_code == 200
assert "settings action=clear key=llm_api_key" in [
rec.message for rec in caplog.records
if rec.name == "app.api.admin_settings"
]
caplog.clear()
admin_client.put(f"{API}/top_n_docs", json={"value": "3"})
r = admin_client.delete(API)
assert r.status_code == 200 and r.json() == {"cleared": 1}
assert "settings action=clear_all count=1" in [
rec.message for rec in caplog.records
if rec.name == "app.api.admin_settings"
]
def test_bad_row_inserted_directly_never_bricks_boot(
db: Session, caplog: pytest.LogCaptureFixture
) -> None:
"""Completion criterion (00_phase, item 5): a malformed row written
straight into ``app_settings`` (bypassing the API's 422 gate — the
operator SQL, the failed migration) must never brick boot. A FRESH
``create_app()`` — the full boot path: fail-loud auth gate, resolver
prime, middleware wiring — runs with the bad row present, the row is
SKIPPED with a logged reason, the rest of the table still applies,
and ``/api/health`` is 200 (A5)."""
# One bad row (the API's 422 gate would refuse it) + one good row:
# the bad one is skipped, the good one stands.
db.execute(text(
"INSERT INTO app_settings (key, value)"
" VALUES ('relevance_threshold', 'not-a-number')"
))
db.execute(text(
"INSERT INTO app_settings (key, value) VALUES ('top_n_docs', '3')"
))
db.commit()
caplog.set_level(logging.ERROR, logger="app.settings")
# The boot path with the bad row in the table — must not raise.
app = create_app()
s = get_settings()
env_only = Settings() # the env value the skipped key falls back to
assert s.relevance_threshold == env_only.relevance_threshold
assert s.top_n_docs == 3 # the rest of the table still applies
assert any(
"settings override ignored" in rec.message
and "relevance_threshold" in rec.message
for rec in caplog.records
)
r = TestClient(app).get("/api/health")
assert r.status_code == 200
def test_catalog_covers_every_settings_field(admin_client) -> None:
"""The catalogue's keys are exactly ``Settings.model_fields`` (the
API-level face of the unit completeness pin — a new setting cannot
be missing from the panel's data)."""
r = admin_client.get(API)
assert r.status_code == 200
assert {item["key"] for item in r.json()["items"]} == set(Settings.model_fields)
assert set(CATALOG) == set(Settings.model_fields)
def test_llm_chat_thinking_level_row_contract(
admin_client: TestClient, db: Session, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Phase 178 (task 01; task 02 extends for the summary row): the
renamed ``llm_chat_thinking_level`` row (the phase-172 knob, the
rename per locked A1) behaves exactly like ``llm_chat_model`` —
the machinery is generic over
the catalogue (D4 / A5), so the per-key contract is: (a) present in
the GET body (group ``LLM (aipi)``, ``str``, effective ``low``
with no row in ``app_settings``, ``env_var`` named); (b) PUT stores
the RAW string in ``app_settings`` (the overlay's env-native
string form) and the response reports the new effective value;
(c) the EMPTY value is storable (the off switch, A2/A4 — the API
stores as-is, no allowlist); (d) DELETE resets to the env value;
(e) the non-admin 403 gate applies (asserted once here; the full
four-route matrix is ``test_auth_matrix_all_four_routes``).
``test_catalogue_shape_and_sources_on_empty_table`` +
``test_catalog_covers_every_settings_field`` already pin the row's
PRESENCE generically — these cases pin the per-key behavior.
"""
# A deterministic base (machine-independent, the kill-switch test's
# idiom): the code default, no row in the table.
monkeypatch.delenv("BOR_LLM_CHAT_THINKING_LEVEL", raising=False)
invalidate_settings()
assert get_settings().llm_chat_thinking_level == "low"
# (a) the row: group, type, env var, effective ``low`` with NO row
# in app_settings, not secret, not restart-required (the LLM layer
# reads the setting per request — live).
r = admin_client.get(API)
assert r.status_code == 200, r.text
item = _item(r.json(), "llm_chat_thinking_level")
assert item["group"] == "LLM (aipi)"
assert item["label"] == "Chat thinking level"
assert item["type"] == "str"
assert item["env_var"] == "BOR_LLM_CHAT_THINKING_LEVEL"
assert item["secret"] is False
assert item["restart_required"] is False
assert item["value"] == "low"
assert item["source"] in ("default", "env") # no row in app_settings
assert item["has_override"] is False
assert item["db_value"] is None
assert item["default"] == "low"
# (b) PUT stores the RAW string (the overlay's D1 — env-native
# string form) and the response reports the new effective value.
r = admin_client.put(f"{API}/llm_chat_thinking_level", json={"value": "high"})
assert r.status_code == 200, r.text
assert r.json()["value"] == "high"
assert r.json()["source"] == "db"
assert r.json()["has_override"] is True
assert r.json()["db_value"] == "high"
assert db.execute(
text("SELECT value FROM app_settings WHERE key = 'llm_chat_thinking_level'")
).scalar() == "high" # stored raw, exactly as sent
assert get_settings().llm_chat_thinking_level == "high" # live, no restart
# (c) the EMPTY value is storable (the off switch, A2/A4 — the
# API stores as-is; no allowlist, no value validation): the row
# holds the empty string and the effective value is ``""``.
r = admin_client.put(f"{API}/llm_chat_thinking_level", json={"value": ""})
assert r.status_code == 200, r.text
assert r.json()["value"] == ""
assert r.json()["has_value"] is False # the effective value is empty
assert db.execute(
text("SELECT value FROM app_settings WHERE key = 'llm_chat_thinking_level'")
).scalar() == "" # stored as-is (the off switch)
assert get_settings().llm_chat_thinking_level == ""
# (d) DELETE: the row is gone and the effective value is back to
# the env value (``low`` — the base pinned above).
r = admin_client.delete(f"{API}/llm_chat_thinking_level")
assert r.status_code == 200, r.text
assert r.json()["value"] == "low"
assert r.json()["has_override"] is False
assert db.execute(
text("SELECT count(*) FROM app_settings WHERE key = 'llm_chat_thinking_level'")
).scalar() == 0 # the row is gone
assert get_settings().llm_chat_thinking_level == "low"
# (e) the non-admin 403 gate still applies (asserted once for this
# key — the router-wide dependency is pinned on ALL four routes in
# ``test_auth_matrix_all_four_routes``).
anon = TestClient(fastapi_app)
r = anon.put(f"{API}/llm_chat_thinking_level", json={"value": "high"})
assert r.status_code == 403 and r.json() == {"detail": "admin only"}
def test_llm_summary_thinking_level_row_contract(
admin_client: TestClient, db: Session, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Phase 178 (task 02): the ``llm_summary_thinking_level`` row (the
new one-shot knob) behaves like the chat row — the machinery is
generic over the catalogue (D4 / A4) — with the summary specifics:
(a) present in the GET body (group ``LLM (aipi)``, ``str``, label
``"Summary thinking level"``, ``env_var``
``BOR_LLM_SUMMARY_THINKING_LEVEL``, effective ``""`` with no row in
``app_settings``); (b) PUT stores the RAW string in
``app_settings`` and the value resolves live WITHOUT a restart;
(c) the EMPTY value is storable (the summary off switch, A2 — the
API stores as-is, no allowlist, an empty stored value is never
rejected as "no value"); (d) DELETE resets to the env value
(``""`` — the code default); (e) the non-admin 403 gate applies;
(f) INDEPENDENCE — the two overlays resolve independently (the
whole point of the split): a saved summary value never changes the
chat row's effective value.
"""
# A deterministic base (machine-independent, the kill-switch test's
# idiom): BOTH knobs at their code defaults, no row in the table.
monkeypatch.delenv("BOR_LLM_SUMMARY_THINKING_LEVEL", raising=False)
monkeypatch.delenv("BOR_LLM_CHAT_THINKING_LEVEL", raising=False)
invalidate_settings()
assert get_settings().llm_summary_thinking_level == ""
assert get_settings().llm_chat_thinking_level == "low"
# (a) the row: group, label, type, env var, effective ``""`` with NO
# row in app_settings, not secret, not restart-required (the LLM
# layer reads the setting per request — live).
r = admin_client.get(API)
assert r.status_code == 200, r.text
item = _item(r.json(), "llm_summary_thinking_level")
assert item["group"] == "LLM (aipi)"
assert item["label"] == "Summary thinking level"
assert item["type"] == "str"
assert item["env_var"] == "BOR_LLM_SUMMARY_THINKING_LEVEL"
assert item["secret"] is False
assert item["restart_required"] is False
assert item["value"] == ""
assert item["source"] == "default" # the env var is unset, no row
assert item["has_override"] is False
assert item["db_value"] is None
assert item["default"] == ""
# (b) PUT stores the RAW string (the overlay's D1 — env-native
# string form) and the response reports the new effective value.
r = admin_client.put(f"{API}/llm_summary_thinking_level", json={"value": "high"})
assert r.status_code == 200, r.text
assert r.json()["value"] == "high"
assert r.json()["source"] == "db"
assert r.json()["has_override"] is True
assert r.json()["db_value"] == "high"
assert db.execute(
text("SELECT value FROM app_settings WHERE key = 'llm_summary_thinking_level'")
).scalar() == "high" # stored raw, exactly as sent
assert get_settings().llm_summary_thinking_level == "high" # live, no restart
# (c) the EMPTY value is storable — the summary off switch (A2; the
# API stores as-is; no allowlist, no value validation): an empty
# stored value must not be rejected as "no value"; the row holds
# the empty string and the effective value is ``""``.
r = admin_client.put(f"{API}/llm_summary_thinking_level", json={"value": ""})
assert r.status_code == 200, r.text
assert r.json()["value"] == ""
assert r.json()["has_value"] is False # the effective value is empty
assert db.execute(
text("SELECT value FROM app_settings WHERE key = 'llm_summary_thinking_level'")
).scalar() == "" # stored as-is (the off switch)
assert get_settings().llm_summary_thinking_level == ""
# (d) DELETE: the row is gone and the effective value is back to
# the env value (``""`` — the base pinned above).
r = admin_client.delete(f"{API}/llm_summary_thinking_level")
assert r.status_code == 200, r.text
assert r.json()["value"] == ""
assert r.json()["has_override"] is False
assert db.execute(
text("SELECT count(*) FROM app_settings WHERE key = 'llm_summary_thinking_level'")
).scalar() == 0 # the row is gone
assert get_settings().llm_summary_thinking_level == ""
# (e) the non-admin 403 gate still applies (asserted once for this
# key — the router-wide dependency is pinned on ALL four routes in
# ``test_auth_matrix_all_four_routes``).
anon = TestClient(fastapi_app)
r = anon.put(f"{API}/llm_summary_thinking_level", json={"value": "high"})
assert r.status_code == 403 and r.json() == {"detail": "admin only"}
# (f) INDEPENDENCE — the whole point of the split: save the summary
# row "high" again; the chat row's effective value is STILL "low"
# (the two overlays resolve independently — a saved summary value
# never touches the chat level).
r = admin_client.put(f"{API}/llm_summary_thinking_level", json={"value": "high"})
assert r.status_code == 200, r.text
assert get_settings().llm_summary_thinking_level == "high"
assert get_settings().llm_chat_thinking_level == "low" # untouched
chat_item = _item(admin_client.get(API).json(), "llm_chat_thinking_level")
assert chat_item["value"] == "low" # the panel agrees, too
admin_client.delete(f"{API}/llm_summary_thinking_level") # no row survives
def test_verbatim_display_rows_contract(
admin_client: TestClient, db: Session, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Phase 182 (task 01, LOCKED A8): the three verbatim-display rows are
present in the ``Agent tools`` group with the right labels and types
(``bool`` + two ``int``s, defaults ``true`` / ``12000`` / ``5``) and
behave through the generic overlay machinery:
(a) the GET rows carry group/label/type/env_var, are neither secret nor
restart-required (the chat request path reads them per turn — live);
(b) a PUT stores the RAW string in ``app_settings`` and the value
resolves live WITHOUT a restart;
(c) ``0`` is storable on both budgets — it is INERT, not an error
(``app_settings`` holds ``"0"``, the effective value is ``0``);
(d) a negative value is refused by the validate-before-store gate (A5)
with a 422 naming the field and NOTHING stored — the same rule that
refuses startup; a ``false`` for the master switch is storable
(the kill switch);
(e) a DELETE resets each to the env default;
(f) the non-admin 403 gate applies.
"""
# A deterministic base (machine-independent, the kill-switch test's
# idiom): the code defaults, no row in the table.
for var in (
"BOR_ANSWER_TEXT",
"BOR_SHOW_TEXT_MAX_CHARS",
"BOR_SHOW_TEXT_MAX_BLOCKS",
):
monkeypatch.delenv(var, raising=False)
invalidate_settings()
# (a) the rows.
r = admin_client.get(API)
assert r.status_code == 200, r.text
expect = {
"answer_text": ("Answer text", "bool", "true", "BOR_ANSWER_TEXT"),
"show_text_max_chars": (
"Show text max chars",
"int",
"12000",
"BOR_SHOW_TEXT_MAX_CHARS",
),
"show_text_max_blocks": (
"Show text max blocks",
"int",
"5",
"BOR_SHOW_TEXT_MAX_BLOCKS",
),
}
for key, (label, typ, default, envname) in expect.items():
item = _item(r.json(), key)
assert item["group"] == "Agent tools"
assert item["label"] == label
assert item["type"] == typ
assert item["env_var"] == envname
assert item["secret"] is False
assert item["masked"] is False
assert item["restart_required"] is False
assert item["overridable"] is True
assert item["value"] == default
assert item["default"] == default
assert item["source"] == "default" # env var unset, no row
assert item["has_override"] is False
assert item["db_value"] is None
# (b) PUT stores the raw string and resolves live, no restart.
for key, sent, resolved in (
("answer_text", "false", False),
("show_text_max_chars", "2000", 2000),
("show_text_max_blocks", "2", 2),
):
r = admin_client.put(f"{API}/{key}", json={"value": sent})
assert r.status_code == 200, r.text
assert r.json()["source"] == "db"
assert r.json()["has_override"] is True
assert r.json()["db_value"] == sent
assert db.execute(
text("SELECT value FROM app_settings WHERE key = :k"), {"k": key}
).scalar() == sent # stored raw, exactly as sent
assert getattr(get_settings(), key) == resolved # live, no restart
# (c) ``0`` is storable on both budgets (inert, not an error).
for key in ("show_text_max_chars", "show_text_max_blocks"):
r = admin_client.put(f"{API}/{key}", json={"value": "0"})
assert r.status_code == 200, r.text
assert r.json()["value"] == "0"
assert db.execute(
text("SELECT value FROM app_settings WHERE key = :k"), {"k": key}
).scalar() == "0"
assert getattr(get_settings(), key) == 0
# (d) a negative value is refused before anything is stored (A5) —
# the 422 names the field, the previous row is untouched.
for key in ("show_text_max_chars", "show_text_max_blocks"):
r = admin_client.put(f"{API}/{key}", json={"value": "-1"})
assert r.status_code == 422, r.text
assert key in r.json()["detail"]
assert db.execute(
text("SELECT value FROM app_settings WHERE key = :k"), {"k": key}
).scalar() == "0" # unchanged
assert getattr(get_settings(), key) == 0 # the live view is undisturbed
# (e) DELETE resets each to the env default (the vars are unset here,
# so the code default wins again).
for key, default in (
("answer_text", True),
("show_text_max_chars", 12_000),
("show_text_max_blocks", 5),
):
r = admin_client.delete(f"{API}/{key}")
assert r.status_code == 200, r.text
assert r.json()["has_override"] is False
assert r.json()["source"] == "default"
assert getattr(get_settings(), key) == default
assert db.execute(
text("SELECT count(*) FROM app_settings WHERE key = :k"), {"k": key}
).scalar() == 0
assert db.execute(text("SELECT count(*) FROM app_settings")).scalar() == 0
# (f) the non-admin 403 gate (asserted once for this family — the full
# four-route matrix is ``test_auth_matrix_all_four_routes``).
anon = TestClient(fastapi_app)
r = anon.put(f"{API}/answer_text", json={"value": "false"})
assert r.status_code == 403 and r.json() == {"detail": "admin only"}
def test_docs_token_is_write_only_masked_end_to_end(
admin_client: TestClient, db: Session, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Phase 173 (task 02, C1): ``docs_token`` is a write-only
``secret=True`` row — the EXISTING generic secret contract pinned
for the new key (no new code): (a) before a save the GET reports
the row in the ``Docs push`` group, ``secret=true``,
``value=null`` (masked) and ``has_value=false`` — with
``docs_user`` a plain ``str`` row at ``""``; (b) a PUT stores it
(200) and the GET reports ``has_value=true`` with the plaintext
NOWHERE in the raw body — the core pin: the token text is absent
from the response JSON, the leak class this phase kills; (c) a
blank PUT 422s (the self-lockout guard — ``DELETE`` is the
explicit clear) and the stored value is unchanged; (d) a DELETE
unsets it (reset to the env value — ``""`` by default).
"""
# A deterministic base (machine-independent, the phase-172 test's
# idiom): an explicit empty process value for both docs
# credentials — the process env RANKS ABOVE the operator's local
# ``.env`` (the conftest pin's reasoning), so this pins the code
# defaults no matter what the machine's ``.env`` carries.
monkeypatch.setenv("BOR_DOCS_USER", "")
monkeypatch.setenv("BOR_DOCS_TOKEN", "")
invalidate_settings()
# (a) the rows: docs_token masked + unset, docs_user plain + empty.
r = admin_client.get(API)
assert r.status_code == 200, r.text
item = _item(r.json(), "docs_token")
assert item["group"] == "Docs push"
assert item["label"] == "Docs token"
assert item["type"] == "str"
assert item["env_var"] == "BOR_DOCS_TOKEN"
assert item["secret"] is True
assert item["restart_required"] is False
assert item["overridable"] is True
assert item["masked"] is True
assert item["value"] is None
assert item["db_value"] is None
assert item["has_value"] is False
assert item["has_override"] is False
assert item["default"] == "" # masked: the code default is never shown
user = _item(r.json(), "docs_user")
assert user["group"] == "Docs push"
assert user["label"] == "Docs user"
assert user["secret"] is False
assert user["masked"] is False
assert user["value"] == ""
assert user["has_value"] is False
# (b) a PUT stores it (200, the RAW string in app_settings) and the
# GET reports has_value=true with the plaintext NOWHERE in the raw
# body (asserted on response.text, not the parsed model).
r = admin_client.put(f"{API}/docs_token", json={"value": "bor-docs-tok-123"})
assert r.status_code == 200, r.text
assert r.json()["masked"] is True
assert r.json()["value"] is None
assert r.json()["has_value"] is True
assert r.json()["source"] == "db"
assert db.execute(
text("SELECT value FROM app_settings WHERE key = 'docs_token'")
).scalar() == "bor-docs-tok-123" # stored, exactly as sent
assert get_settings().docs_token == "bor-docs-tok-123" # live, no restart
r = admin_client.get(API)
assert r.status_code == 200
item = _item(r.json(), "docs_token")
assert item["has_value"] is True
assert item["value"] is None
assert item["db_value"] is None
assert "bor-docs-tok-123" not in r.text # NOWHERE in the raw body
# (c) a blank PUT is a 422 (a blank token would self-lockout —
# DELETE is the explicit clear) and the stored value is unchanged.
r = admin_client.put(f"{API}/docs_token", json={"value": " "})
assert r.status_code == 422
assert "BOR_DOCS_TOKEN must not be blank" in r.json()["detail"]
assert db.execute(
text("SELECT value FROM app_settings WHERE key = 'docs_token'")
).scalar() == "bor-docs-tok-123" # unchanged
# (d) a DELETE unsets it: the row is gone and the effective value is
# back to the env value (the pinned "" — the anonymous state, C5).
r = admin_client.delete(f"{API}/docs_token")
assert r.status_code == 200, r.text
assert r.json()["source"] == "env" # the pinned empty BOR_DOCS_TOKEN
assert r.json()["has_override"] is False
assert r.json()["has_value"] is False
assert get_settings().docs_token == "" # the override is gone
assert db.execute(
text("SELECT count(*) FROM app_settings WHERE key = 'docs_token'")
).scalar() == 0 # the row is gone
def test_docs_repo_refuses_embedded_credentials(
admin_client: TestClient, db: Session, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Phase 173 (task 02, C4): a ``docs_repo`` PUT with a
``://<user>:<pass>@`` segment is a 422 that NAMES the Docs user /
Docs token fields (asserted in the detail, not just present) and
stores NOTHING — the leak cannot be re-created from the UI. Clean
``https://`` URLs, local paths and ``ssh://`` remotes pass
untouched; a ``user:``-looking subdomain (no trailing ``@``) does
NOT false-positive. A legacy ``.env`` value with embedded
credentials is NOT a PUT — the task-04 startup migration handles
that path."""
# A deterministic base: explicit empty/valid process values for the
# docs settings — the process env ranks above the operator's local
# ``.env`` (the conftest pin's reasoning), so the base is the code
# defaults no matter what the machine's ``.env`` carries (a local
# ``BOR_DOCS_REPO`` with embedded credentials is EXACTLY the leak
# class under test — it must not leak into this pin).
monkeypatch.setenv("BOR_DOCS_REPO", "")
monkeypatch.setenv("BOR_DOCS_BRANCH", "bor-docs")
invalidate_settings()
# The 422: names the replacement fields, stores nothing.
r = admin_client.put(
f"{API}/docs_repo", json={"value": "https://u:pass@gitea.example.com/o/r.git"}
)
assert r.status_code == 422
detail = r.json()["detail"]
assert "Docs user" in detail
assert "Docs token" in detail
assert db.execute(text("SELECT count(*) FROM app_settings")).scalar() == 0
# The failed write did not disturb the live view.
assert get_settings().docs_repo == ""
# Clean https:// URL: passes (the branch validator is satisfied by
# the non-empty docs_branch default — the repo is set, the token
# shape is valid).
r = admin_client.put(f"{API}/docs_repo", json={"value": "https://gitea.example.com/o/r.git"})
assert r.status_code == 200, r.text
assert r.json()["db_value"] == "https://gitea.example.com/o/r.git"
assert get_settings().docs_repo == "https://gitea.example.com/o/r.git"
assert admin_client.delete(f"{API}/docs_repo").status_code == 200
# A local path: untouched.
r = admin_client.put(f"{API}/docs_repo", json={"value": "/srv/docs-repo"})
assert r.status_code == 200, r.text
assert r.json()["db_value"] == "/srv/docs-repo"
assert admin_client.delete(f"{API}/docs_repo").status_code == 200
# A ``user:``-looking subdomain: the trailing ``@`` requirement keeps
# the regex from false-positiving — passes untouched.
r = admin_client.put(
f"{API}/docs_repo", json={"value": "https://sub.domain.example.com/r.git"}
)
assert r.status_code == 200, r.text
assert r.json()["db_value"] == "https://sub.domain.example.com/r.git"
assert admin_client.delete(f"{API}/docs_repo").status_code == 200
# An ssh:// remote (credentials via SSH keys, never in the URL):
# untouched.
r = admin_client.put(
f"{API}/docs_repo", json={"value": "ssh://git@gitea.example.com/o/r.git"}
)
assert r.status_code == 200, r.text
assert r.json()["db_value"] == "ssh://git@gitea.example.com/o/r.git"
assert admin_client.delete(f"{API}/docs_repo").status_code == 200
assert db.execute(text("SELECT count(*) FROM app_settings")).scalar() == 0