peter
bf99444ded
The Files library starts here. A user's every download, clip and upload now lands in a source folder made per origin (YouTube, Plex clips, uploads, one per other site), decided once in downloads/folders.py at every place a job is made. How long a file stays moves from the shared asset to the user's job: stamped from its folder when it is done, never for a file the user keeps. The GC warns each user once, then takes the expired file out of their library; the cache cap never evicts a kept file, and an evicted one leaves every holder's library. The media mount is /files now (FILES_ROOT, FILES_HOST_PATH, volume siftlode_files); docs/self-hosting.md says how to carry an old volume over. Migration 0102 places every existing job and carries each done file's expiry over from its asset. safe_abs_path takes a missing path and answers None, which drops one baselined type error.
619 lines
22 KiB
Python
619 lines
22 KiB
Python
"""A failure is reported once, with its code, and a repeating one is not a
|
|
flood.
|
|
|
|
Each case here is a place where the first sweep got the line wrong in one of two directions: a
|
|
failure that left no line at all (a download the watchdog killed), or one that left too many (the
|
|
same edit failure twice, a warning per segment request, per settings read).
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
import yt_dlp
|
|
|
|
from app import logs, sysconfig, worker
|
|
from app.downloads import pipeline
|
|
from app.downloads.stated import StatedFailure
|
|
from app.models import DownloadJob, MediaAsset
|
|
from app.plex import stream as plex_stream
|
|
|
|
|
|
class _Sink(logging.Handler):
|
|
def __init__(self):
|
|
super().__init__(level=logging.NOTSET)
|
|
self.lines: list[str] = []
|
|
self.setFormatter(logs.JsonFormatter())
|
|
self.addFilter(logs.ContextFilter())
|
|
|
|
def emit(self, record):
|
|
self.lines.append(self.format(record))
|
|
|
|
def objects(self) -> list[dict]:
|
|
return [json.loads(line) for line in self.lines]
|
|
|
|
def events(self, code: str) -> list[dict]:
|
|
return [o for o in self.objects() if o.get("event") == code]
|
|
|
|
|
|
@pytest.fixture
|
|
def sink():
|
|
logger = logging.getLogger("siftlode")
|
|
handler = _Sink()
|
|
previous = logger.level
|
|
logger.addHandler(handler)
|
|
logger.setLevel(logs.TRACE)
|
|
try:
|
|
yield handler
|
|
finally:
|
|
logger.removeHandler(handler)
|
|
logger.setLevel(previous)
|
|
|
|
|
|
class _NoDb:
|
|
"""A session that finds nothing and commits nothing — these cases are about the line."""
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
return False
|
|
|
|
def get(self, *a, **k):
|
|
return None
|
|
|
|
def commit(self):
|
|
pass
|
|
|
|
|
|
@pytest.fixture
|
|
def finalized(monkeypatch):
|
|
"""Whether `finalize_if_running` fires: True unless a case says a route got there first."""
|
|
state = {"fires": True}
|
|
monkeypatch.setattr(worker, "SessionLocal", _NoDb)
|
|
monkeypatch.setattr(pipeline, "SessionLocal", _NoDb)
|
|
fire = lambda db, job_id, **fields: state["fires"] # noqa: E731
|
|
monkeypatch.setattr(worker.service, "finalize_if_running", fire)
|
|
return state
|
|
|
|
|
|
# --- the worker's own failures ---------------------------------------------------------------
|
|
@pytest.mark.parametrize("cause", ["stalled", "timeout"])
|
|
def test_a_download_the_watchdog_kills_leaves_a_coded_line(sink, finalized, cause):
|
|
"""The child is killed, so it never reports; before this the worker did not either, and a
|
|
stalled download failed with no line anywhere."""
|
|
worker._fail_running(42, error="stopped", error_code="network", cause=cause)
|
|
|
|
(line,) = sink.events("download.job.failed")
|
|
assert line["level"] == "WARNING"
|
|
assert line["attrs"] == {"job_id": 42, "error_code": "network", "cause": cause}
|
|
|
|
|
|
def test_a_job_a_route_already_stopped_is_not_reported_as_failed(sink, finalized):
|
|
finalized["fires"] = False
|
|
worker._fail_running(42, error="stopped", error_code="network", cause="stalled")
|
|
|
|
assert sink.events("download.job.failed") == []
|
|
|
|
|
|
def test_a_caller_with_its_own_line_is_not_reported_twice(sink, finalized):
|
|
worker._fail_running(42, error="Worker error", error_code=None)
|
|
|
|
assert sink.events("download.job.failed") == []
|
|
|
|
|
|
def test_a_child_that_died_is_reported_with_its_exit_code(sink, finalized):
|
|
worker._fail_running(
|
|
42, error="exited", error_code="unknown", cause="child_exit", exit_code=137
|
|
)
|
|
|
|
(line,) = sink.events("download.job.failed")
|
|
assert line["attrs"]["exit_code"] == 137
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("fires", "code"), [(True, "download.job.failed"), (False, "download.worker.error")]
|
|
)
|
|
def test_a_worker_error_is_one_line_that_says_what_it_did_to_the_job(sink, fires, code):
|
|
"""A route's cancel that landed first is not a failure — the error is still reported, with
|
|
its stack, but not as the job failing."""
|
|
try:
|
|
raise KeyError("a worker bug")
|
|
except KeyError as exc:
|
|
worker._report_worker_error(42, exc, "crash", lambda: fires)
|
|
|
|
(line,) = [o for o in sink.objects() if o["level"] in ("WARNING", "ERROR")]
|
|
assert line["event"] == code
|
|
assert "KeyError" in line["exc"]
|
|
|
|
|
|
def test_a_worker_error_is_reported_even_when_failing_the_job_raises(sink):
|
|
def db_down():
|
|
raise ConnectionError("the database went away")
|
|
|
|
with pytest.raises(ConnectionError):
|
|
worker._report_worker_error(42, KeyError("a worker bug"), "crash", db_down)
|
|
|
|
(line,) = sink.events("download.worker.error")
|
|
assert line["attrs"]["cause"] == "crash"
|
|
|
|
|
|
def test_a_crash_the_database_will_not_record_does_not_end_the_worker_thread(monkeypatch, sink):
|
|
"""The crash handler writes to the database; if that raises too, the thread must go on to
|
|
the next job — a dead thread is a download slot gone until the worker restarts."""
|
|
claims = iter([42, 43])
|
|
seen = []
|
|
|
|
def claim():
|
|
job = next(claims, None)
|
|
if job is None:
|
|
worker._stop.set()
|
|
return job
|
|
|
|
def crash(job_id):
|
|
seen.append(job_id)
|
|
raise RuntimeError("boom")
|
|
|
|
def db_down(*a, **k):
|
|
raise ConnectionError("the database went away")
|
|
|
|
monkeypatch.setattr(worker, "_claim_job", claim)
|
|
monkeypatch.setattr(worker, "_process_job", crash)
|
|
monkeypatch.setattr(worker, "_fail_running", db_down)
|
|
monkeypatch.setattr(worker.settings, "download_poll_seconds", 0)
|
|
worker._stop.clear()
|
|
try:
|
|
worker._worker_loop(0)
|
|
finally:
|
|
worker._stop.clear()
|
|
assert seen == [42, 43]
|
|
|
|
|
|
class _StoppedProc:
|
|
"""A child that is still tearing down when the supervisor polls, after a route paused it."""
|
|
|
|
pid = 4242
|
|
returncode = None
|
|
|
|
def wait(self, timeout=None):
|
|
import subprocess
|
|
|
|
raise subprocess.TimeoutExpired("child", timeout or 0)
|
|
|
|
|
|
def test_the_worker_stopping_a_paused_child_is_not_a_second_abort_line(monkeypatch, sink):
|
|
"""The child logs `download.job.aborted` when it notices the pause itself; the worker's line
|
|
is its own fact — it had to stop the child — so one pause is never two aborts."""
|
|
monkeypatch.setattr(worker, "_spawn_download_child", lambda job_id: _StoppedProc())
|
|
monkeypatch.setattr(worker, "_job_liveness", lambda job_id: ("paused", 1.0))
|
|
monkeypatch.setattr(worker, "_stop_and_free", lambda *a: None)
|
|
|
|
worker._run_supervised(42, 7)
|
|
|
|
assert sink.events("download.job.aborted") == []
|
|
assert len(sink.events("download.child.stopped")) == 1
|
|
|
|
|
|
# --- an edit failure, once ---------------------------------------------------------------------
|
|
def test_an_edit_failure_is_one_line_with_its_code_and_our_own_stack(sink, finalized):
|
|
try:
|
|
raise TypeError("a spec shape nobody expected")
|
|
except TypeError as exc:
|
|
pipeline._fail_edit(9, 3, str(exc), exc)
|
|
|
|
(line,) = sink.events("edit.job.failed")
|
|
assert line["attrs"] == {"job_id": 9, "asset_id": 3, "error_type": "TypeError"}
|
|
assert "TypeError" in line["exc"]
|
|
assert len([o for o in sink.objects() if o["level"] in ("WARNING", "ERROR")]) == 1
|
|
|
|
|
|
def test_an_edit_that_ffmpeg_refused_names_no_file_and_carries_no_stack(sink, finalized):
|
|
"""ffmpeg's stderr tail names its input, whose path is the channel and the title. The message
|
|
is scrubbed; the stack is left off, because the message is the whole story."""
|
|
msg = "ffmpeg failed (1): /downloads/Some_Channel/Season_2024/A_Private_Title_a3.mp4: Invalid data"
|
|
pipeline._fail_edit(9, 3, msg, StatedFailure(msg))
|
|
|
|
(line,) = sink.events("edit.job.failed")
|
|
assert "A_Private_Title" not in json.dumps(line)
|
|
assert "exc" not in line
|
|
|
|
|
|
def test_an_edit_the_database_would_not_fail_is_not_reported_as_failed(sink, monkeypatch):
|
|
"""The job is still `running`: the runner reports this failure with the write's, and the
|
|
worker fails the job (`child_exit`). A line here as well would be a second, and a false one."""
|
|
|
|
class _Down:
|
|
def __enter__(self):
|
|
raise ConnectionError("the database went away")
|
|
|
|
def __exit__(self, *exc):
|
|
return False
|
|
|
|
monkeypatch.setattr(pipeline, "SessionLocal", _Down)
|
|
with pytest.raises(ConnectionError):
|
|
pipeline._fail_edit(9, 3, "boom", KeyError("x"))
|
|
|
|
assert sink.events("edit.job.failed") == []
|
|
|
|
|
|
class _EditDb(_NoDb):
|
|
"""An edit job whose source is ready, for `_process_edit` to get as far as its encode."""
|
|
|
|
def get(self, model, _id):
|
|
if model is DownloadJob:
|
|
return SimpleNamespace(source_asset_id=7, edit_spec={}, user_id=1)
|
|
if model is MediaAsset:
|
|
return SimpleNamespace(
|
|
status="ready", rel_path="src.mp4", width=640, height=360, duration_s=10
|
|
)
|
|
return None
|
|
|
|
|
|
def test_an_edit_that_crashes_is_said_once(sink, finalized, monkeypatch, tmp_path):
|
|
"""The line `_process_edit`'s handler used to log before `_fail_edit` logged it again."""
|
|
(tmp_path / "src.mp4").write_bytes(b"x")
|
|
monkeypatch.setattr(pipeline, "SessionLocal", _EditDb)
|
|
monkeypatch.setattr(pipeline.settings, "files_root", str(tmp_path))
|
|
monkeypatch.setattr(pipeline, "_edit_staging_dir", lambda asset_id: tmp_path / "staging")
|
|
|
|
def odd_spec(*a, **k):
|
|
raise TypeError("a spec shape nobody expected")
|
|
|
|
monkeypatch.setattr(pipeline, "_set_job", odd_spec)
|
|
|
|
pipeline._process_edit(9, 3)
|
|
|
|
(line,) = [o for o in sink.objects() if o["level"] in ("WARNING", "ERROR")]
|
|
assert line["event"] == "edit.job.failed"
|
|
assert "TypeError" in line["exc"]
|
|
|
|
|
|
def test_an_edit_whose_source_is_gone_is_found_by_the_same_code(sink, finalized):
|
|
pipeline._fail_edit(9, 3, "The source file is missing.")
|
|
|
|
(line,) = sink.events("edit.job.failed")
|
|
assert line["attrs"] == {"job_id": 9, "asset_id": 3}
|
|
|
|
|
|
# --- a download failure, scrubbed ---------------------------------------------------------------
|
|
def test_a_failed_download_line_does_not_put_back_what_the_adapter_took_out(sink):
|
|
"""yt-dlp logs its error through the adapter (scrubbed) and then raises the same text, which
|
|
the pipeline logs again. Unscrubbed, that second line is the URL the first one hid."""
|
|
msg = "ERROR: Unsupported URL: https://example.org/private/path?id=123"
|
|
pipeline._log_download_failed(5, msg, "unknown", yt_dlp.utils.DownloadError(msg))
|
|
|
|
(line,) = sink.events("download.job.failed")
|
|
assert "private/path" not in json.dumps(line)
|
|
assert "example.org/…" in line["msg"]
|
|
# Unclassified, but yt-dlp's own error: its stack is the `except`, its foot the raw text.
|
|
assert "exc" not in line
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"exc",
|
|
[
|
|
RuntimeError("dictionary changed size during iteration"),
|
|
RecursionError("maximum recursion depth exceeded"),
|
|
NotImplementedError(),
|
|
FileNotFoundError(2, "No such file", "/files/Some_Channel/Season_2024/A_Private_Title"),
|
|
],
|
|
ids=lambda e: type(e).__name__,
|
|
)
|
|
def test_a_bug_that_python_raises_keeps_its_stack_and_names_no_file(sink, exc):
|
|
"""Only the failures WE raise with the whole story in the message go without a stack. The
|
|
same types are what Python raises for a bug of ours, and then the stack is the only lead."""
|
|
try:
|
|
raise exc
|
|
except (RuntimeError, OSError) as caught:
|
|
pipeline._log_download_failed(5, str(caught), "unknown", caught)
|
|
|
|
(line,) = sink.events("download.job.failed")
|
|
assert type(exc).__name__ in line["exc"]
|
|
assert "A_Private_Title" not in json.dumps(line)
|
|
|
|
|
|
def test_a_stack_does_not_carry_the_download_error_it_was_raised_over(sink):
|
|
"""A database error raised inside `except DownloadError` has yt-dlp's text as its context,
|
|
and a plain `exc_info` prints that link unscrubbed."""
|
|
# A variable, as in the pipeline: a frame prints its SOURCE line, and that is code, not data.
|
|
text = "ERROR: [youtube] dQw4w9WgXcQ: Sign in to confirm"
|
|
try:
|
|
try:
|
|
raise yt_dlp.utils.DownloadError(text)
|
|
except yt_dlp.utils.DownloadError:
|
|
raise ConnectionError("server closed the connection unexpectedly") # noqa: B904
|
|
except ConnectionError as exc:
|
|
pipeline._log_download_failed(5, str(exc), "unknown", exc)
|
|
|
|
(line,) = sink.events("download.job.failed")
|
|
assert "During handling of the above exception" in line["exc"]
|
|
assert "Sign in to confirm" in line["exc"]
|
|
assert "dQw4w9WgXcQ" not in json.dumps(line)
|
|
|
|
|
|
def test_an_unclassified_failure_of_our_own_keeps_its_stack(sink):
|
|
try:
|
|
raise KeyError("requested_formats")
|
|
except KeyError as exc:
|
|
pipeline._log_download_failed(5, str(exc), "unknown", exc)
|
|
|
|
(line,) = sink.events("download.job.failed")
|
|
assert "KeyError" in line["exc"]
|
|
assert line["attrs"]["error_type"] == "KeyError"
|
|
|
|
|
|
def test_the_metadata_backfill_names_the_asset_and_not_the_download(sink, monkeypatch):
|
|
from app.downloads import adapters, backfill
|
|
|
|
url = "https://www.youtube.com/watch?v=dQw4w9WgXcQ"
|
|
|
|
class _Refusing:
|
|
def __init__(self, opts):
|
|
pass
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
return False
|
|
|
|
def extract_info(self, target, download):
|
|
raise yt_dlp.utils.DownloadError(
|
|
f"ERROR: [youtube] dQw4w9WgXcQ: Video unavailable ({target})"
|
|
)
|
|
|
|
monkeypatch.setattr(backfill.yt_dlp, "YoutubeDL", _Refusing)
|
|
assert backfill._extract_meta(url, adapters.YouTubeAdapter(), 11) is None
|
|
|
|
lines = [o for o in sink.objects() if "meta extract failed" in o["msg"]]
|
|
assert lines and all("dQw4w9WgXcQ" not in json.dumps(o) for o in lines)
|
|
|
|
|
|
def test_the_metadata_backfill_takes_out_a_generic_name_whatever_it_holds(sink, monkeypatch):
|
|
from app.downloads import adapters, backfill
|
|
|
|
url = "https://example.org/v/Our%20Trip%3A%20Day%201.mp4"
|
|
|
|
class _Refusing:
|
|
def __init__(self, opts):
|
|
pass
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
return False
|
|
|
|
def extract_info(self, target, download):
|
|
raise yt_dlp.utils.DownloadError("ERROR: [generic] Our Trip: Day 1: Unable to download")
|
|
|
|
monkeypatch.setattr(backfill.yt_dlp, "YoutubeDL", _Refusing)
|
|
assert backfill._extract_meta(url, adapters.SourceAdapter(), 11) is None
|
|
|
|
lines = [o for o in sink.objects() if "meta extract failed" in o["msg"]]
|
|
assert lines and all("Day 1" not in json.dumps(o) for o in lines)
|
|
|
|
|
|
# --- a repeating failure, once ------------------------------------------------------------------
|
|
def test_a_stale_setting_is_reported_once_not_on_every_read(sink, monkeypatch):
|
|
spec = next(s for s in sysconfig.SPECS if s.type == "int")
|
|
monkeypatch.setattr(sysconfig, "_raw_override", lambda db, key: "not-a-number")
|
|
monkeypatch.setattr(sysconfig, "_coerce_warned", set())
|
|
|
|
for _ in range(3):
|
|
sysconfig.get(None, spec.key)
|
|
|
|
assert len(sink.events("sysconfig.override_uncoercible")) == 1
|
|
|
|
|
|
class _RefusingDir:
|
|
"""A session directory on a disk that has gone away: every probe raises."""
|
|
|
|
def __truediv__(self, name):
|
|
return SimpleNamespace(exists=self._refuse)
|
|
|
|
@staticmethod
|
|
def _refuse():
|
|
raise PermissionError("the disk went read-only")
|
|
|
|
|
|
def test_a_refusing_session_directory_is_reported_once_per_session(sink):
|
|
"""Every segment request probes, and hls.js retries each 404 — a line per probe is a flood."""
|
|
session = plex_stream.HlsSession("k", Path("."), None, 0.0, ())
|
|
session.dir = _RefusingDir()
|
|
|
|
for seq in (3, 4, 5):
|
|
assert not plex_stream.artifact_may_arrive(session, plex_stream.segment_name(seq, None))
|
|
|
|
(line,) = sink.events("plex.stream.segment_probe_failed")
|
|
assert line["attrs"]["reason"] == "PermissionError"
|
|
|
|
|
|
def test_a_retried_attempt_names_the_host_and_not_the_video(sink, monkeypatch, tmp_path):
|
|
"""The first attempt's DownloadError is logged when the fallback set takes over — one line
|
|
before the adapter's scrubbed copy of the same text would have hidden it."""
|
|
url = "https://www.youtube.com/watch?v=dQw4w9WgXcQ"
|
|
attempts = iter(
|
|
[yt_dlp.utils.DownloadError(f"ERROR: [youtube] dQw4w9WgXcQ: Sign in ({url})"), {"id": "x"}]
|
|
)
|
|
|
|
class _YDL:
|
|
def __init__(self, opts):
|
|
pass
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
return False
|
|
|
|
def extract_info(self, target, download):
|
|
outcome = next(attempts)
|
|
if isinstance(outcome, Exception):
|
|
raise outcome
|
|
return outcome
|
|
|
|
adapter = SimpleNamespace(
|
|
client_sets=lambda authed: [["web"], ["tv"]],
|
|
resolve_url=lambda target, opts: target,
|
|
ydl_opts_extra=lambda clients: {},
|
|
retriable=lambda text: True,
|
|
)
|
|
monkeypatch.setattr(pipeline.yt_dlp, "YoutubeDL", _YDL)
|
|
monkeypatch.setattr(pipeline, "_touch_job", lambda job_id: None)
|
|
monkeypatch.setattr(pipeline, "_set_job", lambda job_id, **fields: None)
|
|
|
|
assert pipeline._extract_with_fallback(1, 1, url, {}, tmp_path, adapter) == {"id": "x"}
|
|
|
|
(line,) = [o for o in sink.objects() if "retrying with fallback" in o["msg"]]
|
|
assert "dQw4w9WgXcQ" not in json.dumps(line)
|
|
|
|
|
|
@pytest.mark.skipif(shutil.which("ffmpeg") is None, reason="ffmpeg not available")
|
|
def test_a_failed_poster_or_storyboard_names_no_file(sink, tmp_path):
|
|
"""ffmpeg's CalledProcessError quotes its argv, and the source path in it is built from the
|
|
channel and the title."""
|
|
from app.downloads import edit as editmod
|
|
|
|
rel = "Some_Channel/Season_2024/Some_Channel_-_20240101_-_A_Private_Title_a3.mp4"
|
|
(tmp_path / rel).parent.mkdir(parents=True)
|
|
(tmp_path / rel).write_bytes(b"not a video")
|
|
asset = SimpleNamespace(id=3, rel_path=rel, duration_s=10.0, storyboard_path=None)
|
|
|
|
assert editmod.ensure_poster(asset, str(tmp_path)) is None
|
|
assert editmod.ensure_storyboard(asset, str(tmp_path)) is None
|
|
|
|
lines = [o["msg"] for o in sink.objects() if "gen skipped" in o["msg"]]
|
|
# The whole message: a long tmp root would push a leaked argv's path past any length cut.
|
|
assert len(lines) == 2
|
|
assert all(
|
|
re.fullmatch(r"\w+ gen skipped for asset 3: CalledProcessError \d+", m) for m in lines
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"err",
|
|
[
|
|
subprocess.TimeoutExpired(["ffmpeg", "-i", "/files/Some_Channel/A_Private_Title.mp4"], 30),
|
|
FileNotFoundError(2, "No such file", "/files/Some_Channel/A_Private_Title.jpg"),
|
|
subprocess.CalledProcessError(1, ["ffmpeg", "-i", "/files/Some_Channel/x.mp4"]),
|
|
],
|
|
ids=lambda e: type(e).__name__,
|
|
)
|
|
def test_an_image_step_failure_is_its_kind_and_nothing_it_quotes(err):
|
|
from app.downloads import edit as editmod
|
|
|
|
code = getattr(err, "returncode", None)
|
|
assert editmod._step_failure(err) == " ".join(
|
|
[type(err).__name__, *([str(code)] if code is not None else [])]
|
|
)
|
|
|
|
|
|
def test_a_thumbnail_that_cannot_be_copied_names_no_file(sink, monkeypatch, tmp_path):
|
|
from app.downloads import edit as editmod
|
|
|
|
src = tmp_path / "thumb.jpg"
|
|
src.write_bytes(b"\xff\xd8\xff rest")
|
|
|
|
def refuse(a, b):
|
|
raise PermissionError(13, "Permission denied", "/files/Some_Channel/A_Private_Title.jpg")
|
|
|
|
monkeypatch.setattr(editmod.shutil, "copyfile", refuse)
|
|
assert editmod.thumbnail_to_jpg(src, tmp_path / "out.jpg") is None
|
|
|
|
(line,) = [o for o in sink.objects() if "thumbnail copy failed" in o["msg"]]
|
|
assert "Some_Channel" not in json.dumps(line)
|
|
|
|
|
|
# --- the child's last stop ------------------------------------------------------------------------
|
|
def test_a_failure_the_child_cannot_record_is_one_scrubbed_line_and_a_non_zero_exit(
|
|
sink, monkeypatch
|
|
):
|
|
"""The pipeline's failure write was refused. Python's own hook would print the chain raw —
|
|
and its context is the download's failure, the URL in it. The worker fails the job on the
|
|
exit code (`child_exit`), so this line does not claim the job failed."""
|
|
from app.downloads import runner
|
|
|
|
url = "https://example.org/v/Our%20Trip%3A%20Day%201.mp4"
|
|
text = f"ERROR: [generic] Our Trip: Day 1: Unable to download webpage ({url})"
|
|
|
|
def download(*a):
|
|
try:
|
|
raise yt_dlp.utils.DownloadError(text)
|
|
except yt_dlp.utils.DownloadError:
|
|
raise ConnectionError("the database went away") # noqa: B904
|
|
|
|
monkeypatch.setattr(pipeline, "_download", download)
|
|
with pytest.raises(SystemExit) as exited:
|
|
runner._run(5, 7, "download", "url", url, {}, 1)
|
|
|
|
assert exited.value.code == 1
|
|
(line,) = [o for o in sink.objects() if o["level"] in ("WARNING", "ERROR")]
|
|
assert line["event"] == "download.child.failed"
|
|
assert "Unable to download webpage" in line["exc"]
|
|
# "Day 1" too: the fallback pattern alone stops at the first `: ` and leaves it.
|
|
assert "Trip" not in json.dumps(line) and "Day 1" not in json.dumps(line)
|
|
|
|
|
|
def test_a_crash_the_database_will_not_record_carries_its_stack_once(monkeypatch, sink):
|
|
claims = iter([42])
|
|
|
|
def claim():
|
|
job = next(claims, None)
|
|
if job is None:
|
|
worker._stop.set()
|
|
return job
|
|
|
|
def crash(job_id):
|
|
raise RuntimeError("the one crash")
|
|
|
|
def db_down(*a, **k):
|
|
raise ConnectionError("the database went away")
|
|
|
|
monkeypatch.setattr(worker, "_claim_job", claim)
|
|
monkeypatch.setattr(worker, "_process_job", crash)
|
|
monkeypatch.setattr(worker, "_fail_running", db_down)
|
|
monkeypatch.setattr(worker.settings, "download_poll_seconds", 0)
|
|
worker._stop.clear()
|
|
try:
|
|
worker._worker_loop(0)
|
|
finally:
|
|
worker._stop.clear()
|
|
|
|
stacks = [o["exc"] for o in sink.objects() if "exc" in o]
|
|
assert sum("the one crash" in s for s in stacks) == 1
|
|
(write,) = sink.events("download.worker.fail_write_failed")
|
|
assert "the database went away" in write["exc"]
|
|
|
|
|
|
def test_a_watchdog_error_on_a_job_it_failed_is_that_jobs_failure_line(
|
|
sink, finalized, monkeypatch
|
|
):
|
|
"""Through the real `_fail_after_kill` → `_fail_running`: the verdict it returns picks the
|
|
event, so a lost `return` would log every such failure as a mere worker error."""
|
|
|
|
class _Alive:
|
|
pid = 4242
|
|
returncode = None
|
|
|
|
def wait(self, timeout=None):
|
|
raise subprocess.TimeoutExpired("child", timeout or 0)
|
|
|
|
def broken(job_id):
|
|
raise KeyError("a watchdog bug")
|
|
|
|
monkeypatch.setattr(worker, "_spawn_download_child", lambda job_id: _Alive())
|
|
monkeypatch.setattr(worker, "_job_liveness", broken)
|
|
monkeypatch.setattr(worker, "_stop_and_free", lambda *a: None)
|
|
|
|
worker._run_supervised(42, 7)
|
|
|
|
(line,) = [o for o in sink.objects() if o["level"] in ("WARNING", "ERROR")]
|
|
assert (line["event"], line["attrs"]["cause"]) == ("download.job.failed", "supervisor_error")
|
|
assert "a watchdog bug" in line["exc"]
|