backend/tests/test_context_executor.py
Rama master · réplica de lectura del repositorio original.
"""I7: scheduler run labels must reach the pool workers a run spawns.
The label rides on the cancel-event contextvar, and contextvars do not cross
into ThreadPoolExecutor workers. Only wanted_search_runner copied the context,
so every other pool a scheduled job used logged `[-]` for its own work.
"""
from __future__ import annotations
import threading
from concurrent.futures import ThreadPoolExecutor
from services.scheduler import cancellation
def _in_run(job_id: str, fn, *, cancelled: bool = False):
event = cancellation.begin_run(job_id)
if cancelled:
event.set()
token = cancellation.activate(event)
try:
return fn(), cancellation.current_run_label()
finally:
cancellation.deactivate(token)
cancellation.end_run(job_id, event)
class TestSubmitWithContext:
def test_plain_submit_loses_the_label(self):
"""Documents the defect the helper exists for."""
def work():
with ThreadPoolExecutor(max_workers=1) as ex:
return ex.submit(cancellation.current_run_label).result()
seen, label = _in_run("wanted_scanner", work)
assert label is not None
assert seen is None
def test_helper_carries_the_label(self):
from utils.context_executor import submit_with_context
def work():
with ThreadPoolExecutor(max_workers=1) as ex:
return submit_with_context(ex, cancellation.current_run_label).result()
seen, label = _in_run("wanted_scanner", work)
assert seen == label
def test_arguments_pass_through(self):
from utils.context_executor import submit_with_context
with ThreadPoolExecutor(max_workers=1) as ex:
got = submit_with_context(ex, lambda a, b, c=None: (a, b, c), 1, 2, c=3).result()
assert got == (1, 2, 3)
def test_runner_alias_is_the_shared_helper(self):
from services.wanted_search_runner import _submit_with_context
from utils.context_executor import submit_with_context
assert _submit_with_context is submit_with_context
def _worker_view():
"""What a pool worker sees: the log id its records carry, and the stop flag."""
from app_logging import _current_request_id
return _current_request_id(), cancellation.abort_requested()
class TestSubmitWithRunLabel:
"""Review I-2: copying the WHOLE context also handed workers the run's cancel
signal (and any other contextvar). The label-only helper carries the log id
and nothing else."""
def test_worker_logs_under_the_run_label(self):
from utils.context_executor import submit_with_run_label
def work():
with ThreadPoolExecutor(max_workers=1) as ex:
return submit_with_run_label(ex, _worker_view).result()
(log_id, _stop), label = _in_run("wanted_scanner", work)
assert log_id == label
def test_worker_does_not_inherit_the_cancel_signal(self):
from utils.context_executor import submit_with_run_label
def work():
with ThreadPoolExecutor(max_workers=1) as ex:
return submit_with_run_label(ex, _worker_view).result()
(_log_id, stop), _ = _in_run("wanted_scanner", work, cancelled=True)
assert stop is False
def test_label_does_not_stick_to_the_pool_thread(self):
from utils.context_executor import submit_with_run_label
with ThreadPoolExecutor(max_workers=1) as ex:
_in_run("wanted_scanner", lambda: submit_with_run_label(ex, _worker_view).result())
log_id, _ = ex.submit(_worker_view).result()
assert log_id == "-"
def test_outside_a_run_nothing_is_attached(self):
from utils.context_executor import submit_with_run_label
with ThreadPoolExecutor(max_workers=1) as ex:
log_id, stop = submit_with_run_label(ex, _worker_view).result()
assert (log_id, stop) == ("-", False)
def test_arguments_pass_through(self):
from utils.context_executor import submit_with_run_label
with ThreadPoolExecutor(max_workers=1) as ex:
got = submit_with_run_label(ex, lambda a, b, c=None: (a, b, c), 1, 2, c=3).result()
assert got == (1, 2, 3)
class TestCallSitesPropagate:
"""The sites converted in rc.2 log under the run label but must NOT see a
cancel: a cancelled run's queued probes were refused by the media gate,
stored as "no streams", and upserted as wrong wanted rows."""
def test_batch_probe_workers_log_under_the_label_without_the_stop(self, monkeypatch):
import services.wanted_item_scanner as scanner
seen: list[tuple] = []
lock = threading.Lock()
def _probe(path, _flag):
with lock:
seen.append(_worker_view())
return {}
monkeypatch.setattr(scanner, "get_media_streams", _probe)
monkeypatch.setattr(
scanner,
"get_settings",
lambda: type("S", (), {"scan_metadata_max_workers": 2})(),
)
_, label = _in_run(
"wanted_scanner",
lambda: scanner.batch_probe(["/a.mkv", "/b.mkv"]),
cancelled=True,
)
# A stopped batch may drop probes that had not started yet; every probe
# that did run logs under the label and does not see the stop.
assert seen and all(view == (label, False) for view in seen), seen
def test_wanted_batch_workers_log_under_the_run_label(self, monkeypatch):
from types import SimpleNamespace
from wanted_search import batch
seen: list[tuple] = []
lock = threading.Lock()
def _process(item_id):
with lock:
seen.append(_worker_view())
return {"status": "found"}
monkeypatch.setattr(batch, "process_wanted_item", _process)
monkeypatch.setattr(
batch, "get_settings", lambda: SimpleNamespace(wanted_max_search_attempts=5)
)
monkeypatch.setattr(
batch, "get_wanted_item", lambda iid: {"id": iid, "search_count": 0, "title": "x"}
)
_, label = _in_run(
"wanted_search", lambda: list(batch.process_wanted_batch(item_ids=[1, 2]))
)
assert len(seen) == 2 and all(v == (label, False) for v in seen), seen