"""E22.5 S4 — yt-dlp explains itself to our log, not to a stdout nobody reads. `quiet: True` was doing more than quieting: with no `logger`, yt-dlp's whole account of a download went to the child process's stdout, so "Requested format is not available", a signature-extraction failure or a POT refusal reached us only as whatever text happened to end up in the raised exception. A failed download's real reason was routinely one line the app never kept. Measured against yt-dlp 2026.07.04 (the pinned version): with `logger` set, yt-dlp calls `debug` for its extraction and progress lines and `warning`/`error` for the rest, and neither `quiet` nor `no_warnings` gates those calls. The cases below pin what we do with them. """ import json import logging import pytest from app import logs from app.downloads import formats URL = "https://www.youtube.com/watch?v=dQw4w9WgXcQ&list=PLsomething" class _Sink(logging.Handler): """The stdout handler's own filter and formatter, so a case sees the bytes a deployment writes.""" 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] @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) @pytest.mark.parametrize( ("method", "level"), [ # Several lines per second during a download, and TRACE is the level a user turns on for # their own session — so the noisy half is exactly the half the consent gates. ("debug", "TRACE"), ("info", "DEBUG"), ("warning", "WARNING"), # yt-dlp calls `error` from `trouble()` just before it raises, on every failed ATTEMPT — # the pipeline's fallback client set then often succeeds. The job's verdict is the # pipeline's `download.job.failed`, not this. ("error", "WARNING"), ], ) def test_each_ytdlp_level_lands_where_it_belongs(sink, method, level): getattr(formats._YDL_LOG, method)("something happened") (line,) = sink.objects() assert line["level"] == level assert line["event"] == "download.ytdlp" assert "something happened" in line["msg"] def test_the_source_url_does_not_travel_with_the_explanation(sink): formats._YDL_LOG.error(f"ERROR: [youtube] {URL}: Requested format is not available") (line,) = sink.objects() # The host survives — it is what says which extractor is failing. The rest is a row of this # person's download history, and the job id already names the job on every one of these lines. assert "www.youtube.com/…" in line["msg"] assert "dQw4w9WgXcQ" not in json.dumps(line) assert "list=PLsomething" not in json.dumps(line) assert "Requested format is not available" in line["msg"] @pytest.mark.parametrize( ("line", "kept", "gone"), [ # The shape nearly every yt-dlp warning and error has: no URL, the id as a prefix. ( "ERROR: [youtube] dQw4w9WgXcQ: Video unavailable", "ERROR: [youtube] …: Video unavailable", "dQw4w9WgXcQ", ), # The generic extractor's "id" is the URL's own basename. ( "WARNING: [generic] my-private-holiday-clip: Falling back on generic information", "WARNING: [generic] …: Falling back on generic information", "my-private-holiday-clip", ), # ...percent-decoded, so it can have spaces: `https://x/My%20Holiday%20in%20Crete.mp4`. ( "ERROR: [generic] My Holiday in Crete: Unable to download webpage: HTTP Error 404", "ERROR: [generic] …: Unable to download webpage: HTTP Error 404", "Holiday", ), ( "[generic] My Holiday in Crete: Downloading webpage", "[generic] …: Downloading webpage", "Holiday", ), # A playlist URL is extracted too (`playlist_items: "1"`), and its title is printed. ( "[youtube:tab] Playlist Our Wedding Videos: Downloading 1 items of 12", "[youtube:tab] Playlist …: Downloading 1 items of 12", "Wedding", ), ( "[download] Finished downloading playlist: Our Wedding Videos", "[download] Finished downloading playlist: …", "Wedding", ), ( "[download] Downloading playlist: Our Wedding Videos", "[download] Downloading playlist: …", "Wedding", ), # An id quoted mid-sentence. ( "hu subtitles not available for dQw4w9WgXcQ", "hu subtitles not available for …", "dQw4w9WgXcQ", ), # `watch?v=…&list=…` under `noplaylist` (`_yes_playlist`), and the multifeed shape. ( "[youtube:tab] Downloading just the video dQw4w9WgXcQ because of --no-playlist", "[youtube:tab] Downloading just the video … because of --no-playlist", "dQw4w9WgXcQ", ), ( "[youtube] Downloading just video dQw4w9WgXcQ because of --no-playlist", "[youtube] Downloading just video … because of --no-playlist", "dQw4w9WgXcQ", ), # `Destination` is yt-dlp's word, not an id; the path after it is built from the title. ( "[download] Destination: /downloads/Some_Channel/Season_2024/A_Private_Title_a7.mp4", "[download] Destination: …", "A_Private_Title", ), ], ) def test_what_names_the_download_comes_out_of_real_ytdlp_shapes(sink, line, kept, gone): formats._YDL_LOG.warning(line) (obj,) = sink.objects() assert obj["msg"] == kept assert gone not in json.dumps(obj) @pytest.mark.parametrize( ("url", "name"), [ ("https://example.org/v/Wedding%20%5B2024%5D.mp4", "Wedding [2024]"), ("https://example.org/v/Our%20Trip%3A%20Day%201.mp4", "Our Trip: Day 1"), ], ) def test_the_jobs_own_generic_name_is_taken_out_whatever_it_holds(url, name): """Brackets and `: ` bound no pattern; the job knows its URL, so the exact name goes.""" line = f"ERROR: [generic] {name}: Unable to download webpage: HTTP Error 404: Not Found" assert formats.generic_id(url) == name with formats.hiding(url): assert formats.log_safe(line) == ( "ERROR: [generic] …: Unable to download webpage: HTTP Error 404: Not Found" ) def test_the_generic_name_is_yt_dlps_own(): from yt_dlp.extractor.common import InfoExtractor for url in ("https://x.org/a/My%20Holiday.mp4", "https://x.org/a/clip/", "https://x.org/b"): assert formats.generic_id(url) == InfoExtractor._generic_id(url) def test_a_download_error_is_scrubbed_the_same_way_wherever_it_is_logged(): """yt-dlp raises the text it has just logged, and the pipeline logs the exception again one line later (`download.job.failed`). The same scrub, or that line puts the URL back.""" text = "ERROR: Unsupported URL: https://example.org/private/path?id=123" assert formats.log_safe(text) == "ERROR: Unsupported URL: example.org/…" @pytest.mark.parametrize( "line", [ "nsig extraction failed: Some formats may be missing", "[VideoConvertor] Converting video from webm to mp4; Destination: x", "[download] 53.2% of ~ 12.00MiB at 1.20MiB/s ETA 00:05", "[download] 100% of 12.00MiB in 00:00:03 at 3.90MiB/s", ], ) def test_a_line_that_names_nothing_keeps_its_words(line): """The scrub takes out what names a download, and nothing an operator reads the line for.""" assert formats.log_safe(line) == line def test_a_line_with_no_url_is_left_alone(sink): formats._YDL_LOG.warning("nsig extraction failed: Some formats may be missing") (line,) = sink.objects() assert "nsig extraction failed: Some formats may be missing" in line["msg"] def test_the_download_options_carry_the_logger(): opts = formats.build_ydl_opts({}, "%(id)s.%(ext)s", lambda d: None) assert opts["logger"] is formats._YDL_LOG def test_a_source_adapter_cannot_silence_it(): """`extra` is the adapter's own options (YouTube's player-client/POT strategy), merged wholesale. The logger is set after that merge for the same reason `cookiefile` and the size cap are: an adapter must not be able to take away the one account of what went wrong that a failed job leaves behind.""" opts = formats.build_ydl_opts( {}, "%(id)s.%(ext)s", lambda d: None, extra={"logger": None, "quiet": False} ) assert opts["logger"] is formats._YDL_LOG