The audit log, the server log and the admin quota view were three pages (the quota one a Stats tab),
each with its own filters, and the background runs had no list at all. Now one page ("Napló" /
"Activity") reads all four, each still its own table on the server: a shared bar (text, a period of
1 h to 90 days defaulting to 7, an account) and each tab's own filters, the whole view in the URL.
The old page ids open their tab.
Server side: the audit log and the new run list (`/api/admin/runs`) filter and page like the server
log, through one keyset helper (`routes/_keyset`) the log now shares; an audit row keeps the request
it was made in (migration 0111), so its log lines are a pivot away, as a run's are (`lrun`). The
quota breakdown takes since/until and one account. The Scheduler rows get a history button to their
job's runs.
Also fixed on the way: re-saving the value in force on the Configuration page staged its override
without committing it (the commit sat inside the audit branch), so the save was lost.
195 lines
8.0 KiB
Python
195 lines
8.0 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.models import SystemConfig
|
|
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"]
|
|
|
|
|
|
def test_saving_the_value_in_force_still_stores_the_override(db, make_user, as_user):
|
|
client = as_user(make_user(role="admin"))
|
|
default = sysconfig.get_int(db, "quota_daily_budget")
|
|
r = client.patch("/api/admin/config/quota_daily_budget", json={"value": default})
|
|
assert r.status_code == 200, r.text
|
|
(item,) = [i for i in r.json()["groups"]["quota"] if i["key"] == "quota_daily_budget"]
|
|
assert item["is_set"] is True
|
|
db.expire_all()
|
|
assert db.get(SystemConfig, "quota_daily_budget") is not None
|