An interval override was a row of its own table beside the one store every other admin knob uses; it is now the `job_interval_<job>` sysconfig key (migration 0110 copies them over and drops the table). The Scheduler page stays its one editor: the Configuration page and the setup wizard neither list nor write those keys, since a write there would skip the reschedule. Each ConfigSpec now states `applies` (required, no default). Every key is `live`, held by a scan that finds no read at import time; a `restart` key is refused until the page can say so. A second scan requires a reader for every key on the Configuration page — `sponsorblock_default` had none since it was added, so it leaves the registry, the translations and the store.
183 lines
7.4 KiB
Python
183 lines
7.4 KiB
Python
"""The sysconfig registry: a group's keys stay together, every group has its section, and a closed-set
|
|
knob is a CHOICE.
|
|
|
|
The admin Configuration page used to keep its own hand-written group order, which had already drifted
|
|
(`explore` was missing, so its tab sorted last). The page now follows the registry: SECTIONS orders the
|
|
groups under their headings, and a group missing there would fall off the page — so each group
|
|
of SPECS sits in exactly one section.
|
|
|
|
A closed-set admin knob is a CHOICE, not free text.
|
|
|
|
`log_level`, `search_source`, `download_layout` and the transcode `hw`/`preset` used to be free-text
|
|
fields: the admin could type anything and every consumer carried its own `or "auto").lower()` guess
|
|
at what a typo meant. Now the value set is a `Literal` in config.py, pydantic holds the env to it at
|
|
boot, sysconfig holds a stored override to it, and the admin page renders a picker from the same list.
|
|
The DB lane (`db` fixture) covers the store; the rest is pure logic.
|
|
"""
|
|
|
|
import ast
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from app import sysconfig
|
|
from app.config import Settings, choices_of
|
|
|
|
CHOICE_KEYS = {s.key for s in sysconfig.SPECS if s.type == "choice"}
|
|
|
|
|
|
def test_each_group_is_one_run_of_keys():
|
|
runs = [
|
|
s.group
|
|
for i, s in enumerate(sysconfig.SPECS)
|
|
if i == 0 or sysconfig.SPECS[i - 1].group != s.group
|
|
]
|
|
assert len(runs) == len(set(runs)), runs
|
|
assert runs[:2] == ["access", "email"]
|
|
|
|
|
|
def test_every_group_sits_in_exactly_one_section():
|
|
listed = [g for _, members in sysconfig.SECTIONS for g in members]
|
|
assert len(listed) == len(set(listed)), listed
|
|
assert set(listed) | set(sysconfig.OWN_PAGE_GROUPS) == {s.group for s in sysconfig.SPECS}
|
|
assert not set(listed) & set(sysconfig.OWN_PAGE_GROUPS)
|
|
assert [sid for sid, _ in sysconfig.SECTIONS] == ["access", "youtube", "media", "system"]
|
|
|
|
|
|
def test_every_closed_set_knob_is_a_choice_whose_options_come_from_config():
|
|
assert CHOICE_KEYS == {
|
|
"search_source",
|
|
"download_layout",
|
|
"log_level",
|
|
"plex_transcode_hw",
|
|
"plex_transcode_preset",
|
|
}
|
|
for s in sysconfig.SPECS:
|
|
field = Settings.model_fields[s.default_attr]
|
|
if s.type == "choice":
|
|
assert s.choices, s.key
|
|
assert field.default in s.choices, s.key
|
|
else:
|
|
# A Literal-typed setting exposed as "str" would be a free-text field again.
|
|
assert s.choices == () and choices_of(field.annotation) == (), s.key
|
|
|
|
|
|
def test_the_env_accepts_any_letter_case_and_rejects_an_unknown_value():
|
|
assert Settings(log_level="debug").log_level == "DEBUG"
|
|
assert Settings(search_source=" API ").search_source == "api"
|
|
with pytest.raises(ValidationError):
|
|
Settings(log_level="verbose")
|
|
|
|
|
|
class _Row:
|
|
def __init__(self, key, value):
|
|
self.key, self.value, self.encrypted = key, value, False
|
|
|
|
|
|
class _FakeDb:
|
|
"""Just enough of a Session for sysconfig's reads: one stored override row per key."""
|
|
|
|
def __init__(self, **rows):
|
|
self.rows = {k: _Row(k, v) for k, v in rows.items()}
|
|
|
|
def get(self, _model, key):
|
|
return self.rows.get(key)
|
|
|
|
def scalars(self, _stmt):
|
|
# describe() reads every override at once.
|
|
return list(self.rows.values())
|
|
|
|
|
|
def test_a_stored_value_reads_back_canonical_and_a_stale_one_falls_back_to_the_default():
|
|
assert sysconfig.get_str(_FakeDb(plex_transcode_hw="OFF"), "plex_transcode_hw") == "off"
|
|
# A value that is no longer (or never was) a choice is not passed through to the consumer.
|
|
assert sysconfig.get_str(_FakeDb(plex_transcode_hw="maybe"), "plex_transcode_hw") == "auto"
|
|
|
|
|
|
def test_describe_hands_the_picker_its_options():
|
|
items = {i["key"]: i for g in sysconfig.describe(_FakeDb())["groups"].values() for i in g}
|
|
assert items["log_level"]["choices"] == ["TRACE", "DEBUG", "INFO", "WARNING", "ERROR"]
|
|
assert items["log_level"]["value"] == "INFO"
|
|
assert items["quota_daily_budget"]["choices"] == []
|
|
|
|
|
|
def test_describe_lists_every_group_under_its_section_in_order():
|
|
d = sysconfig.describe(_FakeDb())
|
|
assert [s["id"] for s in d["sections"]] == ["access", "youtube", "media", "system"]
|
|
assert d["sections"][0]["groups"] == ["access", "google", "email"]
|
|
listed = [g for s in d["sections"] for g in s["groups"]]
|
|
assert sorted(listed) == sorted(d["groups"])
|
|
|
|
|
|
def test_set_value_stores_the_canonical_spelling_and_refuses_anything_else(db):
|
|
sysconfig.set_value(db, "search_source", "API")
|
|
assert sysconfig.get_str(db, "search_source") == "api"
|
|
with pytest.raises(ValueError, match="scrape, api"):
|
|
sysconfig.set_value(db, "search_source", "bing")
|
|
db.rollback()
|
|
|
|
|
|
# --- when a changed value takes effect --------------------------------------------------------------
|
|
|
|
_APP = Path(__file__).resolve().parent.parent / "app"
|
|
_READERS = {"get", "get_str", "get_int", "get_bool"}
|
|
|
|
|
|
def _key_reads() -> list[tuple[str, str, bool]]:
|
|
"""Every `sysconfig.get*(db, "<key>")` in the app outside the registry: (key, where, inside a
|
|
function)."""
|
|
found: list[tuple[str, str, bool]] = []
|
|
|
|
def walk(node: ast.AST, path: Path, in_fn: bool) -> None:
|
|
for child in ast.iter_child_nodes(node):
|
|
fn = in_fn or isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda))
|
|
if (
|
|
isinstance(child, ast.Call)
|
|
and isinstance(child.func, ast.Attribute)
|
|
and isinstance(child.func.value, ast.Name)
|
|
and child.func.value.id == "sysconfig"
|
|
and child.func.attr in _READERS
|
|
and len(child.args) >= 2
|
|
and isinstance(child.args[1], ast.Constant)
|
|
and isinstance(child.args[1].value, str)
|
|
):
|
|
found.append((child.args[1].value, f"{path.name}:{child.lineno}", in_fn))
|
|
walk(child, path, fn)
|
|
|
|
for path in _APP.rglob("*.py"):
|
|
if path.name != "sysconfig.py":
|
|
walk(ast.parse(path.read_text(encoding="utf-8")), path, False)
|
|
return found
|
|
|
|
|
|
def test_every_key_states_when_it_applies_and_none_needs_a_restart_yet():
|
|
with pytest.raises(TypeError):
|
|
sysconfig.ConfigSpec("k", "int", "quota", "quota_daily_budget") # pyright: ignore[reportCallIssue]
|
|
# The Configuration page cannot tell the admin that a setting waits for a restart; a key that
|
|
# does needs that first (a badge on it and a banner once it changed).
|
|
assert {s.applies for s in sysconfig.SPECS} == {"live"}
|
|
|
|
|
|
def test_a_live_key_is_read_where_it_is_used_never_at_import():
|
|
reads = _key_reads()
|
|
assert len(reads) > 50, len(
|
|
reads
|
|
) # the scan sees the reads (it would pass vacuously otherwise)
|
|
assert [(k, at) for k, at, in_fn in reads if not in_fn] == []
|
|
|
|
|
|
def test_every_key_on_the_configuration_page_has_a_reader():
|
|
read = {k for k, _, _ in _key_reads()}
|
|
page = {s.key for s in sysconfig.SPECS if s.group not in sysconfig.OWN_PAGE_GROUPS}
|
|
assert sorted(page - read) == []
|
|
|
|
|
|
def test_the_intervals_are_keys_of_the_scheduler_page_only():
|
|
keys = {s.key for s in sysconfig.SPECS if s.group == "scheduler"}
|
|
assert keys == {sysconfig.interval_key(j) for j in sysconfig.JOB_INTERVAL_DEFAULTS}
|
|
assert all(len(k) <= 64 for k in keys) # system_config.key
|
|
assert sysconfig.config_page_spec("job_interval_rss_poll") is None
|
|
assert sysconfig.config_page_spec("quota_daily_budget") is not None
|
|
assert "scheduler" not in sysconfig.describe(_FakeDb())["groups"]
|