"""ffmpeg's diagnosis survives the log line, and its banner does not (#2968). ffmpeg prints ~20 lines of version and build banner first and its actual error last, so the ``stderr[:200]`` most call sites used kept the banner and dropped the error. The reporter's H2D logged twelve capture failures that way; every one of them was the same 200 characters of ``--prefix=/usr --extra-version=`` and none of them said why the capture failed. #925 already solved this for the camera streaming endpoint. These tests cover the shared module the other seven call sites now go through, and the two things that were only ever true of the private copy: it takes bytes, and it masks credentials for callers that never did. """ import inspect import pytest from backend.app.utils.ffmpeg_output import NO_FFMPEG_OUTPUT, summarize_ffmpeg_stderr # Verbatim from the reporter's log, trimmed to the width the old truncation # allowed through. The point of the fixture is that 200 characters of it carry # no information at all. _REAL_BANNER = """ffmpeg version 7.1.4-0+deb13u1 Copyright (c) 2000-2026 the FFmpeg developers built with gcc 14 (Debian 14.2.0-19) configuration: --prefix=/usr --extra-version=0+deb13u1 --toolchain=hardened --enable-gpl libavutil 59. 39.100 / 59. 39.100 libavcodec 61. 19.101 / 61. 19.101 libavformat 61. 7.100 / 61. 7.100 libavdevice 61. 3.100 / 61. 3.100 libavfilter 10. 4.100 / 10. 4.100 libswscale 8. 3.100 / 8. 3.100 libswresample 5. 3.100 / 5. 3.100 libpostproc 58. 3.100 / 58. 3.100 """ class TestTheDiagnosisSurvives: def test_the_error_is_kept_and_the_banner_is_not(self): """The whole point: the last line, not the first 200 characters.""" stderr = _REAL_BANNER + "[rtsp @ 0x5f] method DESCRIBE failed: 401 Unauthorized\n" result = summarize_ffmpeg_stderr(stderr) assert "method DESCRIBE failed: 401 Unauthorized" in result assert "ffmpeg version" not in result assert "--prefix=/usr" not in result def test_the_old_truncation_would_have_kept_none_of_it(self): """Guards the claim the fix rests on rather than asserting it in prose: 200 characters from the front of a real failure is banner only.""" stderr = _REAL_BANNER + "[rtsp @ 0x5f] method DESCRIBE failed: 401 Unauthorized\n" assert "DESCRIBE" not in stderr[:200] def test_input_analysis_is_kept(self): """Indented, but not banner. ``Duration:`` and ``Stream #0:0`` explain the error above them and are the reason the match is on exact prefixes rather than on leading whitespace.""" stderr = _REAL_BANNER + ( "Input #0, rtsp, from 'rtsp://192.0.2.1:322/streaming/live/1':\n" " Duration: N/A, start: 0.000000, bitrate: N/A\n" " Stream #0:0: Video: h264, yuv420p, 1920x1080\n" "Output file is empty, nothing was encoded\n" ) result = summarize_ffmpeg_stderr(stderr) assert "Duration: N/A" in result assert "Stream #0:0: Video: h264" in result assert "Output file is empty" in result def test_only_the_last_lines_are_kept(self): """A chatty decoder must not rotate the log file on one failure.""" stderr = _REAL_BANNER + "\n".join(f"error line {i}" for i in range(40)) lines = summarize_ffmpeg_stderr(stderr).splitlines() assert len(lines) == 10 assert lines[-1] == "error line 39" def test_a_banner_only_failure_says_so(self): """Empty, so the caller substitutes a phrase. ``failed: `` with nothing after it reads like a truncation bug rather than a silent printer.""" assert summarize_ffmpeg_stderr(_REAL_BANNER) == "" assert (summarize_ffmpeg_stderr(_REAL_BANNER) or NO_FFMPEG_OUTPUT) == NO_FFMPEG_OUTPUT class TestWhatTheCallSitesUsedToGetWrong: def test_bytes_are_accepted(self): """Every call site held bytes and decoded them itself.""" assert "Connection refused" in summarize_ffmpeg_stderr(b"rtsp://192.0.2.1: Connection refused\n") def test_undecodable_bytes_do_not_raise(self): """ffmpeg copies stream fragments into its messages, so a bare ``.decode()`` could raise UnicodeDecodeError while reporting an unrelated failure -- losing the diagnosis to a second exception.""" result = summarize_ffmpeg_stderr(b"\xff\xfe broken input\nInvalid data found\n") assert "Invalid data found" in result def test_the_access_code_is_masked(self): """ffmpeg echoes its input URL back, and four of the call sites logged it unmasked. The mask is part of the summary so it cannot be skipped.""" stderr = b"Error opening input file rtsp://bblp:12345678@192.0.2.1:322/streaming/live/1.\n" result = summarize_ffmpeg_stderr(stderr) assert "12345678" not in result assert "[REDACTED]" in result # Host and user survive, or the line stops being useful for diagnosis. assert "192.0.2.1:322" in result assert "bblp" in result def test_a_credential_masked_before_the_cut_not_after(self): """Truncating first would leave a URL with no ``@`` for the pattern to anchor on, and the secret in the log.""" stderr = "\n".join(f"noise {i}" for i in range(30)) stderr += "\nOpening rtsp://user:hunter2@192.0.2.1:322/live and 40 more characters of tail\n" result = summarize_ffmpeg_stderr(stderr) assert "hunter2" not in result @pytest.mark.parametrize("empty", ["", None, b""]) def test_nothing_in_nothing_out(self, empty): assert summarize_ffmpeg_stderr(empty) == "" def test_a_single_enormous_line_is_bounded(self): """Ten lines only bounds the record if the lines are sane, and ffmpeg quotes back what the peer sent it. The tail is what is kept.""" stderr = _REAL_BANNER + "x" * 50_000 + " Connection refused\n" result = summarize_ffmpeg_stderr(stderr) assert len(result) < 2_100 assert result.endswith("Connection refused") assert result.startswith("...") def test_an_ordinary_diagnosis_is_never_trimmed(self): """The ceiling must not be reachable by real ffmpeg output.""" stderr = _REAL_BANNER + "\n".join(f"[rtsp @ 0x5f] error line {i}" for i in range(10)) assert not summarize_ffmpeg_stderr(stderr).startswith("...") class TestEveryCallSiteGoesThroughIt: """The defect was seven copies of the same truncation, not one bad line. Asserted against the source because the alternative -- driving all seven subprocesses -- tests ffmpeg, and because the failure mode being guarded is somebody adding an eighth. """ @pytest.mark.parametrize( "module_path", [ "backend.app.services.camera", "backend.app.services.external_camera", "backend.app.services.layer_timelapse", "backend.app.services.timelapse_processor", "backend.app.services.archive", "backend.app.api.routes.camera", ], ) def test_no_module_truncates_stderr_by_hand(self, module_path): import importlib source = inspect.getsource(importlib.import_module(module_path)) for lineno, raw in enumerate(source.splitlines(), 1): # Comments discuss the defect by name -- this file's own fix notes # do -- so only what executes is checked. line = raw.split("#", 1)[0] if "stderr" not in line: continue assert "stderr.decode()[:" not in line, f"{module_path}:{lineno} truncates stderr from the front" assert "stderr_text[:" not in line, f"{module_path}:{lineno} truncates stderr from the front" assert 'stderr.decode(errors="replace")[:' not in line, ( f"{module_path}:{lineno} truncates stderr from the front" ) # A bare decode is the other half of the defect: it can raise # UnicodeDecodeError while reporting an unrelated failure, and it # leaves the input URL's credentials unmasked. assert "stderr.decode()" not in line, f"{module_path}:{lineno} decodes stderr by hand"