"""Continuous stderr draining for streaming ffmpeg (_FfmpegStderrTail). ffmpeg is spawned with stderr=PIPE, and stderr used to be read only when something had already gone wrong — so for the life of a stream nobody read that pipe. ffmpeg writes its banner, the input analysis and then a progress line at a steady rate, so a 64 KiB pipe fills eventually, ffmpeg blocks writing to it, frames stop, and the stream's own read timeout fires with nothing in the log explaining that we starved it. How long that takes is unmeasured and evidently long — one H2D upstream ran 21m36s without stalling — so this is a bounded resource being treated as unbounded rather than an observed failure. These tests pin the four properties that matter: the pipe is always drained, the retained tail is bounded, the tail is what the error paths report, and it goes through the same redaction funnel as every other stderr log in this module. """ from __future__ import annotations import asyncio import pytest from backend.app.api.routes import camera pytestmark = pytest.mark.asyncio class _Reader: """Feeds queued chunks, then blocks like a live-but-quiet ffmpeg.""" def __init__(self, chunks: list[bytes], then_block: bool = True) -> None: self._chunks = list(chunks) self._then_block = then_block self.reads = 0 async def read(self, _size: int = -1) -> bytes: self.reads += 1 if self._chunks: return self._chunks.pop(0) if self._then_block: await asyncio.Event().wait() # never returns, never EOFs return b"" class _Proc: def __init__(self, reader, pid: int = 88010) -> None: self.pid = pid self.returncode = None self.stdout = None self.stderr = reader @pytest.fixture(autouse=True) def _no_leaked_tails(): yield # Last-resort teardown only — this fixture is sync, so it cancels without # awaiting. Tests are expected to aclose() their own tails. for tail in list(camera._stderr_tails.values()): if tail._task is not None: tail._task.cancel() camera._stderr_tails.clear() async def _settle() -> None: """Let the pump task run.""" for _ in range(5): await asyncio.sleep(0) async def test_it_keeps_draining_a_stream_that_never_closes_stderr(): """The whole point: the pipe is read continuously, not on demand.""" reader = _Reader([b"first\n", b"second\n"]) tail = camera._FfmpegStderrTail(_Proc(reader)) await _settle() assert reader.reads >= 3, "pump stopped reading instead of following the pipe" assert "second" in (tail.text() or "") await tail.aclose() async def test_the_retained_tail_is_bounded(): """A long-running stream must not turn the pipe into unbounded memory.""" oversized = b"x" * (camera._FFMPEG_STDERR_TAIL_BYTES * 3) tail = camera._FfmpegStderrTail(_Proc(_Reader([oversized]))) await _settle() assert len(tail._buffer) == camera._FFMPEG_STDERR_TAIL_BYTES await tail.aclose() async def test_the_tail_keeps_the_newest_output(): """Recent output is what explains a failure; the banner gets stripped anyway.""" filler = b"stale-line\n" * 4000 tail = camera._FfmpegStderrTail(_Proc(_Reader([filler, b"Connection timed out\n"]))) await _settle() text = tail.text() or "" assert "Connection timed out" in text assert len(tail._buffer) <= camera._FFMPEG_STDERR_TAIL_BYTES await tail.aclose() async def test_read_ffmpeg_stderr_defers_to_the_collector(): """Two readers on one StreamReader raise, so the on-demand read must not touch a pipe the collector owns.""" reader = _Reader([b"Server returned 401 Unauthorized\n"]) process = _Proc(reader) tail = camera._FfmpegStderrTail(process) await _settle() reads_before = reader.reads result = await camera._read_ffmpeg_stderr(process) assert "401 Unauthorized" in (result or "") assert reader.reads == reads_before, "on-demand read raced the collector" await tail.aclose() async def test_read_ffmpeg_stderr_still_reads_the_pipe_without_a_collector(): """An immediately-failed ffmpeg has no collector; that path must still work.""" process = _Proc(_Reader([b"Server returned 404 Not Found\n"], then_block=False)) result = await camera._read_ffmpeg_stderr(process) assert "404 Not Found" in (result or "") async def test_the_tail_redacts_the_access_code(): """ffmpeg echoes its input URL, which carries the printer's access code. This is a new stderr-to-log path, so it gets the same guarantee as the rest: everything goes through _summarize_ffmpeg_stderr. """ secret = "12345678" leaky = f"[rtsp @ 0x55] Failed to resolve rtsp://bblp:{secret}@127.0.0.1:8554/streaming/live/1\n" tail = camera._FfmpegStderrTail(_Proc(_Reader([leaky.encode()]))) await _settle() text = tail.text() or "" assert secret not in text, "access code leaked into a log line" # Assert the line SURVIVED with the credential masked, not that it was # dropped — otherwise this passes whenever the summariser happens to filter # the line out, and proves nothing about redaction. assert "Failed to resolve" in text, "line was filtered, so redaction is untested" assert "[REDACTED]" in text await tail.aclose() async def test_close_releases_ownership_and_is_idempotent(): process = _Proc(_Reader([b"line\n"])) tail = camera._FfmpegStderrTail(process) await _settle() assert camera._stderr_tails.get(process.pid) is tail await tail.aclose() await tail.aclose() # must not raise assert process.pid not in camera._stderr_tails async def test_a_process_without_stderr_is_handled(): """Fakes and some spawn paths pass stderr=None; must not register or crash.""" process = _Proc(None, pid=88099) tail = camera._FfmpegStderrTail(process) assert tail.text() is None assert process.pid not in camera._stderr_tails await tail.aclose() async def test_the_stream_generator_owns_then_releases_the_collector(monkeypatch): """Lifecycle inside the real generator: registered while streaming, gone after. The other generator tests use fakes with stderr=None, so they never build a collector at all — this is the one that would catch a missing close() or a reader race between the collector and teardown. """ printer_id = 8842 stream_id = f"{printer_id}-fanout-stderrtail" class _FrameThenBlock: def __init__(self) -> None: self._sent = False async def read(self, _size: int = -1) -> bytes: if self._sent: await asyncio.Event().wait() # stay alive, don't trigger reconnect self._sent = True return b"\xff\xd8frame\xff\xd9" class _Proc2: def __init__(self) -> None: self.pid = 88042 self.returncode = None self.stdout = _FrameThenBlock() self.stderr = _Reader([b"Stream #0:0: Video: h264\n"]) def terminate(self) -> None: self.returncode = 0 def kill(self) -> None: self.returncode = -9 async def wait(self) -> int: if self.returncode is None: self.returncode = 0 return self.returncode process = _Proc2() class _FakeServer: def close(self) -> None: pass async def wait_closed(self) -> None: pass async def _fake_exec(*_a, **_kw): return process async def _fake_proxy(_ip, _port): return 48777, _FakeServer() monkeypatch.setattr(camera, "get_ffmpeg_path", lambda: "/fake/ffmpeg") monkeypatch.setattr(camera, "create_tls_proxy", _fake_proxy) monkeypatch.setattr(camera.asyncio, "create_subprocess_exec", _fake_exec) stream = camera.generate_rtsp_mjpeg_stream( ip_address="192.0.2.44", access_code="c", model="P2S", fps=15, stream_id=stream_id, disconnect_event=asyncio.Event(), printer_id=printer_id, ) try: chunk = await asyncio.wait_for(anext(stream), timeout=5.0) assert b"frame" in chunk assert process.pid in camera._stderr_tails, "generator did not take stderr ownership" await asyncio.wait_for(stream.aclose(), timeout=5.0) assert process.pid not in camera._stderr_tails, "collector outlived its stream" finally: camera._active_streams.pop(stream_id, None) camera._disconnect_events.pop(stream_id, None) camera._stream_last_frame_times.pop(stream_id, None) camera._last_frames.pop(printer_id, None) camera._last_frame_times.pop(printer_id, None) camera._stream_start_times.pop(printer_id, None) camera._spawned_ffmpeg_pids.pop(process.pid, None) async def test_terminate_skips_stderr_while_a_collector_owns_it(): """_terminate_ffmpeg must not add a second reader to an owned pipe.""" reader = _Reader([b"tearing down\n"]) class _Killable(_Proc): def terminate(self): self.returncode = 0 def kill(self): self.returncode = -9 async def wait(self): if self.returncode is None: self.returncode = 0 return self.returncode process = _Killable(reader, pid=88020) tail = camera._FfmpegStderrTail(process) await _settle() reads_before = reader.reads await asyncio.wait_for(camera._terminate_ffmpeg(process, "88020-fanout-abcd"), timeout=2.0) # The collector, not _terminate_ffmpeg, is the only reader that advanced. assert reader.reads >= reads_before assert tail.text() is not None await tail.aclose()