peter
33dec56e45
The admin page's download_worker_concurrency and log_level were stored but never reached the worker: it read the env once at boot (worker.py) and never looked again. The worker now re-reads both every 10 s (_sync_knobs): extra loop threads start at once, surplus ones retire after their current job, and a changed level is applied to the process. An unreadable database keeps what already runs.
110 lines
3.6 KiB
Python
110 lines
3.6 KiB
Python
"""The worker obeys the two admin knobs that concern it — `download_worker_concurrency` and
|
|
`log_level` — as stored in the settings page, not only the env it booted with. It used to read
|
|
`settings.download_worker_concurrency` and `settings.log_level` once at start, so the admin page
|
|
showed values the worker never applied. No DB or real threads of work here: the stored values come
|
|
from a patched `sysconfig`, and the loop threads are stand-ins that only record their index."""
|
|
|
|
import threading
|
|
|
|
import pytest
|
|
|
|
from app import worker
|
|
|
|
|
|
class _NoSession:
|
|
def __enter__(self):
|
|
return None
|
|
|
|
def __exit__(self, *exc):
|
|
return False
|
|
|
|
|
|
@pytest.fixture
|
|
def stored(monkeypatch):
|
|
"""The settings page's values, served to the worker instead of a database."""
|
|
values = {"download_worker_concurrency": 3, "log_level": "DEBUG"}
|
|
monkeypatch.setattr(worker, "SessionLocal", _NoSession)
|
|
monkeypatch.setattr(worker.sysconfig, "get_int", lambda _db, key: int(values[key]))
|
|
monkeypatch.setattr(worker.sysconfig, "get_str", lambda _db, key: str(values[key]))
|
|
return values
|
|
|
|
|
|
@pytest.fixture
|
|
def levels(monkeypatch):
|
|
applied: list[str] = []
|
|
monkeypatch.setattr(worker.logs, "set_level", applied.append)
|
|
return applied
|
|
|
|
|
|
@pytest.fixture
|
|
def loops(monkeypatch):
|
|
"""Stand-in loop threads: each records its index and parks until the worker stops."""
|
|
started: list[int] = []
|
|
lock = threading.Lock()
|
|
|
|
def fake_loop(idx: int) -> None:
|
|
with lock:
|
|
started.append(idx)
|
|
worker._stop.wait()
|
|
|
|
monkeypatch.setattr(worker, "_worker_loop", fake_loop)
|
|
yield started
|
|
worker._stop.set()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clean_stop():
|
|
worker._stop.clear()
|
|
yield
|
|
worker._stop.clear()
|
|
|
|
|
|
def test_worker_boots_with_the_stored_concurrency_and_log_level(monkeypatch, stored, levels, loops):
|
|
monkeypatch.setattr(worker.settings, "download_worker_concurrency", 2)
|
|
monkeypatch.setattr(worker.settings, "worker_enabled", True)
|
|
monkeypatch.setattr(worker.logs, "configure_logging", lambda _source: None)
|
|
monkeypatch.setattr(worker.signal, "signal", lambda *_a: None)
|
|
monkeypatch.setattr(worker, "_wait_for_schema", lambda: True)
|
|
monkeypatch.setattr(worker.logsink, "start", lambda: None)
|
|
monkeypatch.setattr(worker, "_recover_orphans", lambda: None)
|
|
|
|
threading.Timer(0.5, worker._stop.set).start()
|
|
worker.main()
|
|
|
|
assert sorted(loops) == [0, 1, 2] # the stored 3, not the env's 2
|
|
assert levels and levels[-1] == "DEBUG"
|
|
|
|
|
|
def test_a_changed_knob_applies_without_a_restart(stored, levels, loops):
|
|
threads: dict[int, threading.Thread] = {}
|
|
level = worker._sync_knobs(threads, None)
|
|
assert level == "DEBUG"
|
|
assert worker._target == 3
|
|
|
|
stored["download_worker_concurrency"] = 5
|
|
stored["log_level"] = "WARNING"
|
|
level = worker._sync_knobs(threads, level)
|
|
|
|
assert worker._target == 5
|
|
assert sorted(threads) == [0, 1, 2, 3, 4]
|
|
assert level == "WARNING" and levels[-1] == "WARNING"
|
|
|
|
# Lowering it retires the surplus threads after their current job: they see the lower target.
|
|
stored["download_worker_concurrency"] = 1
|
|
worker._sync_knobs(threads, level)
|
|
assert worker._target == 1
|
|
|
|
|
|
def test_an_unreadable_database_keeps_what_runs(monkeypatch, levels, loops):
|
|
def boom():
|
|
raise RuntimeError("db down")
|
|
|
|
monkeypatch.setattr(worker, "SessionLocal", boom)
|
|
monkeypatch.setattr(worker, "_target", 2)
|
|
threads: dict[int, threading.Thread] = {}
|
|
|
|
assert worker._sync_knobs(threads, "INFO") == "INFO"
|
|
assert worker._target == 2
|
|
assert sorted(threads) == [0, 1] # the loops still run on the boot value
|
|
assert levels == []
|