phase: 168_admin_settings_panel
Build and Push Containers / build-and-push-app (push) Successful in 4m4s
Build and Push Containers / build-and-push-db (push) Successful in 15s

**Phase 168 — Admin Settings panel — final verification: PASS**

1. **Implemented/verified:** `/settings.html` (admin-only) with the 75-field catalogue (`app/core/settings_catalog.py`), DB-over-env overrides in Postgres (`app_settings`, migration 0028 at head, `settings_overlay.py` resolver, `app/api/admin_settings.py` — 4 admin-only routes), live-apply on save with no restart + restart-required badges, Reset/Clear all, write-only secrets, `.env`-never-written, `BOR_SETTINGS_DB=0` kill switch.
2. **Phase E2E: 12/12** — run with `E2E_APP_PORT=8200` (ports 8123/8124 are held by an unrelated "undercooked" project; no process killed, no code change — the documented env override).
3. **Regression set: all 7 suites green in isolation** (theme tab, API tokens, configurable brand, global tuning, admin auth, cache busting, wide desktop column — 38 tests).
4. **Full suite: unit+integration exit 0 (4066 passed), app/ coverage 99%** (gate >90%); **ruff clean, pyright 0 errors / 0 warnings.**
5. **Log contract (rule 10) verified live:** per-turn + per-response byte-budget lines present; settings write lines carry key + action, **never the value**.
6. **Byte-budget audit: all endpoints under the 262,144 B budget**; settings catalogue 23,513 B (gzip 3,326).
7. **Defect fixed (one):** criterion 5 had no integration pin — added `test_bad_row_inserted_directly_never_bricks_boot` to `tests/integration/test_admin_settings_api.py` (malformed row inserted directly → fresh `create_app()` boots, row skipped with logged reason, rest of table applies, `/api/health` 200) — 12/12 green.
8. **Docs pass:** `README.md` and `AGENTS.md` reviewed in full — both consistent with the as-built feature (Settings page table, Settings section, Configuration-reference note, `.env.example` block); **no changes required**.
9. **Cleanup:** audit app killed (by PID); `app_settings` left empty (0 rows); all changes left in the working tree (no commit, per protocol).
10. **Completion criteria:** all 7 checkboxes verified green (criterion 7's commit is the harness's).
11. **Next pending phase:** `170_edit_bubble_sizing`.
This commit is contained in:
2026-10-10 01:05:14 -04:00
parent abe7388dda
commit 1200437bf8
79 changed files with 7148 additions and 378 deletions
+9
View File
@@ -35,6 +35,15 @@ SESSION_SECRET = "test-session-secret-0123456789abcdef0123456789abcdef"
os.environ.setdefault("BOR_ADMIN_PASSWORD", ADMIN_PASSWORD)
os.environ.setdefault("BOR_SESSION_SECRET", SESSION_SECRET)
# Phase 168 (task 02): the DB-over-env overlay
# (app/core/settings_overlay) is pinned OFF for every in-process suite
# — the shared test DB must never retune the app under test (the same
# poison-guard reasoning as the phase-140 BOR_PDF_PICTORIAL pin in
# tests/e2e/conftest.py): a row left behind by one suite would
# silently change every later suite's settings. Tests that WANT the
# overlay set BOR_SETTINGS_DB=1 + get_settings.cache_clear() themselves.
os.environ.setdefault("BOR_SETTINGS_DB", "0")
# Phase 61 (defect fix): the app under test must see the code DEFAULTS,
# not an operator's local (gitignored) ``.env`` — ``Settings`` loads
# ``env_file=".env"`` from the repo root, and a machine-specific corpus
+11
View File
@@ -52,6 +52,17 @@ SESSION_SECRET = "e2e-session-secret-0123456789abcdef0123456789abcdef"
# (``env["BOR_PDF_PICTORIAL"] = "1"`` in its module app).
os.environ["BOR_PDF_PICTORIAL"] = "0"
# Phase 168 (task 02): the DB-over-env overlay
# (app/core/settings_overlay) is pinned OFF for every E2E app — the
# cross-suite poison guard (the phase-140 BOR_PDF_PICTORIAL pin, same
# reasoning): a row saved by one suite (tests/e2e/
# test_admin_settings_panel.py opts in with BOR_SETTINGS_DB=1 in its
# own module app fixture — the test_pictorial_pdf.py pattern) would
# otherwise silently retune every later suite's app subprocess. The
# line rides into every app's env through the house ``env =
# dict(os.environ)`` pattern.
os.environ["BOR_SETTINGS_DB"] = "0"
def _wait_http(url: str, timeout: float = 40.0) -> None:
deadline = time.monotonic() + timeout
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,518 @@
"""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)`` (75),
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: 75 items (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 (75 items ≈ 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)
@@ -0,0 +1,217 @@
"""Phase 168 (task 03): the end-to-end proof of "no restart", in-process.
After ``invalidate_settings()`` (the save-side wrapper), the NEXT
request behaves under the new value — pinned for the three settings
that are NOT covered by the cache clear alone or that prove it end
to end: the retrieval honesty gate (``relevance_threshold`` — the
per-use read, ``app/api/chat.py``'s ``plan_turn`` gate), the shared
``LLMClient`` (rebuilt: model + timeout from the new ``Settings``
snapshot, ``app/api/chat.py::get_llm``), and the chat concurrency
semaphore (``chat_max_concurrent``). Plus the kill-switch counterpart
(``BOR_SETTINGS_DB=0`` with rows present → env values are back) and
the ``scripts/``-consumer surface (``import_extension_set``).
The rows are inserted directly through the ``db`` session (the admin
API is task 04); the save itself is ``invalidate_settings()`` —
exactly what task 04's PUT/DELETE call after the write.
Real Postgres (``podman compose up -d db``); no LLM involved.
Requires: podman compose up -d db
"""
from __future__ import annotations
import uuid
from collections.abc import Iterator
from datetime import UTC, datetime
import pytest
from sqlalchemy import text
from sqlalchemy.orm import Session
from app.api import chat as chat_api
from app.config import get_settings, invalidate_settings
from app.core import settings_overlay
from app.core.settings_overlay import register_invalidator
from app.models import Document
from app.rag.retriever import RetrievedChunk
@pytest.fixture(autouse=True)
def _clean_state(
db: Session, monkeypatch: pytest.MonkeyPatch
) -> Iterator[None]:
"""A clean ``app_settings`` table + a fresh overlay/hook slate.
``reset_state_for_tests()`` clears the process-wide hook list (a
prior suite's test may have wiped the hook ``app.main`` registered
at import — re-wire what ``create_app()`` wires: the chat runtime
reset), then the table is emptied and the switch is ON for the
test. Teardown deletes the rows and invalidates (runs the reset
hook — a test-built ``LLMClient``/semaphore snapshot never leaks
into a later suite) + clears the cache; ``monkeypatch`` then
restores ``BOR_SETTINGS_DB`` and the next resolution is the
conftest-pinned env-only view.
"""
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 _save(db: Session, key: str, value: str) -> None:
"""One upserted override row (the panel's Save, minus the API)."""
db.execute(
text(
"INSERT INTO app_settings (key, value) VALUES (:k, :v)"
" ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value"
),
{"k": key, "v": value},
)
db.commit()
def _doc(title: str) -> Document:
return Document(
id=uuid.uuid4(),
source="Homelab",
path=f"{title.lower().replace(' ', '-')}.md",
full_path="/tmp/doc.md",
title=title,
content=f"{title} content body for the gate pin",
summary=None,
content_hash="0" * 64,
created_at=datetime(2024, 6, 15, 12, 0, 0, tzinfo=UTC),
)
def _chunk(doc: Document, score: float, cosine: float) -> RetrievedChunk:
"""A non-FTS candidate: *cosine* is the honesty-gate input."""
return RetrievedChunk(
chunk_id=uuid.uuid4(),
position=0,
content=doc.content[:32],
score=score,
document=doc,
cosine=cosine,
)
# ---------------------------------------------------------------------------
# the per-use read: the A8 gate behaves under the saved threshold
# ---------------------------------------------------------------------------
def test_relevance_threshold_db_value_bites_in_the_gate(db: Session) -> None:
base = get_settings().relevance_threshold
if base >= 0.95:
pytest.skip("env threshold too close to 1.0 to bracket a DB override")
# A DB value over the base; a cosine BETWEEN them — grounded under
# the base, deflected under the DB value (no FTS: fts_hits = 0).
db_value = min(base + 0.2, 0.9)
cosine = (base + db_value) / 2
chunk = _chunk(_doc("Gate Pin"), score=0.1, cosine=cosine)
assert chat_api.plan_turn([chunk], get_settings()).deflected is False
_save(db, "relevance_threshold", str(db_value))
invalidate_settings()
assert get_settings().relevance_threshold == db_value
# the SAME chunk, the SAME function, the overlay-fed settings —
# the gate now deflects (no process restart in between)
assert chat_api.plan_turn([chunk], get_settings()).deflected is True
# ---------------------------------------------------------------------------
# the LLM client snapshot: Save swaps the shared LLMClient
# ---------------------------------------------------------------------------
def test_llm_client_rebuilt_from_saved_values(db: Session) -> None:
_save(db, "llm_chat_model", "gpt-test")
_save(db, "llm_timeout", "7")
invalidate_settings()
llm = chat_api.get_llm()
# the rebuilt client holds the DB-overridden snapshot…
assert llm.settings.llm_chat_model == "gpt-test"
assert llm.settings.llm_timeout == 7.0
# …and its AsyncOpenAI transport was built from it (base_url +
# the timeout's connect value — the way llm.py's __init__ builds
# it; the openai client enforces a trailing slash on base_url)
assert llm._client.base_url == ( # pyright: ignore[reportPrivateUsage]
llm.settings.llm_base_url.rstrip("/") + "/"
)
assert llm._client._client.timeout.connect == 7.0 # pyright: ignore[reportPrivateUsage]
def test_llm_client_repoints_on_the_next_save(db: Session) -> None:
_save(db, "llm_chat_model", "gpt-first")
invalidate_settings()
assert chat_api.get_llm().settings.llm_chat_model == "gpt-first"
# the NEXT save re-points the singleton (not just the first)
_save(db, "llm_chat_model", "gpt-second")
invalidate_settings()
llm = chat_api.get_llm()
assert llm.settings.llm_chat_model == "gpt-second"
# ---------------------------------------------------------------------------
# the chat concurrency semaphore follows the saved cap
# ---------------------------------------------------------------------------
def test_chat_semaphore_follows_saved_cap(db: Session) -> None:
base_cap = get_settings().chat_max_concurrent
first = 3 if base_cap != 3 else 5
_save(db, "chat_max_concurrent", str(first))
invalidate_settings()
assert get_settings().chat_max_concurrent == first
sem = chat_api._get_chat_semaphore()
assert sem._value == first # pyright: ignore[reportPrivateUsage]
# the next save re-points the semaphore too
second = 5 if first == 3 else 3
_save(db, "chat_max_concurrent", str(second))
invalidate_settings()
rebuilt = chat_api._get_chat_semaphore()
assert rebuilt is not sem
assert rebuilt._value == second # pyright: ignore[reportPrivateUsage]
# ---------------------------------------------------------------------------
# the kill switch: rows present, switch off → the env values are back
# ---------------------------------------------------------------------------
def test_kill_switch_returns_env_values(
db: Session, monkeypatch: pytest.MonkeyPatch
) -> None:
base = get_settings().input_placeholder
_save(db, "input_placeholder", "DB-WINS-PLACEHOLDER")
invalidate_settings()
assert get_settings().input_placeholder == "DB-WINS-PLACEHOLDER"
monkeypatch.setenv("BOR_SETTINGS_DB", "0")
invalidate_settings()
assert get_settings().input_placeholder == base
# ---------------------------------------------------------------------------
# the scripts/ consumer: the importer reads the same overlay
# ---------------------------------------------------------------------------
def test_import_extensions_override_reaches_script_consumer(db: Session) -> None:
base = get_settings().import_extension_set
# a REAL change — guard against a dev .env that already carries
# exactly this set (a no-op would still pass the final assert)
if base == {".md", ".txt", ".csv"}:
saved, expect = "md,txt,csv,ini", {".md", ".txt", ".csv", ".ini"}
else:
saved, expect = "md,txt,csv", {".md", ".txt", ".csv"}
_save(db, "import_extensions", saved)
invalidate_settings()
assert get_settings().import_extension_set == expect
@@ -88,6 +88,7 @@ SHELL_BACKED_PAGES = {
"/history.html": "index.html", # phase 76 task 03
"/tokens.html": "index.html", # phase 79 task 06
"/theme.html": "index.html", # phase 91 task 04
"/settings.html": "index.html", # phase 168 task 05
}
+283
View File
@@ -0,0 +1,283 @@
"""Integration: migration 0028 (app_settings) schema contract
(phase 168, task 01).
Drives the **real Alembic engine** against the live dev database
(``podman compose up -d db``), mirroring the house pattern of
``test_migration_0017.py`` (information_schema assertions on the state
the migration must leave). The tests target the 0027 → 0028 step
explicitly so later migrations cannot break them:
* upgrade 0027 → 0028 → the ``app_settings`` table exists with the
full column contract — ``key`` VARCHAR(100) NOT NULL and the
primary key (one row per BARE ``Settings`` field name; the PK is
the uniqueness guard — two rows for one setting would make
"which one wins" undefined), ``value`` TEXT NOT NULL (the env-
native string form; the reset is a DELETE, never an empty value),
``updated_at`` TIMESTAMPTZ NOT NULL with the now() server default
(house style) — while the 0027 schema (``documents.
page_strip_mode``) survives;
* inserted rows round-trip (``relevance_threshold`` / ``"0.55"`` and
``import_extensions`` / ``"md,txt"``);
* a duplicate ``key`` raises ``IntegrityError`` (the PK pin) and a
NULL ``value`` is rejected (NOT NULL);
* downgrade to 0027 → the table is GONE (A13 — reversible), the rest
of the schema (``documents.page_strip_mode``, ``ui_settings``)
survives;
* upgrade back to 0028 → the table is back (round-trip).
The ``alembic`` fixture guarantees the DB ends at head even if a test
fails or the process is interrupted.
"""
from __future__ import annotations
from collections.abc import Iterator
from typing import Any
import pytest
from alembic.config import Config
from sqlalchemy import text
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from alembic import command
from app.db import db_available
@pytest.fixture()
def alembic(db: Session) -> Iterator[Config]:
"""Real Alembic config bound to the dev DB (URL from app settings).
Starts at head (repairs an interrupted earlier run); teardown
upgrades to head no matter what happened, so the dev DB is never
left below head.
"""
if not db_available():
pytest.skip("Postgres not reachable — run `podman compose up -d db` first")
cfg = Config() # no alembic.ini file — env.py gets the URL from app config
cfg.set_main_option("script_location", "alembic")
command.upgrade(cfg, "head")
try:
yield cfg
finally:
command.upgrade(cfg, "head")
def _version(db: Session) -> str | None:
return db.execute(text("SELECT version_num FROM alembic_version")).scalar()
def _table_exists(db: Session, table: str) -> bool:
return (
db.execute(
text("SELECT 1 FROM information_schema.tables WHERE table_name = :t"),
{"t": table},
).scalar()
is not None
)
def _column(db: Session, table: str, column: str) -> tuple[Any, ...] | None:
"""(data_type, is_nullable, column_default, character_maximum_length)
for one table column."""
row = db.execute(
text(
"SELECT data_type, is_nullable, column_default, character_maximum_length"
" FROM information_schema.columns"
" WHERE table_name = :t AND column_name = :c"
),
{"t": table, "c": column},
).fetchone()
return tuple(row) if row is not None else None
def _pk_columns(db: Session, table: str) -> list[str]:
"""The table's PRIMARY KEY columns in ordinal position."""
rows = db.execute(
text(
"SELECT kcu.column_name"
" FROM information_schema.table_constraints tc"
" JOIN information_schema.key_column_usage kcu"
" ON kcu.constraint_name = tc.constraint_name"
" AND kcu.table_name = tc.table_name"
" WHERE tc.table_name = :t"
" AND tc.constraint_type = 'PRIMARY KEY'"
" ORDER BY kcu.ordinal_position"
),
{"t": table},
).fetchall()
return [r[0] for r in rows]
def _clear_rows(db: Session) -> None:
db.execute(text("DELETE FROM app_settings"))
db.commit()
def test_upgrade_to_0028_creates_app_settings(db: Session, alembic: Config) -> None:
"""Upgrade 0027 → 0028: the table exists with the full column
contract (``key`` VARCHAR(100) NOT NULL + the sole PK; ``value``
TEXT NOT NULL; ``updated_at`` TIMESTAMPTZ NOT NULL with the now()
server default), and the table is ABSENT at 0027 while the 0027
``documents`` schema survives the upgrade."""
command.downgrade(alembic, "0027") # start from the pre-0028 state
assert _version(db) == "0027"
assert not _table_exists(db, "app_settings"), (
"app_settings must be absent at 0027"
)
command.upgrade(alembic, "0028")
assert _version(db) == "0028", "alembic_version must be at 0028"
assert _table_exists(db, "app_settings"), "the table must exist at 0028"
key = _column(db, "app_settings", "key")
assert key is not None, "app_settings.key is missing"
assert key[0] == "character varying", "key must be VARCHAR"
assert key[1] == "NO", "key must be NOT NULL (PK)"
assert key[2] is None, "key must have no server default"
assert key[3] == 100, "key must be String(100)"
value = _column(db, "app_settings", "value")
assert value is not None, "app_settings.value is missing"
assert value[0] == "text", "value must be TEXT (the env-native string form)"
assert value[1] == "NO", "value must be NOT NULL (reset is a DELETE, never empty)"
updated = _column(db, "app_settings", "updated_at")
assert updated is not None, "app_settings.updated_at is missing"
assert updated[0] == "timestamp with time zone", "updated_at must be TIMESTAMPTZ"
assert updated[1] == "NO", "updated_at must be NOT NULL"
assert updated[2] is not None and "now" in str(updated[2]), (
"updated_at must carry the now() server default (house style)"
)
assert _pk_columns(db, "app_settings") == ["key"], (
"key must be the sole primary key (the uniqueness guard —"
" two rows for one setting make 'which one wins' undefined)"
)
# The 0027 schema survives the additive upgrade.
strip = _column(db, "documents", "page_strip_mode")
assert strip is not None, "documents.page_strip_mode (0027) must survive"
assert strip[3] == 16, "page_strip_mode must stay String(16)"
def test_inserted_rows_round_trip(db: Session, alembic: Config) -> None:
"""At 0028, the env-native string forms round-trip verbatim
(``relevance_threshold`` / ``"0.55"``, ``import_extensions`` /
``"md,txt"``) and ``updated_at`` is stamped by the server default."""
command.upgrade(alembic, "head")
try:
db.execute(
text(
"INSERT INTO app_settings (key, value)"
" VALUES ('relevance_threshold', '0.55')"
)
)
db.execute(
text(
"INSERT INTO app_settings (key, value)"
" VALUES ('import_extensions', 'md,txt')"
)
)
db.commit()
rows = db.execute(
text(
"SELECT key, value, updated_at"
" FROM app_settings ORDER BY key"
)
).fetchall()
assert len(rows) == 2, "both rows must be stored"
assert rows[0][0] == "import_extensions", "key must round-trip verbatim"
assert rows[0][1] == "md,txt", "the CSV string form must round-trip"
assert rows[0][2] is not None, "updated_at must be stamped (server default)"
assert rows[1][0] == "relevance_threshold"
assert rows[1][1] == "0.55", "the numeric string form must round-trip"
finally:
_clear_rows(db)
def test_duplicate_key_is_rejected(db: Session, alembic: Config) -> None:
"""The PK is the uniqueness guard: two rows for one setting would
make 'which one wins' undefined, so the second insert must raise
``IntegrityError``."""
command.upgrade(alembic, "head")
try:
db.execute(
text("INSERT INTO app_settings (key, value) VALUES ('llm_retries', '3')")
)
db.commit()
with pytest.raises(IntegrityError):
db.execute(
text(
"INSERT INTO app_settings (key, value)"
" VALUES ('llm_retries', '5')"
)
)
db.rollback() # the IntegrityError aborts the open transaction
finally:
_clear_rows(db)
def test_null_value_is_rejected(db: Session, alembic: Config) -> None:
"""``value`` is NOT NULL: 'clear the override' is a DELETE of the
row, never a NULL/empty value, so the resolver never has to
distinguish 'no override' from 'override to empty'."""
command.upgrade(alembic, "head")
try:
with pytest.raises(IntegrityError):
db.execute(
text(
"INSERT INTO app_settings (key, value)"
" VALUES ('chat_max_concurrent', NULL)"
)
)
db.rollback() # the IntegrityError aborts the open transaction
stored = db.execute(
text("SELECT count(*) FROM app_settings")
).scalar()
assert stored == 0, "the rejected row must not be stored"
finally:
_clear_rows(db)
def test_downgrade_to_0027_drops_the_table(db: Session, alembic: Config) -> None:
"""Downgrade 0028 → 0027: the table is gone (A13 — fully
reversible) while the rest of the schema survives (the 0027
``documents.page_strip_mode`` column, ``ui_settings``,
``api_tokens``). The values are re-enterable from the panel —
they are an override of the env, not a source of truth."""
command.downgrade(alembic, "0027")
assert _version(db) == "0027"
assert not _table_exists(db, "app_settings"), "app_settings must be dropped"
strip = _column(db, "documents", "page_strip_mode")
assert strip is not None and strip[3] == 16, (
"documents.page_strip_mode (0027) must survive the downgrade"
)
semantic = _column(db, "ui_settings", "ok_bg")
assert semantic is not None and semantic[3] == 7, (
"ui_settings.ok_bg must survive the downgrade"
)
token_col = _column(db, "api_tokens", "token_hash")
assert token_col is not None and token_col[0] == "character varying", (
"api_tokens.token_hash must survive the downgrade"
)
def test_upgrade_round_trip_restores_the_table(db: Session, alembic: Config) -> None:
"""Downgrade to 0027, then upgrade back to 0028: the table is back
with the column contract and PK intact (A13 round-trip)."""
command.downgrade(alembic, "0027")
command.upgrade(alembic, "0028")
assert _version(db) == "0028", "round-trip upgrade must land at 0028"
assert _table_exists(db, "app_settings"), "the table must be back"
key = _column(db, "app_settings", "key")
assert key is not None, "app_settings.key must be back"
assert key[0] == "character varying", "key must be VARCHAR after the round-trip"
assert key[1] == "NO", "key must be NOT NULL after the round-trip"
assert key[3] == 100, "key must be String(100) after the round-trip"
value = _column(db, "app_settings", "value")
assert value is not None, "app_settings.value must be back"
assert value[0] == "text", "value must be TEXT after the round-trip"
assert value[1] == "NO", "value must be NOT NULL after the round-trip"
assert _pk_columns(db, "app_settings") == ["key"]
+174
View File
@@ -0,0 +1,174 @@
"""Unit: the admin settings router's pure helpers (phase 168, task 04)
— the env-native stringification (``_display``), the control-kind
derivation (``_field_type``), the ``_item`` builder's masked and
default branches, and the LOCKED A1 static pins: the module never
opens a file (``.env`` is never written), never shells out and never
``eval``s a value.
The DB-driven behavior (the four routes, the candidate-set validation,
the auth matrix) is pinned in
``tests/integration/test_admin_settings_api.py``.
"""
from __future__ import annotations
import pathlib
import re
import pytest
from app.api.admin_settings import _display, _field_type, _item
from app.config import Settings
from app.core.settings_overlay import LIST_FIELDS
def test_display_bools_are_the_env_spelling() -> None:
"""bool → ``"true"``/``"false"`` — checked BEFORE int (``isinstance(True, int)``
is ``True`` — a bool field must never stringify through ``str()``'s
``"True"``)."""
assert _display("stream_thinking", True) == "true"
assert _display("stream_thinking", False) == "false"
# An int value for an int field is the plain str (no bool confusion).
assert _display("llm_retries", 3) == "3"
def test_display_list_field_is_json() -> None:
"""The one ``list[str]`` field (``suggestions``) stringifies as the
JSON array the env spelling would carry (K2: stored in the same
JSON form, pydantic parses it exactly like the env var)."""
assert set(LIST_FIELDS) == {"suggestions"}
assert _display("suggestions", ["a", "b"]) == '["a", "b"]'
assert _display("suggestions", []) == "[]"
def test_display_scalars_are_plain_strings() -> None:
"""Everything else is ``str()`` — floats, ints, strings, in the
env-native form (what ``app_settings`` stores, D1)."""
assert _display("relevance_threshold", 0.55) == "0.55"
assert _display("llm_timeout", 7.0) == "7.0"
assert _display("app_name", "Brain of Reese") == "Brain of Reese"
assert _display("import_extensions", "md,txt") == "md,txt"
def test_field_type_covers_all_kinds() -> None:
"""The panel control kind per annotation: ``bool`` before ``int``
(a bool annotation is never misread as int), the one list field is
``json``, the rest fall through to their annotation."""
assert _field_type("stream_thinking") == "bool"
assert _field_type("llm_retries") == "int"
assert _field_type("relevance_threshold") == "float"
assert _field_type("app_name") == "str"
assert _field_type("suggestions") == "json"
# Every field resolves to one of the five wire kinds.
kinds = {"int", "float", "bool", "str", "json"}
for key in Settings.model_fields:
assert _field_type(key) in kinds
def test_item_masks_secret_fields() -> None:
"""LOCKED A4: a secret field's item is write-only on the wire —
``masked=True``, ``value=None``, ``db_value=None``,
``default=""`` (the code default is never shown for a secret),
``has_value`` = is a non-empty value in effect at all (the test
env sets a non-empty admin password / session secret)."""
effective = Settings()
assert effective.admin_password # tests/conftest.py pins it non-empty
item = _item("admin_password", {}, effective)
assert item.masked is True
assert item.value is None
assert item.db_value is None
assert item.default == ""
assert item.has_value is True
assert item.secret is True
assert item.key == "admin_password"
assert item.env_var == "BOR_ADMIN_PASSWORD"
# A stored row for a secret field never exposes the text either.
# (The effective value is forced via an init kwarg — the operator's
# local `.env` may legitimately carry a key, so the test does not
# assume a specific effective value.)
item = _item("llm_api_key", {"llm_api_key": "sk-x"}, Settings(llm_api_key="sk-x"))
assert item.masked is True
assert item.value is None
assert item.db_value is None
assert item.has_value is True
assert item.has_override is True
item = _item("llm_api_key", {}, Settings(llm_api_key=""))
assert item.masked is True
assert item.value is None
assert item.db_value is None
assert item.has_value is False # no value in effect at all
def test_item_reports_env_default_and_db_sources(
monkeypatch: pytest.MonkeyPatch
) -> None:
"""The ``source`` badge: ``db`` for an in-effect overridable row,
``env`` when the ``BOR_*`` var is in the process environment,
``default`` otherwise — and a non-overridable row is NEVER an
override (LOCKED A3: the overlay drops it, so a hand-inserted
``database_url`` row is not in effect)."""
import os
effective = Settings()
# "env": tests/conftest.py sets BOR_RELEVANCE_THRESHOLD.
assert os.environ.get("BOR_RELEVANCE_THRESHOLD") is not None
item = _item("relevance_threshold", {}, effective)
assert item.source == "env"
# The value is the EFFECTIVE float stringified ("0.30" → 0.3 → "0.3").
assert item.value == str(float(os.environ["BOR_RELEVANCE_THRESHOLD"]))
assert item.has_override is False
# "default": no BOR_DOCS_WORK_DIR in the test environment.
assert os.environ.get("BOR_DOCS_WORK_DIR") is None
item = _item("docs_work_dir", {}, effective)
assert item.source == "default"
assert item.value == effective.docs_work_dir
# "db": the row is in effect only while the overlay switch is ON
# (the conftest pin is 0 — flip it for the pin, it restores after).
monkeypatch.setenv("BOR_SETTINGS_DB", "1")
item = _item("relevance_threshold", {"relevance_threshold": "0.9"}, effective)
assert item.source == "db"
assert item.has_override is True
# Non-overridable: a stored row is not an override (LOCKED A3) —
# even with the switch on.
item = _item("database_url", {"database_url": "postgresql://x"}, effective)
assert item.has_override is False
assert item.masked is True
# The code default stringifies (non-secret fields).
item = _item("llm_retries", {}, effective)
assert item.default == "3"
assert item.value == "3"
assert item.db_value is None
def test_item_db_value_is_the_stored_string() -> None:
"""A stored non-secret override exposes ``db_value`` (the stored
string, D1) and ``has_override`` — even while the overlay switch
is OFF (the row is stored but not in effect: the badge is
env/default, the panel's ``overlay_enabled`` flag explains the
rest)."""
effective = Settings()
item = _item("llm_retries", {"llm_retries": "7"}, effective)
assert item.has_override is True
assert item.db_value == "7"
assert item.source == "default" # switch off (the conftest pin): not in effect
def test_module_never_writes_files_or_shells_out() -> None:
"""LOCKED A1 + the module's third invariant (the unit pin): the
router never opens a file (``.env`` is never written by any code
path), never shells out and never ``eval``s a value. (The docstring
legitimately mentions ``.env`` — the pins are on the CALLS, the
``test_settings_overlay`` house style.)"""
source = (
pathlib.Path(__file__).resolve().parents[2]
/ "app"
/ "api"
/ "admin_settings.py"
).read_text()
assert re.search(r"\bopen\s*\(", source) is None, (
"app/api/admin_settings.py must never open() a file (LOCKED A1: "
".env is never written)"
)
assert "subprocess" not in source
assert "os.system" not in source
assert re.search(r"\beval\s*\(", source) is None
assert "import shutil" not in source and "shutil." not in source
+1
View File
@@ -218,6 +218,7 @@ def test_html_pages_include_history() -> None:
"/history.html",
"/tokens.html", # phase 79 task 06: the admin tokens page (shell route)
"/theme.html", # phase 91 task 04: the admin theme page (shell route)
"/settings.html", # phase 168 task 05: the admin settings page (shell route)
"/shared.html", # phase 51: the shared page's static path
"/doc-edit.html", # phase 59: the doc edit screen (task 06)
):
+13 -1
View File
@@ -128,8 +128,11 @@ def test_view_map_covers_the_shell_paths() -> None:
assert '"/theme.html": "theme"' in view_body, (
"phase 91 task 04 folds the Theme view into the shell"
)
assert '"/settings.html": "settings"' in view_body, (
"phase 168 task 05 folds the Settings view into the shell"
)
# The view names are the #view-<name> section slugs in index.html.
for name in ("chat", "tuning", "history", "tokens", "theme"):
for name in ("chat", "tuning", "history", "tokens", "theme", "settings"):
assert f'id="view-{name}"' in _html(), f"missing the #view-{name} section"
@@ -226,6 +229,10 @@ def test_only_non_chat_views_have_lazy_modules() -> None:
assert 'theme: () => import("./theme.js")' in mods_body, (
"the Theme view module is lazy-imported on first show (phase 91)"
)
assert 'settings: () => import("./settings.js")' in mods_body, (
"the Settings view module is lazy-imported on first show "
"(phase 168, a static specifier the bundle inlines)"
)
assert '"chat"' not in mods_body, "the chat view has no lazy module"
assert 'import("./app.js")' not in js, "app.js must never be lazy-imported"
@@ -288,6 +295,11 @@ def test_router_writes_active_state_title_and_meta() -> None:
"Set the palette and branding — the theme is baked into every served page, "
"live on the first paint."
) in js
assert 'settings: "Settings · Brain of Reese"' in js
assert (
"Edit any application setting — a saved value wins over the environment, "
"live on save; reset hands control back."
) in js
# The brand composition (phase 39's window.BOR_BRAND, read at
# write time — never a hardcoded stamp).
assert 'window.BOR_BRAND || "Brain of Reese"' in js
+21
View File
@@ -4,6 +4,8 @@ from __future__ import annotations
from fastapi.testclient import TestClient
import app.main as main_mod
from app.config import get_settings
from app.core import settings_overlay
def test_create_app_warns_and_serves_api_only_without_static_dir(
@@ -30,3 +32,22 @@ def test_create_app_wires_caching_middleware() -> None:
# reach for ``__name__`` the same way starlette's own __repr__ does.
middleware_names = [getattr(mw.cls, "__name__", "") for mw in app2.user_middleware]
assert "CachingMiddleware" in middleware_names
def test_create_app_boots_with_the_switch_on(monkeypatch) -> None:
"""Phase 168 (task 02/03): the boot prime with the kill switch ON
— the ``settings overrides loaded count=…`` branch (the OFF branch
is exercised by every other suite's app import; the overlay's
engine is fail-soft, so a DB-down unit environment still boots).
``create_app`` also wires the chat runtime reset hook here (task
03) — the ``register_invalidator`` pin lives in
``test_settings_live_apply.py``."""
settings_overlay.reset_state_for_tests()
monkeypatch.setenv("BOR_SETTINGS_DB", "1")
app2 = main_mod.create_app()
assert app2 is not None
# restore: the switch reverts via monkeypatch; the cache clear
# ensures a boot-time row set never leaks into later suites'
# settings, and the hook slate is fresh for them too.
settings_overlay.reset_state_for_tests()
get_settings.cache_clear()
+100
View File
@@ -0,0 +1,100 @@
"""Phase 168 (task 02): the settings catalogue is complete and consistent.
The Settings panel (task 05) is GENERATED from
``app/core/settings_catalog.py`` — never hand-written — so these pins
are load-bearing: ``set(CATALOG) == set(Settings.model_fields)`` fails
the build the moment a new ``Settings`` field lands without its
metadata (it can never silently disappear from the panel), the group
table is closed both directions, and the capability sets are exactly
the LOCKED decisions (A3/A4).
"""
from __future__ import annotations
from app.config import Settings
from app.core.settings_catalog import (
CATALOG,
GROUPS,
NON_OVERRIDABLE,
RESTART_REQUIRED,
SECRETS,
SettingMeta,
env_var,
)
def test_catalog_covers_exactly_model_fields() -> None:
"""The completeness pin (both directions, the diff in the message)."""
catalog_keys = set(CATALOG)
field_names = set(Settings.model_fields)
missing = sorted(field_names - catalog_keys)
extra = sorted(catalog_keys - field_names)
assert not missing, (
f"Settings fields missing from the catalogue (invisible in the "
f"panel): {missing}"
)
assert not extra, (
f"Catalogue entries with no Settings field (stale): {extra}"
)
assert set(CATALOG) == set(Settings.model_fields)
def test_groups_closed_both_directions() -> None:
catalog_groups = {meta.group for meta in CATALOG.values()}
only_in_groups = sorted(set(GROUPS) - catalog_groups)
only_in_catalog = sorted(catalog_groups - set(GROUPS))
assert not only_in_groups, f"GROUPS entries with no catalogue member: {only_in_groups}"
assert not only_in_catalog, f"Catalogue groups missing from GROUPS: {only_in_catalog}"
def test_every_meta_carries_its_own_key() -> None:
for key, meta in CATALOG.items():
assert isinstance(meta, SettingMeta)
assert meta.key == key
def test_secrets_are_exactly_the_masked_set() -> None:
# LOCKED A4: write-only, masked — the 3 named secrets +
# database_url (read-only, but its URL carries credentials).
assert set(SECRETS) == {
"llm_api_key",
"admin_password",
"session_secret",
"database_url",
}
def test_non_overridable_is_exactly_locked_a3() -> None:
# LOCKED A3: database_url builds the engine that would read the
# override (chicken-and-egg brick); embedding_dim is fixed in the
# chunks table by A6 (a dim override would corrupt retrieval).
assert set(NON_OVERRIDABLE) == {"database_url", "embedding_dim"}
def test_restart_required_is_the_boot_wired_set() -> None:
# LOCKED A4: logging, the gzip middleware (compress_min_bytes), the
# static mount (static_dir), create_engine (db_pool_*), and
# SessionMiddleware (session_*).
assert set(RESTART_REQUIRED) == {
"log_level",
"compress_min_bytes",
"static_dir",
"db_pool_size",
"db_pool_max_overflow",
"db_pool_recycle",
"session_secret",
"session_max_age",
"session_cookie",
}
def test_env_var_is_bare_key_upper_prefixed() -> None:
assert env_var("relevance_threshold") == "BOR_RELEVANCE_THRESHOLD"
for key in CATALOG:
assert env_var(key).startswith("BOR_")
def test_labels_nonempty_and_unique() -> None:
labels = [meta.label for meta in CATALOG.values()]
assert all(label.strip() for label in labels), "empty label in the catalogue"
duplicates = sorted({label for label in labels if labels.count(label) > 1})
assert not duplicates, f"duplicate labels: {duplicates}"
+608
View File
@@ -0,0 +1,608 @@
"""Unit: the phase-168 Settings view wiring (task 05) — the static
source-level pins for the shell's eighth view.
The browser behavior is E2E-gated (the phase-168 story suite —
``test_admin_settings_panel.py``, task 06); here we pin the
source-level invariants the "views of one document" architecture
depends on, so a silent regression is caught without a browser
(house pattern: ``tests/unit/test_thinking_scroll.py`` /
``test_frontend_router.py`` read the sources and assert on their
mechanisms).
Pinned (task 05 spec):
* ``index.html`` carries the ``#view-settings`` section (ship-hidden
+ inert + focusable — the WCAG pair, AGENTS.md rule 5) with the
``#settings-gate`` (ship-hidden, its Sign in returning to the
Settings view), the ship-hidden ``#settings-content`` holding the
``#settings-disabled`` kill-switch banner (D6), the ONE
``role="status"`` region, the ``role="alert"`` error line, the
client-side search input, the ``.table-wrap`` catalogue with the
visually-hidden caption + the four-column thead, and the action
row (``#settings-save`` — ships disabled — ``#settings-clear-all``
+ the aria-hidden ``#settings-dirty`` hint); the ``#nav-settings``
link ships hidden (admin-only).
* ``router.js`` maps ``/settings.html`` in ALL THREE tables
(VIEW / VIEW_PATH / VIEW_MODULES — a static specifier so the
Containerfile's esbuild stage inlines it into the router bundle)
+ the per-view title/description (the router is the SINGLE writer
of both).
* ``app/main.py``'s shell tuple and ``app/core/caching.py``'s
``HTML_PAGES`` both list ``/settings.html`` (the no-cache +
``?v=`` contract, A19).
* ``header.js`` reveals ``#nav-settings`` keyed off ``admin`` (NOT
``signedIn`` — a token user never sees it), the same two-line
null-safe pattern as the other admin-only links.
* ``settings.js`` builds the DOM with ``createElement`` /
``textContent`` — NO ``innerHTML`` (template-literal or not) and
NO ``http(s)://`` reference (rule 6 — everything local; the
no-HTML-strings house rule).
* ``styles.css`` declares ``.settings-table`` + the three
``.settings-src.is-*`` variants + the dirty-row marker and adds
NO new ``:root`` variable (the phase-156 B3 rule — the badges
are solid existing tokens).
* the FOUR pages that carry the Theme link (index / document /
login / shared) now carry the Settings link too (the phase-34
one-bar contract — test_nav_consistency pins the parity).
"""
from __future__ import annotations
import re
from pathlib import Path
FRONTEND = Path(__file__).resolve().parents[2] / "frontend"
ASSETS = FRONTEND / "assets"
REPO = Path(__file__).resolve().parents[2]
def _read(path: Path) -> str:
assert path.is_file(), f"missing {path}"
return path.read_text(encoding="utf-8")
def _html() -> str:
return _read(FRONTEND / "index.html")
# ---------- the shell markup: the #view-settings section ----------
def test_settings_view_section_ships_hidden_inert_focusable() -> None:
"""The #view-settings section lives in the shell's single main,
ships hidden AND inert (the WCAG a11y pair — a hidden view must
not receive focus or keyboard traversal), and is focusable
(tabindex="-1" — the router focuses it on user-initiated
switches)."""
html = _html()
view = html.find('<section class="view" id="view-settings"')
assert view != -1, "the #view-settings section must be in the shell"
tag_end = html.find(">", view)
tag = html[view:tag_end]
assert "hidden" in tag and "inert" in tag, (
"the folded view ships hidden AND inert"
)
assert 'tabindex="-1"' in tag, "the target view is focusable"
assert 'aria-label="Settings"' in tag, "the view is labelled"
main_end = html.find("</main>", view)
assert view < main_end, "the view section lives inside the single main"
# The 72rem .container — no new width token (PLAN §7.1).
body = html[view:main_end]
assert 'class="container settings-shell"' in body, (
"the view rides the standard 72rem .container (no new width)"
)
def test_settings_page_head_states_the_rule_in_the_owner_terms() -> None:
"""The page-head: <h1>Settings</h1> + a .page-sub that states
the rule in the owner's terms — the value is stored in the
database and wins over the environment variables; clearing hands
control back to the environment (and the .env file is never
written — LOCKED A1)."""
html = _html()
view = html.find('<section class="view" id="view-settings"')
main_end = html.find("</main>", view)
body = html[view:main_end]
assert "<h1>Settings</h1>" in body
sub = re.search(r'<p class="page-sub">([\s\S]*?)</p>', body)
assert sub, "the .page-sub must exist"
text = " ".join(sub.group(1).split())
assert "stored in the database" in text, (
"the sub states: the value is stored in the database"
)
assert "win over" in text, "the sub states: it wins over the env vars"
assert "hands control back to the environment" in text, (
"the sub states: clearing hands control back to the environment"
)
def test_settings_gate_is_the_tokens_gate_shape_ships_hidden() -> None:
"""The anonymous sign-in gate: the EXACT #tokens-gate shape —
the .sources-gate visual language, ship-hidden, its heading
naming what is admin-only, and its Sign in returning to the
Settings view (?next=/settings.html — the no-JS fallback)."""
html = _html()
view = html.find('<section class="view" id="view-settings"')
main_end = html.find("</main>", view)
body = html[view:main_end]
gate = re.search(r'<section[^>]*id="settings-gate"[^>]*>', body)
assert gate, "the #settings-gate must exist"
assert "hidden" in gate.group(0), "#settings-gate must ship hidden"
assert 'class="sources-gate"' in gate.group(0), (
"the gate reuses the .sources-gate visual language"
)
assert 'aria-labelledby="settings-gate-title"' in gate.group(0)
assert "<h2 id=\"settings-gate-title\">Sign in to change settings</h2>" in body
assert 'href="/login.html?next=/settings.html"' in body, (
"the gate's Sign in returns to the Settings view (no-JS fallback)"
)
def test_settings_content_scaffold_and_feedback_regions() -> None:
"""The ship-hidden #settings-content holds, in order: the
#settings-disabled kill-switch banner (role=alert, ship-hidden —
D6: the mechanism is off, not the panel), the ONE status region
(role=status, aria-live=polite — the #tokens-status precedent),
the role=alert error line (ship-hidden — the #theme-error
precedent), the search row (label + type=search input), the
.table-wrap catalogue (the visually-hidden caption + the
four-column thead with the visually-hidden Actions header), and
the action row (Save — ships DISABLED — Clear all + the
aria-hidden dirty hint)."""
html = _html()
view = html.find('<section class="view" id="view-settings"')
main_end = html.find("</main>", view)
body = html[view:main_end]
content = re.search(r'<div[^>]*id="settings-content"[^>]*>', body)
assert content and "hidden" in content.group(0), (
"#settings-content must ship hidden (anonymous-safe)"
)
banner = re.search(
r'<div[^>]*class="kb-banner is-error"[^>]*id="settings-disabled"[^>]*>'
r"|<div[^>]*id=\"settings-disabled\"[^>]*>",
body,
)
assert banner, "the #settings-disabled banner must exist"
btag_match = re.search(r'<div[^>]*id="settings-disabled"[^>]*>', body)
assert btag_match, "the #settings-disabled banner tag must match"
btag = btag_match.group(0)
assert 'role="alert"' in btag and "hidden" in btag, (
"the banner is a ship-hidden role=alert"
)
status = re.search(
r'<span[^>]*id="settings-status"[^>]*>', body
)
assert status, "the #settings-status region must exist"
stag = status.group(0)
assert 'role="status"' in stag and 'aria-live="polite"' in stag, (
"the ONE status region (the #tokens-status precedent)"
)
error = re.search(r'<p[^>]*id="settings-error"[^>]*>', body)
assert error, "the #settings-error line must exist"
etag = error.group(0)
assert 'role="alert"' in etag and "hidden" in etag, (
"the error line is a ship-hidden role=alert"
)
# The search row: the label + the type=search input (client-side,
# the phase-101 D4 contract — no fetch).
assert '<label for="settings-filter">Search settings</label>' in body
filter_input = re.search(
r'<input[^>]*id="settings-filter"[^>]*>', body, re.S
)
assert filter_input, "the #settings-filter input must exist"
assert 'type="search"' in filter_input.group(0), (
"the filter is a type=search input"
)
# The scroll card: the .table-wrap + the visually-hidden caption
# + the four-column thead (the Actions header visually-hidden —
# the row buttons carry their own aria-labels).
assert 'class="table-wrap"' in body, (
"the catalogue rides the .table-wrap card (the phase-156 "
"bounded scroll + sticky header contract)"
)
table = re.search(r'<table[^>]*class="settings-table"[^>]*>', body)
assert table, "the .settings-table must exist"
thead = body[body.find('class="settings-table"') : body.find("</thead>")]
assert '<caption class="visually-hidden">' in thead, (
"the caption is visually-hidden (the phase-86 table pattern)"
)
assert thead.count('<th scope="col">') == 4, (
"the table is four columns (Setting | Value | Source | Actions)"
)
assert '<th scope="col">Setting</th>' in thead
assert '<th scope="col">Value</th>' in thead
assert '<th scope="col">Source</th>' in thead
assert (
'<th scope="col"><span class="visually-hidden">Actions</span></th>'
in thead
), "the Actions header is visually-hidden (the row buttons are labelled)"
# The action row: Save (type=button, ships DISABLED — no dirty
# rows yet), Clear all (type=button), the aria-hidden dirty hint.
save = re.search(r'<button[^>]*id="settings-save"[^>]*>', body)
assert save, "the #settings-save button must exist"
stag = save.group(0)
assert 'type="button"' in stag and "disabled" in stag, (
"Save is a type=button that ships disabled (no dirty rows)"
)
clear_all = re.search(r'<button[^>]*id="settings-clear-all"[^>]*>', body)
assert clear_all, "the #settings-clear-all button must exist"
assert 'type="button"' in clear_all.group(0)
assert "Clear all overrides" in body
hint = re.search(r'<span[^>]*id="settings-dirty"[^>]*>', body)
assert hint, "the #settings-dirty hint must exist"
assert 'aria-hidden="true"' in hint.group(0), (
"the hint is aria-hidden (the rows + the status carry the info)"
)
def test_settings_nav_link_ships_hidden_in_the_shell() -> None:
"""The #nav-settings link (the shell's eighth view) ships hidden
(admin-only) and UNstamped — the router is the single writer of
the active state, and a token user (role "user") must never see
the link (header.js reveals it for admin only)."""
html = _html()
match = re.search(r'<a[^>]*id="nav-settings"[^>]*>', html)
assert match, "the shell must carry the #nav-settings nav link"
link = match.group(0)
assert 'href="/settings.html"' in link, "the Settings link's href"
assert "hidden" in link, "#nav-settings ships hidden (admin-only)"
assert "is-active" not in link, "no static active stamp on the Settings link"
def test_nav_settings_sits_after_nav_theme() -> None:
"""The one-bar contract: the Settings link is the LAST admin link
(after #nav-theme) — the navbar order is part of the
test_nav_consistency inventory, so the position is pinned."""
html = _html()
theme_i = html.find('id="nav-theme"')
settings_i = html.find('id="nav-settings"')
sign_in_i = html.find('id="sign-in-link-mobile"')
assert 0 <= theme_i < settings_i < sign_in_i, (
"the Settings link sits after Theme and before the auth links"
)
# ---------- router.js: /settings.html in ALL THREE tables ----------
def test_router_js_maps_settings_in_all_three_tables() -> None:
"""router.js: VIEW maps /settings.html → settings, VIEW_PATH maps
settings → /settings.html, and VIEW_MODULES lazy-imports
./settings.js with a STATIC specifier (so the Containerfile's
esbuild --bundle of router.js inlines it — settings.js needs no
new esbuild line because no page references it directly;
tests/integration/test_containerfile_assets.py proves the
coverage). The router is the SINGLE writer of the per-view
title/meta — settings has both entries."""
js = _read(ASSETS / "router.js")
view_start = js.find("const VIEW = {")
view_body = js[view_start : js.find("\n}", view_start)]
assert '"/settings.html": "settings"' in view_body, (
"phase 168 task 05 folds the Settings view into the shell"
)
path_start = js.find("const VIEW_PATH = {")
path_body = js[path_start : js.find("\n}", path_start)]
assert 'settings: "/settings.html"' in path_body, (
"VIEW_PATH must map settings → /settings.html"
)
mods_start = js.find("const VIEW_MODULES = {")
mods_body = js[mods_start : js.find("\n}", mods_start)]
assert 'settings: () => import("./settings.js")' in mods_body, (
"the Settings view module is lazy-imported on first show "
"(a static specifier — the bundle inlines it)"
)
# The per-view head values (the router's single-writer contract).
assert 'settings: "Settings · Brain of Reese"' in js, (
"the Settings view's title (carried through brandName)"
)
assert (
"Edit any application setting — a saved value wins over the environment, "
"live on save; reset hands control back."
) in js, "the Settings view's meta description"
# ---------- the server-side registration (shell route + caching) ----------
def test_shell_route_and_html_pages_carry_settings() -> None:
"""app/main.py's _shell_routes tuple and caching.HTML_PAGES both
list /settings.html — without the tuple the deep link 404s
(the static mount has no settings.html file), without the
HTML_PAGES entry it would serve unversioned asset refs, which
the immutable-for-a-year asset caching would pin to stale CSS
after a deploy (the phase-35 lesson, the A19 contract)."""
main_py = _read(REPO / "app" / "main.py")
app_start = main_py.find("def create_app")
mount_i = main_py.find('app.mount("/", StaticFiles')
assert 0 <= app_start < mount_i, "create_app + the static mount must exist"
routes_block = main_py[app_start:mount_i]
assert '"/settings.html"' in routes_block, (
"main.py's shell tuple must list /settings.html (deep link)"
)
caching_py = _read(REPO / "app" / "core" / "caching.py")
pages_start = caching_py.find("HTML_PAGES: tuple")
pages_end = caching_py.find("\n)\n", pages_start)
pages_block = caching_py[pages_start:pages_end]
assert '"/settings.html"' in pages_block, (
"caching.HTML_PAGES must list /settings.html (no-cache + ?v=)"
)
# The shell route comes AFTER the API router registration and
# BEFORE the static mount (routes-first — the docstring contract).
include_i = main_py.find("app.include_router(admin_settings_router")
assert 0 <= app_start < include_i < mount_i, (
"the shell route sits between the API routers and the static mount"
)
# ---------- header.js: the reveal is admin-keyed, null-safe ----------
def test_header_js_reveals_settings_link_for_admin_only() -> None:
"""header.js reveals #nav-settings for role admin — the same
ship-hidden / reveal-for-admin contract as the other admin-only
links: the two-line reveal (`hidden = !admin`) sits in
initSharedHeader, null-safe (a page without the link is a
no-op), and the gate is the `admin` flag (role === "admin") —
a token user (role "user") never sees the link. The lookup +
reveal sit AFTER the whoami resolution (the admin flag exists
only once fetchWhoami has settled)."""
header_js = _read(ASSETS / "header.js")
fn = header_js.find("export async function initSharedHeader")
assert fn != -1, "initSharedHeader must exist"
body = header_js[fn:]
lookup = body.find('document.querySelector("#nav-settings")')
assert lookup != -1, "header.js must look up #nav-settings"
reveal = body.find("navSettings.hidden = !admin")
assert 0 <= lookup < reveal, (
"the reveal must be the two-line pattern: null-safe lookup, "
'then hidden = !admin (the admin flag — role === "admin")'
)
whoami = body.find("const whoami = await fetchWhoami()")
admin_flag = body.find('const admin = whoami.role === "admin"')
assert 0 <= whoami < admin_flag < lookup, (
"the reveal keys off the resolved admin flag"
)
# The gate is `admin`, NOT `signedIn` (a token user never sees
# it — the phase-79 contract).
reveal_line = body[lookup:reveal + 60]
assert "signedIn" not in reveal_line, (
"the reveal must key off admin (a token user never sees the link)"
)
# ---------- settings.js: DOM APIs only, no CDN ----------
def test_settings_js_builds_dom_with_element_apis_only() -> None:
"""settings.js follows the phase-76 view-module contract:
`export async function mount(root)` is the entry, the whoami
gate (fetchIsAdmin) runs in mount and the anonymous branch shows
the gate + hides the content + RETURNS with NO
/api/admin/settings request (the router 403s anonymous), and the
bor:view-refresh listener is armed ONLY in the ADMIN branch
(after the gate), gated on the started flag. The DOM is built
with createElement + textContent — NO innerHTML (template
literal or not: the XSS-safe-by-construction house rule) and NO
http(s):// reference (rule 6 — everything is local; the fetch
paths are same-origin)."""
js = _read(ASSETS / "settings.js")
assert "export async function mount(root)" in js, (
"mount(root) must be the module's entry (the phase-76 fold)"
)
assert 'import { fetchIsAdmin } from "./header.js";' in js, (
"the view imports ONLY the shared cached whoami promise"
)
mount_i = js.find("export async function mount(root)")
gate_i = js.find("if (!(await fetchIsAdmin()))")
assert 0 <= mount_i < gate_i, "the whoami gate must run in mount"
branch = js[gate_i:js.find("return;", gate_i)]
assert "fetch(" not in branch, (
"the anonymous branch must not fetch anything"
)
assert "gateEl.hidden = false" in branch
assert "contentEl.hidden = true" in branch
# The re-show refresh: armed in the ADMIN branch only (after the
# gate), gated on the first load (started).
listener = js.find('addEventListener("bor:view-refresh"')
assert 0 <= gate_i < listener, (
"the refresh listener is armed only in the ADMIN branch (after the gate)"
)
assert re.search(r"if \(started\) void loadSettings\(\)", js), (
"the listener is gated on the first load (started)"
)
# No HTML-string DOM building (the house rule) — innerHTML is
# absent from the file entirely (a template-literal innerHTML
# would also match this check).
assert "innerHTML" not in js, (
"the DOM is built with createElement + textContent — no "
"innerHTML anywhere (XSS-safe by construction)"
)
# No CDN (rule 6): no http(s):// reference in the module — the
# fetch paths are same-origin (/api/admin/settings…).
assert not re.search(r"https?://", js), (
"no CDN — every reference is local (AGENTS.md rule 6)"
)
# The API surface: GET the catalogue, PUT one value, DELETE one /
# DELETE all (the phase-168 API contract, task 04).
assert 'fetch("/api/admin/settings")' in js, "the catalogue GET"
assert "fetch(`/api/admin/settings/${key}`" in js, (
"the per-key PUT / DELETE"
)
# The §7.4 lifecycle: the save button is aria-busy + labelled
# while in flight, re-enabled in the same path (success AND
# failure — the saveInFlight guard).
assert 'setAttribute("aria-busy", "true")' in js
assert "removeAttribute(\"aria-busy\")" in js
assert 'textContent = "Saving…"' in js
def test_settings_js_dirty_tracking_and_stop_at_first_failure() -> None:
"""The K4 contract, pinned at source level: the dirty state is a
module Map (key → value); markDirty compares the control value
against the SERVED value (a masked field's served value is "" —
an empty input means "leave unchanged", never an accidental
wipe); Save walks the dirty rows IN DOM ORDER, PUTs { value },
and breaks at the first failure (the 422 detail lands in the
role=alert region, the load re-applies the remaining edits,
focus goes to the offending input); a successful save lands the
`Saved N settings.` status line + the restart-required
suffix."""
js = _read(ASSETS / "settings.js")
assert "const dirty = new Map();" in js, (
"the dirty state is a module Map"
)
mark = js.find("function markDirty(")
mark_body = js[mark:js.find("\n }", mark)]
assert 'item.secret ? "" : item.value ?? ""' in mark_body, (
"the served value is '' for a masked field (never an accidental wipe)"
)
save = js.find("async function saveSettings()")
save_body = js[save:js.find("\n }", save)]
assert "for (const row of dataRows())" in save_body, (
"Save walks the dirty rows in DOM order"
)
put_i = save_body.find("JSON.stringify({ value })")
break_i = save_body.find("break; // stop at the first failure (K4)")
assert 0 <= put_i < break_i, "the loop breaks at the first failure (K4)"
assert "showError(" in save_body, "the 422 detail lands in the error region"
assert "input.focus()" in save_body, (
"focus goes to the offending input (K4)"
)
assert '`Saved ${saved.length} settings.`' in save_body, (
"the one status line (role=status)"
)
assert "applies on the next restart" in save_body, (
"a saved restart-required row says so (the owner is never "
"told 'saved' and then sees nothing change)"
)
# The reset + clear-all contract: the DELETE one/all + the
# settled status lines.
assert "`${envVar} is back to the environment value.`" in js, (
"the per-row Reset status line"
)
assert "Cleared ${cleared} overrides." in js, (
"the Clear-all status line (N from the response)"
)
# The two-step confirm (the house history-confirm-* classes,
# focus to Yes, Escape / No cancels).
assert 'label.textContent = "Clear all overrides?"' in js
assert 'yes.className = "history-confirm-yes"' in js
assert 'no.className = "history-confirm-no"' in js
assert "yes.focus()" in js, "focus moves to Yes (keyboard confirm)"
assert "e.key === \"Escape\"" in js, "Escape cancels (the task contract)"
# ---------- styles.css: the settings block, no new :root tokens ----------
def test_styles_css_settings_block_uses_existing_tokens_only() -> None:
"""styles.css declares the .settings-table (full width — the
.git-sources-table language, no skinny list) + the three
.settings-src.is-* variants (SOLID existing tokens — B3) + the
dirty-row marker, and adds NO new :root variable (the phase-156
B3 rule — the badges are text, the color is decoration; a themed
page's #bor-theme override of the token handles the color)."""
css = _read(ASSETS / "styles.css")
assert "---------- Settings view (phase 168) ----------" in css, (
"the Settings view block must exist"
)
block = re.search(r"\.settings-table \{([\s\S]*?)\n\}", css)
assert block, "styles.css must style .settings-table"
assert "width: 100%" in block.group(1), (
"the catalogue is full width (no skinny list — PLAN §7.1)"
)
for variant, bg in (
("is-db", "var(--brand-soft)"),
("is-env", "var(--surface)"),
("is-default", "var(--line)"),
):
rule = re.search(
rf"\.settings-src\.{variant} \{{([\s\S]*?)\n\}}", css
)
assert rule, f"styles.css must style .settings-src.{variant}"
body = rule.group(1)
assert bg in body, f".settings-src.{variant} is the solid {bg} token (B3)"
# The badge is TEXT (B5): the text is never color-alone — the
# variant always pairs the background with an existing ink
# token (never a color declaration without a text color).
assert "color: var(--" in body, (
f".settings-src.{variant} carries an existing ink token (text, not color alone)"
)
# The dirty-row marker (the --brand left border on the row).
assert "settings-row.is-dirty" in css, (
"the dirty-row marker must exist (unsaved survives grayscale)"
)
dirty = re.search(r"\.settings-row\.is-dirty[^{]*\{([\s\S]*?)\n\}", css)
assert dirty and "var(--brand)" in dirty.group(1), (
"the dirty marker is the --brand left border"
)
# No new :root variable (B3, the phase-156 rule): the settings
# block must not declare a custom property — only var(--…)
# references of the pre-existing tokens (a themed page's
# #bor-theme override of the token handles the color).
settings_block = css[css.find("Settings view (phase 168)"): css.find(
"Shared page (phase 51"
)]
declarations = re.findall(r"^\s{2,}--[\w-]+\s*:", settings_block, re.M)
assert not declarations, (
f"the settings block must not declare :root variables (B3): "
f"{declarations}"
)
# The touch targets: the ≥44px rules (the house language) + the
# mobile 640px block stacks the search + action rows.
for selector in ("#settings-filter", ".settings-control"):
rule = re.search(rf"{re.escape(selector)} \{{([\s\S]*?)\n\}}", css)
assert rule, f"styles.css must style {selector}"
assert "min-height: 44px" in rule.group(1), f"{selector}: the ≥44px target"
mobile = re.search(
r"@media \(max-width: 640px\) \{([\s\S]*?)\n\}\n", css
)
assert mobile, "the 640px media query must exist"
mbody = mobile.group(1)
assert ".settings-search { flex-direction: column; align-items: stretch; }" in mbody
assert "#settings-save { width: 100%; }" in mbody, (
"the Save pill is full width on phones (the ≥44px target stands)"
)
# ---------- the four Theme-bearing pages carry the Settings link ----------
def test_settings_link_ships_on_every_theme_bearing_page() -> None:
"""The phase-34 one-bar contract — the SAME nav ships on every
page (test_nav_consistency pins the header inventory PARITY
across the shell pages, the document viewer, and the login
page), so #nav-settings (ship-hidden, admin-only) must be in
the #app-nav of EVERY header-bearing page: the shell +
document.html + login.html + shared.html. The doc-edit flow
page ships the reduced header (no admin links at all) and is
out of the contract."""
for page in (
FRONTEND / "index.html",
FRONTEND / "document.html",
FRONTEND / "login.html",
FRONTEND / "shared.html",
):
text = _read(page)
# The page must carry BOTH links — the Theme link (the
# phase-91 precedent) and the Settings link (this task).
theme = re.search(r'<a[^>]*id="nav-theme"[^>]*>', text)
assert theme, f"{page.name} must carry the #nav-theme nav link"
match = re.search(r'<a[^>]*id="nav-settings"[^>]*>', text)
assert match, (
f"{page.name} must carry the #nav-settings nav link (one-bar)"
)
link = match.group(0)
assert 'href="/settings.html"' in link, (
f"{page.name}: the Settings link's href"
)
assert "hidden" in link, (
f"{page.name}: #nav-settings ships hidden (admin-only)"
)
assert "is-active" not in link, (
f"{page.name}: no static active stamp on the Settings link"
)
# The Settings link sits AFTER the Theme link (the order
# test_nav_consistency's inventory compares).
assert text.find('id="nav-theme"') < text.find('id="nav-settings"'), (
f"{page.name}: the Settings link follows the Theme link"
)
+188
View File
@@ -0,0 +1,188 @@
"""Phase 168 (task 03): Save re-points the two in-process singletons.
The "takes effect on Save" contract (D3): ``invalidate_settings()``
clears the ``get_settings()`` cache (the ~70 per-use settings) AND runs
the registered reset hooks — ``app.api.chat.reset_runtime_caches``
(nulls the shared ``LLMClient`` and the chat concurrency semaphore,
re-registered by ``create_app()`` at the composition root because
``app/core`` must never import ``app/api``, PLAN §3). These pins are
unit level: the chat globals are set directly, the overlay hooks are
registered directly (the autouse fixture gives every test a clean hook
list and restores every global it touches — a leaked ``LLMClient``
holding a test settings snapshot would poison the next suite's
singleton).
"""
from __future__ import annotations
import logging
from collections.abc import Iterator
from pathlib import Path
import pytest
import app.main as app_main
from app.api import chat as chat_api
from app.config import get_settings, invalidate_settings
from app.core import settings_overlay
@pytest.fixture(autouse=True)
def _clean_state() -> Iterator[None]:
"""Clean hook list + saved chat globals, restored on teardown.
``reset_state_for_tests()`` clears the process-wide hook list (a
prior suite's test may have wiped the hook ``app.main`` registered
at import — these tests register their own); the ``chat_api``
globals are saved/restored around the test so a built ``LLMClient``
or semaphore never leaks into another test's singleton slot.
"""
settings_overlay.reset_state_for_tests()
saved = {
"_llm": chat_api._llm,
"_chat_semaphore": chat_api._chat_semaphore,
"_chat_active": chat_api._chat_active,
}
yield
chat_api._llm = saved["_llm"]
chat_api._chat_semaphore = saved["_chat_semaphore"]
chat_api._chat_active = saved["_chat_active"]
settings_overlay.reset_state_for_tests()
# monkeypatch restores BOR_* env first (finalized before this
# fixture) — the cache clear re-resolves to the conftest-pinned
# env-only view.
get_settings.cache_clear()
# ---------------------------------------------------------------------------
# reset_runtime_caches — nulls exactly the two settings snapshots
# ---------------------------------------------------------------------------
def test_reset_nulls_llm_and_semaphore() -> None:
chat_api._llm = object()
chat_api._chat_semaphore = object()
chat_api._chat_active = 7
chat_api.reset_runtime_caches()
assert chat_api._llm is None
assert chat_api._chat_semaphore is None
# _chat_active counts live turns — a reset must not touch it (the
# new chat_max_concurrent applies as permits are taken).
assert chat_api._chat_active == 7
def test_reset_is_idempotent() -> None:
chat_api.reset_runtime_caches()
chat_api.reset_runtime_caches() # a double Save must not raise
assert chat_api._llm is None
assert chat_api._chat_semaphore is None
# ---------------------------------------------------------------------------
# get_llm / _get_chat_semaphore — lazy rebuild from the CURRENT settings
# ---------------------------------------------------------------------------
def test_get_llm_singleton_contract() -> None:
chat_api._llm = None
first = chat_api.get_llm()
second = chat_api.get_llm()
# The singleton contract still holds — the same instance until a
# reset (the pin that stops an "always construct" shortcut that
# would break connection pooling).
assert first is second
def test_get_llm_rebuilds_after_reset() -> None:
chat_api._llm = None
first = chat_api.get_llm()
chat_api.reset_runtime_caches()
second = chat_api.get_llm()
assert second is not first
assert second.settings is get_settings()
def test_get_chat_semaphore_rebuilds_after_reset(
monkeypatch: pytest.MonkeyPatch,
) -> None:
chat_api._chat_semaphore = None
base = chat_api._get_chat_semaphore()
base_value = base._value # pyright: ignore[reportPrivateUsage]
# Drive the cap to a value different from the current one (the
# env is the base — the kill switch is off, conftest pin), so the
# rebuild's new _value is unambiguous.
target = 3 if base_value != 3 else 5
monkeypatch.setenv("BOR_CHAT_MAX_CONCURRENT", str(target))
get_settings.cache_clear()
chat_api.reset_runtime_caches()
rebuilt = chat_api._get_chat_semaphore()
assert rebuilt is not base
assert rebuilt._value == target # pyright: ignore[reportPrivateUsage]
# restore: the env reverts via monkeypatch, and the fixture's cache
# clear re-resolves the settings.
get_settings.cache_clear()
# ---------------------------------------------------------------------------
# the create_app() wiring — the composition root registers the hook
# ---------------------------------------------------------------------------
def test_create_app_wires_the_reset_hook() -> None:
"""``register_invalidator(reset_runtime_caches)`` inside
``create_app`` — the static pin that stops the wiring being
"cleaned up" (without it a Save silently stops retuning the LLM
client)."""
src = Path(app_main.__file__).read_text()
start = src.index("def create_app(")
end = src.index("\napp = create_app()", start)
body = src[start:end]
assert "register_invalidator(reset_runtime_caches)" in body
def test_app_core_imports_no_app_api() -> None:
"""Dependency direction (PLAN §3): ``app/core/`` never imports an
``app.api`` symbol — the hook exists so the composition root can
wire the reset without inverting it."""
core = Path(app_main.__file__).resolve().parent / "core"
for py in sorted(core.glob("*.py")):
src = py.read_text()
for line in src.splitlines():
stripped = line.strip()
assert not (
stripped.startswith("from app.api")
or stripped.startswith("import app.api")
), f"{py.name} imports app.api (PLAN §3 direction)"
# ---------------------------------------------------------------------------
# invalidate_settings — cache clear + hook run, a failed reset stands
# ---------------------------------------------------------------------------
def test_invalidate_runs_hooks_and_re_resolves(
monkeypatch: pytest.MonkeyPatch,
) -> None:
calls: list[str] = []
settings_overlay.register_invalidator(lambda: calls.append("hook"))
get_settings.cache_clear()
get_settings() # a cached hit under the ORIGINAL env
monkeypatch.setenv("BOR_APP_NAME", "Live-Apply-Check")
invalidate_settings() # clears the cache + runs the hooks
assert calls == ["hook"]
# the re-resolution sees the NEW env — the cache really was cleared
assert get_settings().app_name == "Live-Apply-Check"
def test_invalidate_survives_a_raising_hook(
caplog: pytest.LogCaptureFixture,
) -> None:
"""A failed reset must not turn a committed save into a 500 — the
exception is logged, never propagated."""
def _boom() -> None:
raise RuntimeError("boom")
settings_overlay.register_invalidator(_boom)
with caplog.at_level(logging.ERROR, logger="app.settings"):
invalidate_settings() # must not raise
assert any("settings invalidator failed" in r.getMessage() for r in caplog.records)
+427
View File
@@ -0,0 +1,427 @@
"""Phase 168 (task 02): the DB-over-env overlay resolver.
Pure/unit level — no live DB: the ``load_overrides`` failure paths use
a stubbed engine, the kill switch is flipped via ``os.environ`` (it is
read per call), and the env base comes from the process environment
(pinned by ``tests/conftest.py`` — ``BOR_SETTINGS_DB=0``).
"""
from __future__ import annotations
import logging
from pathlib import Path
import pytest
from sqlalchemy.exc import OperationalError, ProgrammingError
from app.config import Settings
from app.core import settings_overlay
from app.core.settings_overlay import (
_coerce,
enabled,
load_overrides,
merge,
register_invalidator,
reset_state_for_tests,
resolve_settings,
run_invalidators,
)
@pytest.fixture(autouse=True)
def _clean_state() -> None:
reset_state_for_tests()
# ---------------------------------------------------------------------------
# enabled() — the BOR_SETTINGS_DB kill switch (D6)
# ---------------------------------------------------------------------------
def test_enabled_matrix(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("BOR_SETTINGS_DB", raising=False)
assert enabled() is True
for value in ("0", "false", "no", "off"):
monkeypatch.setenv("BOR_SETTINGS_DB", value)
assert enabled() is False, value
for value in ("1", "true"):
monkeypatch.setenv("BOR_SETTINGS_DB", value)
assert enabled() is True, value
# read per call — a flip is visible without any cache clear
monkeypatch.setenv("BOR_SETTINGS_DB", "off")
assert enabled() is False
monkeypatch.setenv("BOR_SETTINGS_DB", "1")
assert enabled() is True
# ---------------------------------------------------------------------------
# _coerce — env-native strings; the one list[str] field is JSON
# ---------------------------------------------------------------------------
def test_coerce_scalars_are_raw_strings() -> None:
# pydantic-settings lax-coerces on Settings() — the raw string goes
# through exactly like the env var spelling.
assert _coerce("relevance_threshold", "0.55") == "0.55"
assert _coerce("llm_retries", "7") == "7"
assert _coerce("stream_thinking", "false") == "false"
assert _coerce("import_extensions", "md,txt") == "md,txt"
def test_coerce_suggestions_json() -> None:
assert set(settings_overlay.LIST_FIELDS) == {"suggestions"}
assert _coerce("suggestions", '["a", "b"]') == ["a", "b"]
with pytest.raises(ValueError):
_coerce("suggestions", "not a json array")
# ---------------------------------------------------------------------------
# merge — layer rows over the env base, fail soft per key (LOCKED A5)
# ---------------------------------------------------------------------------
def test_merge_happy_path_env_base() -> None:
base = Settings()
merged, skipped = merge({"relevance_threshold": "0.55"})
assert skipped == []
assert merged.relevance_threshold == 0.55
assert isinstance(merged.relevance_threshold, float)
# everything else is byte-equal to the env base
for name in Settings.model_fields:
if name != "relevance_threshold":
assert getattr(merged, name) == getattr(base, name), name
def test_merge_skips_unknown_and_non_overridable() -> None:
base = Settings()
merged, skipped = merge(
{
"no_such_field": "x", # orphan row from a removed setting
"database_url": base.database_url, # LOCKED A3 — read-only
"embedding_dim": "512", # LOCKED A3 — read-only
}
)
keys = {key for key, _ in skipped}
assert keys == {"no_such_field", "database_url", "embedding_dim"}
# the base values stand — a hand-inserted row never hijacks them
assert merged.database_url == base.database_url
assert merged.embedding_dim == base.embedding_dim
assert merged.embedding_dim == 768
def test_merge_drops_one_invalid_value_keeps_the_rest() -> None:
base = Settings()
merged, skipped = merge(
{
"relevance_threshold": "not-a-float",
"llm_retries": "7",
}
)
assert [key for key, _ in skipped] == ["relevance_threshold"]
assert merged.llm_retries == 7
# the bad row falls back to the env/base value
assert merged.relevance_threshold == base.relevance_threshold
def test_merge_malformed_suggestions_json_is_skipped_not_fatal() -> None:
base = Settings()
merged, skipped = merge({"suggestions": "definitely not json"})
assert [key for key, _ in skipped] == ["suggestions"]
assert merged.suggestions == base.suggestions
assert isinstance(merged.suggestions, list)
def test_merge_valid_suggestions_json_overrides() -> None:
merged, skipped = merge({"suggestions": '["only this"]'})
assert skipped == []
assert merged.suggestions == ["only this"]
def test_merge_cross_field_violation_skipped_not_raised(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setenv("BOR_RELEVANCE_THRESHOLD", "0.30")
monkeypatch.setenv("BOR_LEXICAL_SUPPORT_FLOOR", "0.15")
# floor 0.9 > threshold 0.55 — the validator's cross-field guard
# (app/config.py) fires; the FLOOR row is dropped, the threshold
# override stands, the floor falls back to the env value.
merged, skipped = merge(
{"relevance_threshold": "0.55", "lexical_support_floor": "0.9"}
)
keys = {key for key, _ in skipped}
assert "lexical_support_floor" in keys
assert merged.relevance_threshold == 0.55
assert merged.lexical_support_floor == 0.15 # env fallback
def test_merge_env_collision_cross_field_terminates(
monkeypatch: pytest.MonkeyPatch,
) -> None:
# The env floor (0.15) is above the DB-overridden threshold (0.1):
# the validator blames the FLOOR, which has no override — its value
# comes from env and cannot be dropped. The resolver must find a
# single override drop that heals the set (the threshold row)
# instead of looping: one unresolvable collision costs the
# colliding row, the rest of the table stands (LOCKED A5 fail-soft).
monkeypatch.setenv("BOR_RELEVANCE_THRESHOLD", "0.30")
monkeypatch.setenv("BOR_LEXICAL_SUPPORT_FLOOR", "0.15")
base = Settings()
merged, skipped = merge({"relevance_threshold": "0.1"})
keys = {key for key, _ in skipped}
assert "relevance_threshold" in keys
assert merged.relevance_threshold == base.relevance_threshold
assert merged.lexical_support_floor == base.lexical_support_floor
def test_merge_bool_spellings_behave_like_env() -> None:
merged_false, skipped = merge({"stream_thinking": "false"})
assert skipped == []
assert merged_false.stream_thinking is False
merged_zero, _ = merge({"stream_thinking": "0"})
assert merged_zero.stream_thinking is False
# untouched fields keep the base value
base = Settings()
assert merged_false.llm_retries == base.llm_retries
def test_merge_empty_rows_is_the_env_base() -> None:
base = Settings()
merged, skipped = merge({})
assert skipped == []
assert merged == base
def test_load_overrides_success_path(monkeypatch: pytest.MonkeyPatch) -> None:
class _Rows:
def __iter__(self):
return iter((("llm_retries", "9"), ("stream_thinking", "false")))
class _FakeConn:
def execute(self, _stmt: object) -> _Rows:
return _Rows()
def __enter__(self):
return self
def __exit__(self, *exc: object) -> bool:
return False
class _FakeEngine:
def connect(self) -> _FakeConn: # type: ignore[no-untyped-def]
return _FakeConn()
monkeypatch.setattr(settings_overlay, "_get_engine", lambda: _FakeEngine())
assert load_overrides() == {"llm_retries": "9", "stream_thinking": "false"}
def test_get_engine_creates_once_and_caches() -> None:
# create_engine is lazy (no connection at creation) — the second
# call must return the SAME engine (the once-per-process engine).
first = settings_overlay._get_engine()
second = settings_overlay._get_engine()
assert first is second
# ---------------------------------------------------------------------------
# merge — the defensive fallback (a validator change in a later release
# blames a field no override owns and no single drop heals the set: the
# env base stands, logged)
# ---------------------------------------------------------------------------
def test_merge_unresolvable_collision_returns_base_logged(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from typing import Any, cast
from pydantic import ValidationError
class _StubSettings:
"""Raises like a broken candidate set; the final no-args call
(the env-base fallback) succeeds."""
model_fields = Settings.model_fields # merge() pre-checks it
calls = 0
def __init__(self, **_kwargs: object) -> None:
_StubSettings.calls += 1
if _StubSettings.calls < 3:
# the runtime shape (msg/error keys) is wider than the
# pyright stub's InitErrorDetails — cast it
raise ValidationError.from_exception_data(
"Settings",
cast(
Any,
[
{
"type": "missing",
"loc": ("lexical_support_floor",),
"msg": "a later-release validator",
"error": ValueError("a later-release validator"),
"input_value": 0.35,
}
],
),
)
monkeypatch.setattr(settings_overlay, "Settings", _StubSettings)
merged, skipped = merge({"relevance_threshold": "0.1"})
assert merged is not None # the env base (the stub's final call)
reasons = {key: reason for key, reason in skipped}
assert "unresolvable cross-field collision" in (
reasons.get("relevance_threshold", "")
or reasons.get("lexical_support_floor", "")
)
# ---------------------------------------------------------------------------
# load_overrides — fail-soft, once-per-process (K3 + D5)
# ---------------------------------------------------------------------------
class _FailingEngine:
"""A stub engine whose connect() raises the given exception."""
def __init__(self, exc: Exception) -> None:
self._exc = exc
def connect(self) -> None: # type: ignore[no-untyped-def]
raise self._exc
def test_load_overrides_generic_failure_warns_once(
monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
monkeypatch.setattr(
settings_overlay,
"_get_engine",
lambda: _FailingEngine(
OperationalError("SELECT 1", {}, OSError("connection refused"))
),
)
with caplog.at_level(logging.WARNING, logger="app.settings"):
assert load_overrides() == {}
assert load_overrides() == {} # the second miss is silent
warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
assert len(warnings) == 1
assert "settings overrides unavailable — env values in effect" in warnings[0].msg
# the exception class is named, the value is not
assert "OperationalError" in warnings[0].getMessage()
def test_load_overrides_missing_table_debugs_once(
monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
monkeypatch.setattr(
settings_overlay,
"_get_engine",
lambda: _FailingEngine(
ProgrammingError(
"SELECT key, value FROM app_settings",
{},
Exception('relation "app_settings" does not exist'),
)
),
)
with caplog.at_level(logging.DEBUG, logger="app.settings"):
assert load_overrides() == {}
assert load_overrides() == {} # the second miss is silent
debugs = [
r
for r in caplog.records
if r.levelno == logging.DEBUG and "app_settings" in r.getMessage()
]
assert len(debugs) == 1
def test_reset_state_restores_the_warning(
monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
monkeypatch.setattr(
settings_overlay,
"_get_engine",
lambda: _FailingEngine(
OperationalError("SELECT 1", {}, OSError("connection refused"))
),
)
with caplog.at_level(logging.WARNING, logger="app.settings"):
load_overrides() # first miss: warns
reset_state_for_tests() # clears the once-per-process flag
load_overrides() # second miss: warns again
warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
assert len(warnings) == 2
# ---------------------------------------------------------------------------
# resolve_settings — the whole overlay, recursion guard (D2)
# ---------------------------------------------------------------------------
def test_resolve_settings_disabled_returns_env_base(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("BOR_SETTINGS_DB", "0")
assert resolve_settings() == Settings()
def test_resolve_settings_enabled_uses_the_rows(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("BOR_SETTINGS_DB", "1")
monkeypatch.setattr(
settings_overlay, "load_overrides", lambda: {"llm_retries": "9"}
)
settings = resolve_settings()
assert settings.llm_retries == 9
def test_resolve_settings_never_reaches_back_into_config() -> None:
"""The recursion guard, source-pinned: the overlay builds
``Settings()`` directly — app.config's resolver delegates HERE."""
src = Path(settings_overlay.__file__).read_text()
assert "get_settings(" not in src
# ---------------------------------------------------------------------------
# the invalidator registry (task 03's seam)
# ---------------------------------------------------------------------------
def test_invalidators_run_in_order_and_survive_a_raising_hook(
caplog: pytest.LogCaptureFixture,
) -> None:
calls: list[str] = []
def _boom() -> None:
raise RuntimeError("boom")
register_invalidator(lambda: calls.append("first"))
register_invalidator(_boom)
register_invalidator(lambda: calls.append("third"))
with caplog.at_level(logging.ERROR, logger="app.settings"):
run_invalidators() # must not raise
assert calls == ["first", "third"]
assert any(
"settings invalidator failed" in r.getMessage() for r in caplog.records
)
# ---------------------------------------------------------------------------
# LOCKED A1 — .env is never written by the app
# ---------------------------------------------------------------------------
def test_phase_files_never_write_the_env_file() -> None:
"""No ``open(`` / ``write_text(`` in the phase-168 modules — the
panel writes Postgres only (LOCKED A1); ``app/api/admin_settings.py``
(task 04) is pinned in the same way once it exists."""
root = Path(settings_overlay.__file__).resolve().parents[2]
for rel in (
"app/core/settings_overlay.py",
"app/config.py",
"app/api/admin_settings.py",
):
path = root / rel
if not path.exists():
continue
src = path.read_text()
assert "open(" not in src, (
f"{rel} must never open() a file (LOCKED A1: .env is never "
f"written by the app)"
)
assert "write_text(" not in src, f"{rel} must never write a file (LOCKED A1)"