"""Tests for the /admin/api persisted-config endpoints. ``admin.py`` exposes ``GET /admin/api/config`` (the allowlisted values with provenance, ``{value, source}``) and ``POST /admin/api/config/{key}`` (persist one allowlisted value to ``config.local.yaml`` with comment-preserving ruamel.yaml round-trip, a backup copy, and whole-config validation of the merged base + overlay via ``RouterConfig`` BEFORE anything touches disk). Each test builds its own isolated router against a temp copy of config.yaml by passing ``base_dir`` (a temp dir) to ``build_router`` — the real repo ``config.yaml`` is never written. The overlay endpoints create and mutate ``/config/config.local.yaml``. It mounts the router on a fresh FastAPI TestClient at ``prefix="/admin"``. """ from __future__ import annotations import re import shutil import sqlite3 import subprocess import threading from pathlib import Path import pytest import yaml from pydantic import ValidationError from fastapi import FastAPI from starlette.testclient import TestClient from admin import ( _CONFIG_ALLOWLIST, _persist_config_block, _persist_config_value, _persist_to, build_router, ) from config import load_config ROOT = Path(__file__).resolve().parent.parent _SENTINEL = "# SENTINEL_PRESERVED_12345" def _insert_sentinel(config_yaml: Path) -> None: """Prepend a unique marker comment just above the ``logging:`` map.""" text = config_yaml.read_text() assert "logging:" in text config_yaml.write_text(text.replace("logging:", f"{_SENTINEL}\nlogging:", 1)) def _make_db(tmp_path: Path) -> sqlite3.Connection: conn = sqlite3.connect(str(tmp_path / "admin.db")) conn.row_factory = sqlite3.Row return conn @pytest.fixture def client(tmp_path, monkeypatch): """A TestClient for an isolated admin router over a temp config.yaml copy. ``base_dir`` is tmp_path, so the ``/admin/api/config`` endpoints read and write ``tmp_path/config/config.yaml`` — never the repo's copy. """ (tmp_path / "config").mkdir(parents=True, exist_ok=True) config_yaml = tmp_path / "config" / "config.yaml" shutil.copyfile(ROOT / "config" / "config.yaml", config_yaml) schema_sql = (ROOT / "config" / "schema.sql").read_text() conn = _make_db(tmp_path) conn.executescript(schema_sql) conn.close() def _db_factory() -> sqlite3.Connection: return _make_db(tmp_path) cfg = load_config(str(config_yaml)) router = build_router(cfg, _db_factory, base_dir=str(tmp_path)) app = FastAPI() app.include_router(router, prefix="/admin") return TestClient(app), config_yaml def test_config_GET_returns_allowlisted_values(client): """GET /admin/api/config returns every allowlisted dotted key.""" tc, _ = client resp = tc.get("/admin/api/config") assert resp.status_code == 200 body = resp.json() for key in ( "logging.level", "objective.quality_tolerance", "objective.max_energy_per_request", "objective.plan_kwh_per_period", "circuit_breaker.enabled", "session_cache.enabled", "session_cache.staleness_seconds", "verification.local_llm_enabled", "local_vision.enabled", "objective.incumbent_cache_pricing", "objective.incumbent_challenger_cache_rate", "pinch.enabled", "pinch.prefix_probe", "pinch.relevance.enabled", "routing.default_flex_preference", ): assert key in body assert body["logging.level"] == {"value": "info", "source": "base", "category": "routing_quality_cost", "advanced": False} assert body["objective.quality_tolerance"] == {"value": 0.10, "source": "base", "category": "routing_quality_cost", "advanced": False} assert body["routing.default_flex_preference"] == { "value": "auto", "source": "base", "category": "routing_quality_cost", "advanced": False, } def test_config_POST_preserves_comments_and_changes_value(client): """A valid write keeps the file's comments AND updates the value.""" tc, config_yaml = client local_yaml = config_yaml.with_name("config.local.yaml") _insert_sentinel(config_yaml) resp = tc.post("/admin/api/config/logging.level", json={"value": "warning"}) assert resp.status_code == 200 body = resp.json() assert body["key"] == "logging.level" assert body["value"] == "warning" assert "restart is required" in body["message"] # The base file is untouched; the overlay now carries the changed value. base_text = config_yaml.read_text() assert _SENTINEL in base_text assert re.search(r"^\s*level:\s*info\s*$", base_text, re.MULTILINE) is not None local_text = local_yaml.read_text() assert re.search( r"^\s*level:\s*warning\s*$", local_text, re.MULTILINE ) is not None def test_config_pinch_prefix_probe_round_trips_through_the_real_validator( client, tmp_path ): """Read true, write false, read back false, and the overlay still loads. The round-trip through ``load_config`` is the point, not decoration. The ``classifier.confidence_threshold`` incident was an admin control writing a value (a raw ``80`` meaning 80%) the config loader would later reject or silently misread, and it was caught before the restart only by luck. A control that writes something the loader will not take back is worse than no control, so this asserts the merged base + overlay parses and that the value survives as a real ``False``, not the string ``"false"``. """ tc, config_yaml = client local_yaml = config_yaml.with_name("config.local.yaml") before = tc.get("/admin/api/config").json() assert before["pinch.prefix_probe"] == {"value": True, "source": "base", "category": "caching_context", "advanced": True} resp = tc.post("/admin/api/config/pinch.prefix_probe", json={"value": False}) assert resp.status_code == 200 assert resp.json()["value"] is False after = tc.get("/admin/api/config").json() assert after["pinch.prefix_probe"] == {"value": False, "source": "overlay", "category": "caching_context", "advanced": True} # The base file never carries an admin write. assert yaml.safe_load(config_yaml.read_text())["pinch"]["prefix_probe"] is True assert yaml.safe_load(local_yaml.read_text())["pinch"]["prefix_probe"] is False # And the merged result parses through RouterConfig, which is what the # service does on its next boot. loaded = load_config(str(config_yaml), include_overlay=True) assert loaded.pinch.prefix_probe is False # Nothing else in the pinch block was disturbed by the overlay merge. assert loaded.pinch.enabled is True assert loaded.pinch.relevance.enabled is True def test_config_pinch_prefix_probe_is_allowlisted_at_the_right_path(client): """The allowlist entry points at ``pinch.prefix_probe``, not a sibling. A wrong path here would write a key ``RouterConfig`` forbids (StrictModel sets ``extra="forbid"``), so the whole-config validation would 422 every save -- but only for this one key, which is exactly the sort of thing a green test run hides. """ assert _CONFIG_ALLOWLIST["pinch.prefix_probe"] == ("pinch", "prefix_probe") def test_config_local_vision_enabled_round_trips_through_the_real_validator(client): """local_vision.enabled persists to the temp overlay and reloads as a bool. Same reasoning as the ``pinch.prefix_probe`` round-trip: the value that reaches disk has to be one the config loader will take back on the next boot. ``local_vision.enabled`` is read per request in the dispatcher, so a write that lands as the string ``"false"`` would silently keep the fallback on. Assert a real ``False`` survives the merged reload. """ tc, config_yaml = client local_yaml = config_yaml.with_name("config.local.yaml") before = tc.get("/admin/api/config").json() assert before["local_vision.enabled"] == {"value": True, "source": "base", "category": "local_hardware", "advanced": False} resp = tc.post("/admin/api/config/local_vision.enabled", json={"value": False}) assert resp.status_code == 200 assert resp.json()["value"] is False after = tc.get("/admin/api/config").json() assert after["local_vision.enabled"] == {"value": False, "source": "overlay", "category": "local_hardware", "advanced": False} # The base file never carries an admin write. assert yaml.safe_load(config_yaml.read_text())["local_vision"]["enabled"] is True assert yaml.safe_load(local_yaml.read_text())["local_vision"]["enabled"] is False # And the merged result parses through RouterConfig, which is what the # service does on its next boot. loaded = load_config(str(config_yaml), include_overlay=True) assert loaded.local_vision.enabled is False # The rest of the local_vision block is untouched by the overlay merge. base_vision = yaml.safe_load(config_yaml.read_text())["local_vision"] assert loaded.local_vision.model == base_vision["model"] assert loaded.local_vision.base_url == base_vision["base_url"] def test_config_incumbent_knobs_round_trip_through_the_real_validator(client): """Both Wave 2 knobs survive a write as genuine Python types. Same reasoning as the ``pinch.prefix_probe`` round-trip: the value that reaches disk has to be one the config loader will take back on the next boot. The extra assertion here is the TYPE -- a real ``bool`` and a real ``float``, not the strings ``"true"`` and ``"0.0"``, because the dial is read straight into an arithmetic expression in ``routing.estimated_cost`` and a string would not fail until a request was already in flight. """ tc, config_yaml = client before = tc.get("/admin/api/config").json() assert before["objective.incumbent_cache_pricing"] == { "value": False, "source": "base", "category": "routing_quality_cost", "advanced": False, } # Shipped blank in config.yaml, which YAML reads as null. assert before["objective.incumbent_challenger_cache_rate"] == { "value": None, "source": "base", "category": "routing_quality_cost", "advanced": False, } assert ( tc.post( "/admin/api/config/objective.incumbent_cache_pricing", json={"value": True}, ).status_code == 200 ) assert ( tc.post( "/admin/api/config/objective.incumbent_challenger_cache_rate", json={"value": 0.0}, ).status_code == 200 ) after = tc.get("/admin/api/config").json() assert after["objective.incumbent_cache_pricing"] == { "value": True, "source": "overlay", "category": "routing_quality_cost", "advanced": False, } assert after["objective.incumbent_challenger_cache_rate"] == { "value": 0.0, "source": "overlay", "category": "routing_quality_cost", "advanced": False, } # The base file never carries an admin write. base = yaml.safe_load(config_yaml.read_text())["objective"] assert base["incumbent_cache_pricing"] is False assert base["incumbent_challenger_cache_rate"] is None loaded = load_config(str(config_yaml), include_overlay=True) assert loaded.objective.incumbent_cache_pricing is True rate = loaded.objective.incumbent_challenger_cache_rate assert isinstance(rate, float) and not isinstance(rate, bool) assert rate == 0.0 # Nothing else in the objective block moved on the overlay merge. assert loaded.objective.quality_tolerance == 0.10 assert loaded.objective.incumbent_rate_min_observations == 25 def test_config_blank_challenger_dial_persists_as_null_not_zero(client): """Clearing the field writes ``null``, which the loader reads as NEUTRAL. ``null`` and ``0.0`` are opposite ends of this dial: null follows ``assumed_cache_rate`` and costs a challenger nothing, 0.0 prices every challenger as a fully cold prompt. A UI that sent 0.0 for an empty input would hand an operator the maximum penalty at the exact moment they were trying to switch the feature off, so the null has to survive all the way to the file AND be resolved by the loader rather than by the UI. """ tc, config_yaml = client local_yaml = config_yaml.with_name("config.local.yaml") # Start at full penalty so a neutral write has something to undo. tc.post( "/admin/api/config/objective.incumbent_challenger_cache_rate", json={"value": 0.0}, ) assert ( yaml.safe_load(local_yaml.read_text())["objective"][ "incumbent_challenger_cache_rate" ] == 0.0 ) resp = tc.post( "/admin/api/config/objective.incumbent_challenger_cache_rate", json={"value": None}, ) assert resp.status_code == 200 overlay = yaml.safe_load(local_yaml.read_text())["objective"] assert "incumbent_challenger_cache_rate" in overlay assert overlay["incumbent_challenger_cache_rate"] is None body = tc.get("/admin/api/config").json() assert body["objective.incumbent_challenger_cache_rate"] == { "value": None, "source": "overlay", "category": "routing_quality_cost", "advanced": False, } loaded = load_config(str(config_yaml), include_overlay=True) assert loaded.objective.incumbent_challenger_cache_rate == ( loaded.objective.assumed_cache_rate ) assert loaded.objective.incumbent_challenger_cache_rate != 0.0 @pytest.mark.parametrize("bad", [-0.1, 1.5, 80]) def test_config_challenger_dial_refuses_values_outside_zero_to_one(client, bad): """The field is a rate in [0, 1], never a percentage. ``80`` is the shape of the ``classifier.confidence_threshold`` incident: the admin UI saved a raw 80 meaning 80% for a field the loader wanted as 0.0-1.0, and every reading would have been below threshold on the next restart. Here RouterConfig validates the MERGED config before a byte reaches disk, so the write is refused instead. """ tc, config_yaml = client local_yaml = config_yaml.with_name("config.local.yaml") resp = tc.post( "/admin/api/config/objective.incumbent_challenger_cache_rate", json={"value": bad}, ) assert resp.status_code == 422 assert "must be in [0, 1]" in resp.json()["detail"] assert not local_yaml.exists() def test_config_incumbent_keys_are_allowlisted_at_the_right_paths(): """A wrong path would write a key StrictModel forbids, 422ing every save.""" assert _CONFIG_ALLOWLIST["objective.incumbent_cache_pricing"] == ( "objective", "incumbent_cache_pricing", ) assert _CONFIG_ALLOWLIST["objective.incumbent_challenger_cache_rate"] == ( "objective", "incumbent_challenger_cache_rate", ) def test_config_staleness_window_round_trips_as_a_real_int(client): """The window survives a write as an ``int`` the loader takes straight back. The type assertion is the point, and it is the ``classifier.confidence_threshold`` lesson in its other form: that incident wrote a raw ``80`` meaning 80% for a field the loader wanted as 0.0-1.0, and nothing caught it until someone looked. Here the units are seconds and the field is declared ``int``, so the UI must not scale it and the write must not float it — a ``1200.0`` on disk is a value ``load_config`` refuses, and a quietly-truncated ``20.5`` is a window the operator never chose. """ tc, config_yaml = client before = tc.get("/admin/api/config").json() assert before["session_cache.staleness_seconds"] == {"value": 1200, "source": "base", "category": "caching_context", "advanced": False} assert ( tc.post( "/admin/api/config/session_cache.staleness_seconds", json={"value": 5}, ).status_code == 200 ) after = tc.get("/admin/api/config").json() assert after["session_cache.staleness_seconds"] == {"value": 5, "source": "overlay", "category": "caching_context", "advanced": False} # The shipped default never moves; the admin write lands in the overlay. base = yaml.safe_load(config_yaml.read_text())["session_cache"] assert base["staleness_seconds"] == 1200 loaded = load_config(str(config_yaml), include_overlay=True) seconds = loaded.session_cache.staleness_seconds assert isinstance(seconds, int) and not isinstance(seconds, bool) assert seconds == 5 # The switch beside it did not move on the overlay merge. assert loaded.session_cache.enabled is True @pytest.mark.parametrize("bad", [0, -1, 4, 7201, 10000, None]) def test_config_staleness_window_refuses_out_of_range_and_blank(client, bad): """Whole-config validation refuses before a byte reaches the overlay. ``0`` is in here on purpose. It looks like "stop reusing classifications" and is not: ``session_cache.put`` still writes and the classifier-failure cascade's ``stale_read`` ignores staleness entirely, so a 0-second window still replays a session's label whenever the classifier is down. The knob that actually stops reuse is ``session_cache.enabled``, which has its own control, so 0 is refused rather than quietly honoured — and ``None`` (a cleared field) with it, since blank is not a setting either. """ tc, config_yaml = client local_yaml = config_yaml.with_name("config.local.yaml") resp = tc.post( "/admin/api/config/session_cache.staleness_seconds", json={"value": bad} ) assert resp.status_code == 422 assert not local_yaml.exists() assert ( tc.get("/admin/api/config").json()["session_cache.staleness_seconds"]["value"] == 1200 ) def test_config_staleness_window_is_allowlisted_at_the_right_path(): """A wrong path would write a key StrictModel forbids, 422ing every save.""" assert _CONFIG_ALLOWLIST["session_cache.staleness_seconds"] == ( "session_cache", "staleness_seconds", ) def test_config_POST_rejects_non_allowlisted_key(client): """classifier.base_url (and any off-allowlist key) is refused with 403.""" tc, _ = client resp = tc.post("/admin/api/config/classifier.base_url", json={"value": "http://x"}) assert resp.status_code == 403 def test_config_POST_invalid_value_leaves_file_unchanged(client): """quality_tolerance=1.5 (>1) fails RouterConfig validation -> 422, no write.""" tc, config_yaml = client before_base = config_yaml.read_text() local_yaml = config_yaml.with_name("config.local.yaml") before_local = local_yaml.read_text() if local_yaml.exists() else "" resp = tc.post( "/admin/api/config/objective.quality_tolerance", json={"value": 1.5} ) assert resp.status_code == 422 assert config_yaml.read_text() == before_base assert ( local_yaml.read_text() if local_yaml.exists() else "" ) == before_local def test_config_GET_includes_routing_default_profile(client): """GET /admin/api/config exposes routing.default_profile with the repo default.""" tc, _ = client resp = tc.get("/admin/api/config") assert resp.status_code == 200 body = resp.json() assert "routing.default_profile" in body assert body["routing.default_profile"] == { "value": "default", "source": "base", "category": "routing_quality_cost", "advanced": False, } def test_config_POST_default_profile_persists_valid_builtin(client): """POST routing.default_profile=batch persists and keeps comments.""" tc, config_yaml = client local_yaml = config_yaml.with_name("config.local.yaml") _insert_sentinel(config_yaml) resp = tc.post( "/admin/api/config/routing.default_profile", json={"value": "batch"} ) assert resp.status_code == 200 body = resp.json() assert body["key"] == "routing.default_profile" assert body["value"] == "batch" # This key is the one allowlisted setting that does NOT need a bounce: the # write path also applies it to the running cfg, because the router reads # cfg.routing.default_profile per request. Every other key here still # answers "restart is required", so the message is deliberately narrow -- # see test_config_POST_other_keys_still_say_restart below. assert "no restart required" in body["message"] assert "restart is required" not in body["message"] # Base comments preserved; value lives in overlay. base_text = config_yaml.read_text() assert _SENTINEL in base_text local_text = local_yaml.read_text() assert re.search( r'^\s*default_profile:\s*["\']?batch["\']?\s*$', local_text, re.MULTILINE ) is not None def test_config_POST_other_keys_still_say_restart(client): """The no-restart answer is scoped to routing.default_profile alone. Every other allowlisted key really does bind at import -- a provider's base_url, the log level -- so a blanket "no restart required" would be a lie that costs someone an afternoon. This is the guard against the narrow exception widening by accident. """ tc, _config_yaml = client resp = tc.post("/admin/api/config/logging.level", json={"value": "DEBUG"}) assert resp.status_code == 200 assert "restart is required" in resp.json()["message"] def test_config_POST_default_profile_rejects_unknown_and_leaves_overlay_unchanged( client, ): """Unknown default_profile returns 422 and does not touch config.local.yaml.""" tc, config_yaml = client local_yaml = config_yaml.with_name("config.local.yaml") before_base = config_yaml.read_text() before_local = local_yaml.read_text() if local_yaml.exists() else "" resp = tc.post( "/admin/api/config/routing.default_profile", json={"value": "nosuchprofile"}, ) assert resp.status_code == 422 detail = resp.json()["detail"] assert "nosuchprofile" in detail assert "default" in detail or "batch" in detail assert config_yaml.read_text() == before_base assert ( local_yaml.read_text() if local_yaml.exists() else "" ) == before_local def test_config_POST_creates_overlay_backup_before_write(client, tmp_path): """A successful write produces a ``config.local.yaml.bak.`` backup copy.""" tc, config_yaml = client local_yaml = config_yaml.with_name("config.local.yaml") _insert_sentinel(config_yaml) resp = tc.post("/admin/api/config/circuit_breaker.enabled", json={"value": True}) assert resp.status_code == 200 backups = sorted(tmp_path.glob("config/config.local.yaml.bak.*")) assert len(backups) == 1 # The backup captured the pre-write empty overlay (just the header). backup_text = backups[0].read_text() assert backup_text.strip() != "" # The current overlay carries the new value. local_text = local_yaml.read_text() assert yaml.safe_load(local_text)["circuit_breaker"]["enabled"] is True def test_config_concurrent_writes_are_atomic_no_zero_byte_backups(tmp_path): """Concurrent writes never truncate config.yaml or leave 0-byte backups.""" config_yaml = tmp_path / "config.yaml" shutil.copyfile(ROOT / "config" / "config.yaml", config_yaml) path = _CONFIG_ALLOWLIST["logging.level"] stop_reader = threading.Event() def reader() -> None: while not stop_reader.is_set(): try: text = config_yaml.read_text() except FileNotFoundError: continue assert text.strip() != "", "config.yaml observed empty" assert "logging:" in text reader_thread = threading.Thread(target=reader) reader_thread.start() def writer(level: str) -> None: _persist_config_value(config_yaml, path, level) threads = [ threading.Thread(target=writer, args=("warning",)), threading.Thread(target=writer, args=("error",)), ] for t in threads: t.start() for t in threads: t.join() stop_reader.set() reader_thread.join() final = config_yaml.read_text() assert re.search( r"^\s*level:\s*(warning|error)\s*$", final, re.MULTILINE ) is not None backups = list(tmp_path.glob("config.yaml.bak.*")) assert backups, "expected at least one backup" for b in backups: assert b.stat().st_size > 0, f"zero-byte backup: {b}" assert b.read_text().strip() != "" def test_profile_create_writes_overlay_and_leaves_base_unchanged( client, tmp_path ): """A profile CRUD write creates config.local.yaml and leaves config.yaml bytes unchanged.""" tc, config_yaml = client _insert_sentinel(config_yaml) base_before = config_yaml.read_bytes() local_yaml = config_yaml.with_name("config.local.yaml") resp = tc.post( "/admin/api/profiles/", json={"name": "minit", "max_tier": 2}, ) assert resp.status_code == 200 body = resp.json() assert body["profile"]["name"] == "minit" assert body["profile"]["definition"]["max_tier"] == 2 assert body["profile"]["source"] == "overlay" assert config_yaml.read_bytes() == base_before assert local_yaml.exists() local_text = local_yaml.read_text() assert "profiles:" in local_text assert "minit:" in local_text assert "max_tier: 2" in local_text def test_profile_create_creates_overlay_backup_before_write(client, tmp_path): tc, config_yaml = client local_yaml = config_yaml.with_name("config.local.yaml") _insert_sentinel(config_yaml) resp = tc.post( "/admin/api/profiles/", json={"name": "minit", "max_tier": 2}, ) assert resp.status_code == 200 backups = sorted(tmp_path.glob("config/config.local.yaml.bak.*")) assert len(backups) == 1 backup_text = backups[0].read_text() assert "profiles:" not in backup_text assert local_yaml.exists() local_text = local_yaml.read_text() assert "profiles:" in local_text assert "minit:" in local_text def test_profile_create_blocked_nested_path_returns_403(client): tc, _ = client resp = tc.post("/admin/api/config/profiles.foo", json={"value": {}}) assert resp.status_code == 403 def test_profile_create_builtin_name_returns_403(client): tc, _ = client resp = tc.post("/admin/api/profiles/", json={"name": "batch", "max_tier": 2}) assert resp.status_code == 403 assert "built-in" in resp.json()["detail"].lower() def test_persist_to_merged_validation_refuses_invalid_block(tmp_path): """Merged-config guard is enforced at the helper level, not the endpoint. ``admin_profile_create`` validates ``RoutingProfile(**profile_dict)`` first, so an endpoint-level collision with a built-in name is impossible. This test directly exercises ``_persist_to`` to prove that when the merged base + overlay trigger an invalid RouterConfig, the overlay validation raises ``ValidationError`` and neither file is modified. """ (tmp_path / "config").mkdir(parents=True, exist_ok=True) base_yaml = tmp_path / "config" / "config.yaml" shutil.copyfile(ROOT / "config" / "config.yaml", base_yaml) overlay_yaml = tmp_path / "config" / "config.local.yaml" base_before = base_yaml.read_bytes() overlay_before = overlay_yaml.read_bytes() if overlay_yaml.exists() else b"" # ``default`` is a built-in profile name; merging an overlay profile named # ``default`` into a base with no such profile conflicts with RouterConfig's # builtins-vs-config validation. profile_dict = {"max_tier": 2} with pytest.raises(ValidationError): _persist_to( base_yaml, overlay_yaml, ("profiles", "default"), profile_dict, ) assert base_yaml.read_bytes() == base_before assert ( overlay_yaml.read_bytes() if overlay_yaml.exists() else b"" ) == overlay_before def test_config_GET_corrupt_yaml_returns_clean_503(tmp_path): """A duplicate-key config.yaml refuses with 503, not a raw 500.""" (tmp_path / "config").mkdir(parents=True, exist_ok=True) config_yaml = tmp_path / "config" / "config.yaml" config_yaml.write_text( (ROOT / "config" / "config.yaml").read_text() + "profiles:\n a:\n min_tier: 1\nprofiles:\n b:\n min_tier: 2\n" ) cfg = load_config(ROOT / "config" / "config.yaml") def _db_factory() -> sqlite3.Connection: return _make_db(tmp_path) router = build_router(cfg, _db_factory, base_dir=str(tmp_path)) app = FastAPI() app.include_router(router, prefix="/admin") tc = TestClient(app, raise_server_exceptions=False) resp = tc.get("/admin/api/config") assert resp.status_code == 503 assert "config.yaml" in resp.json()["detail"] def test_config_concurrent_block_writes_are_atomic_no_zero_byte_backups(tmp_path): """Concurrent _persist_config_block writes never truncate config or leave 0-byte backups.""" config_yaml = tmp_path / "config.yaml" shutil.copyfile(ROOT / "config" / "config.yaml", config_yaml) stop_reader = threading.Event() def reader() -> None: while not stop_reader.is_set(): try: text = config_yaml.read_text() except FileNotFoundError: continue assert text.strip() != "", "config.yaml observed empty" assert "logging:" in text reader_thread = threading.Thread(target=reader) reader_thread.start() def writer(name: str, max_tier: int) -> None: _persist_config_block(config_yaml, ("profiles", name), {"max_tier": max_tier}) threads = [ threading.Thread(target=writer, args=("alpha", 1)), threading.Thread(target=writer, args=("beta", 2)), ] for t in threads: t.start() for t in threads: t.join() stop_reader.set() reader_thread.join() final = yaml.safe_load(config_yaml.read_text()) assert final["profiles"]["alpha"]["max_tier"] == 1 assert final["profiles"]["beta"]["max_tier"] == 2 backups = list(tmp_path.glob("config.yaml.bak.*")) assert backups, "expected at least one backup" for b in backups: assert b.stat().st_size > 0, f"zero-byte backup: {b}" assert b.read_text().strip() != "" # --------------------------------------------------------------------------- # Overlay survivability / provenance edge-case tests # --------------------------------------------------------------------------- _OVERLAY_HEADER_PREFIX = "# Machine-local overlay" def test_overlay_survives_git_restore_of_base(tmp_path): """A machine-local overlay (config.local.yaml) survives a git checkout of the base file — it is gitignored and therefore unaffected by ``git checkout -- config/config.yaml``.""" repo = tmp_path / "repo" repo.mkdir() (repo / "config").mkdir() base_yaml = repo / "config" / "config.yaml" shutil.copyfile(ROOT / "config" / "config.yaml", base_yaml) local_yaml = repo / "config" / "config.local.yaml" local_yaml.write_text( f"{_OVERLAY_HEADER_PREFIX} — gitignored, never committed.\n" "logging:\n level: warning\n" ) subprocess.run( ["git", "init", "-q", str(repo)], check=True, capture_output=True, ) subprocess.run( ["git", "-C", str(repo), "config", "user.email", "test@test.com"], check=True, capture_output=True, ) subprocess.run( ["git", "-C", str(repo), "config", "user.name", "test"], check=True, capture_output=True, ) (repo / ".gitignore").write_text("config.local.yaml\n") subprocess.run( ["git", "-C", str(repo), "add", "config/config.yaml", ".gitignore"], check=True, capture_output=True, ) subprocess.run( ["git", "-C", str(repo), "commit", "-m", "init", "-q"], check=True, capture_output=True, ) text = base_yaml.read_text() base_yaml.write_text(text.replace('level: "info"', 'level: "debug"')) subprocess.run( ["git", "-C", str(repo), "checkout", "--", "config/config.yaml"], check=True, capture_output=True, ) assert local_yaml.exists() overlay_text = local_yaml.read_text() assert "level: warning" in overlay_text restored = base_yaml.read_text() assert 'level: "info"' in restored or "level: info" in restored def test_config_GET_shows_overlay_source_when_overlay_exists(client): """When ``config.local.yaml`` overrides one allowlisted key, GET /admin/api/config reports ``source: "overlay"`` for that key and ``source: "base"`` for the rest.""" tc, config_yaml = client local_yaml = config_yaml.with_name("config.local.yaml") local_yaml.write_text( f"{_OVERLAY_HEADER_PREFIX} — gitignored, never committed.\n" "logging:\n level: warning\n" ) resp = tc.get("/admin/api/config") assert resp.status_code == 200 body = resp.json() assert body["logging.level"]["source"] == "overlay" assert body["logging.level"]["value"] == "warning" assert body["objective.quality_tolerance"]["source"] == "base" def test_first_write_creates_overlay_with_header(client): """POST to an allowlisted key when no overlay exists creates ``config.local.yaml`` starting with the expected comment header.""" tc, config_yaml = client local_yaml = config_yaml.with_name("config.local.yaml") assert not local_yaml.exists(), "overlay must not exist before the write" resp = tc.post( "/admin/api/config/circuit_breaker.enabled", json={"value": True} ) assert resp.status_code == 200 assert local_yaml.exists() text = local_yaml.read_text() assert text.startswith("# Machine-local overlay") assert "# Written by the admin portal" in text local = yaml.safe_load(text) assert local["circuit_breaker"]["enabled"] is True def test_second_write_keeps_header(client): """A second admin write preserves the machine-local header comment.""" tc, config_yaml = client local_yaml = config_yaml.with_name("config.local.yaml") resp = tc.post("/admin/api/config/logging.level", json={"value": "warning"}) assert resp.status_code == 200 resp = tc.post("/admin/api/config/circuit_breaker.enabled", json={"value": True}) assert resp.status_code == 200 text = local_yaml.read_text() assert text.startswith("# Machine-local overlay") # The header must appear exactly once. assert text.count("# Machine-local overlay") == 1 local = yaml.safe_load(text) assert local["logging"]["level"] == "warning" assert local["circuit_breaker"]["enabled"] is True