bambuddy/backend/tests/unit/test_camera_usb_stream_cleanup.py

"""External/USB camera ffmpeg-leak cleanup (#2675, reporter @bitbarista).

An external USB (V4L2) camera's ffmpeg used to be reachable only from its own
stream generator's ``finally`` — which an abrupt client disconnect can skip
(same cancellation-timing class as #776). Because external streams never
registered into ``_active_streams`` / ``_disconnect_events`` / the spawned-PID
map, both ``/camera/stop`` and ``cleanup_orphaned_streams`` were structurally
blind to the leak, leaving ``/dev/videoN`` locked. The fix registers the external
ffmpeg into the same registries the RTSP path uses.
"""

from __future__ import annotations

import asyncio
import time
from contextlib import suppress
from unittest.mock import mock_open, patch

import pytest

from backend.app.api.routes import camera
from backend.app.services import external_camera


async def _instant_sleep(*_args, **_kwargs) -> None:
    """Drop-in for asyncio.sleep that returns immediately (no self-recursion)."""
    return None


class _CleanProc:
    """ffmpeg that terminates cleanly when asked."""

    def __init__(self, pid: int) -> None:
        self.pid = pid
        self.returncode = None
        # Real Process objects always expose these (None when not piped), and
        # _terminate_ffmpeg drains them so a full pipe can't wedge the exit.
        self.stdout = None
        self.stderr = None

    def terminate(self) -> None:
        self.returncode = 0

    def kill(self) -> None:
        self.returncode = -9

    async def wait(self) -> int:
        return self.returncode if self.returncode is not None else 0


class _ImmediateEOFReader:
    async def read(self, _size: int = -1) -> bytes:
        return b""


class _UsbProc:
    """ffmpeg for a USB stream: yields no frames, exits at first read."""

    def __init__(self, pid: int = 52001) -> None:
        self.pid = pid
        self.returncode = None
        self.stdout = _ImmediateEOFReader()
        self.stderr = _ImmediateEOFReader()

    def terminate(self) -> None:
        self.returncode = 0

    def kill(self) -> None:
        self.returncode = -9

    async def wait(self) -> int:
        return 0


# ---------------------------------------------------------------------------
# 1. The stream generator hands its ffmpeg process to the on_process callback
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_stream_usb_registers_process_via_on_process(monkeypatch):
    """``_stream_usb`` must call ``on_process`` with the spawned ffmpeg so the
    route can register it — this is the linchpin of the whole fix."""

    class _FakePath:
        def __init__(self, _p: str) -> None:
            pass

        def exists(self) -> bool:
            return True

    proc = _UsbProc()

    async def fake_create_subprocess_exec(*_args, **_kwargs):
        return proc

    monkeypatch.setattr(external_camera, "get_ffmpeg_path", lambda: "/fake/ffmpeg")
    monkeypatch.setattr(external_camera, "Path", _FakePath)
    monkeypatch.setattr(external_camera.asyncio, "create_subprocess_exec", fake_create_subprocess_exec)
    monkeypatch.setattr(external_camera.asyncio, "sleep", _instant_sleep)

    captured: list[object] = []
    stream = external_camera._stream_usb("/dev/video0", 10, on_process=captured.append)
    try:
        async for _frame in stream:
            pass
    finally:
        with suppress(Exception):
            await stream.aclose()

    assert captured == [proc], "the spawned ffmpeg process must be handed to on_process"


# ---------------------------------------------------------------------------
# 2. /camera/stop now finds and kills a registered external USB process
#    (the reported {"stopped": 0} → {"stopped": 1})
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_stop_endpoint_terminates_registered_external_process(monkeypatch):
    monkeypatch.setattr(camera, "_FFMPEG_KILL_TIMEOUT", 0.05)
    monkeypatch.setattr(camera, "get_subscriber_count", lambda _key: 0)

    async def fake_shutdown(_key):
        return False

    monkeypatch.setattr(camera, "shutdown_broadcaster", fake_shutdown)

    printer_id = 7
    sid = f"{printer_id}-ext-abc12345"
    proc = _CleanProc(pid=52010)
    event = asyncio.Event()
    camera._active_streams[sid] = proc
    camera._disconnect_events[sid] = event
    camera._spawned_ffmpeg_pids[proc.pid] = time.time()
    camera._stream_last_frame_times[sid] = time.time()

    try:
        result = await camera.stop_camera_stream(printer_id, _=None)
        assert result["stopped"] == 1
        assert proc.returncode is not None, "the external ffmpeg must be terminated"
        assert event.is_set(), "the stream's stop event must be signalled"
        # Registry fully cleaned so it can't be double-reaped.
        assert sid not in camera._active_streams
        assert sid not in camera._disconnect_events
        assert proc.pid not in camera._spawned_ffmpeg_pids
    finally:
        camera._active_streams.pop(sid, None)
        camera._disconnect_events.pop(sid, None)
        camera._spawned_ffmpeg_pids.pop(proc.pid, None)
        camera._stream_last_frame_times.pop(sid, None)


# ---------------------------------------------------------------------------
# 3. The orphan janitor reaps a stale registered external USB stream
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_cleanup_janitor_reaps_stale_external_usb_stream(monkeypatch):
    monkeypatch.setattr(camera, "_FFMPEG_KILL_TIMEOUT", 0.05)
    monkeypatch.setattr(camera, "_scan_bambu_ffmpeg_pids", lambda: [])

    import os

    proc = _CleanProc(pid=os.getpid())  # real pid so layer-2 existence check keeps it
    sid = "7-ext-deadbeef"
    now = time.time()
    camera._active_streams[sid] = proc
    camera._spawned_ffmpeg_pids[proc.pid] = now - 120  # spawned long ago
    camera._stream_last_frame_times[sid] = now - 60  # stale: no frames >30s
    camera._disconnect_events[sid] = asyncio.Event()

    try:
        await asyncio.wait_for(camera.cleanup_orphaned_streams(), timeout=2.0)
        assert proc.returncode is not None, "stale external ffmpeg must be killed"
        assert sid not in camera._active_streams
    finally:
        camera._active_streams.pop(sid, None)
        camera._spawned_ffmpeg_pids.pop(proc.pid, None)
        camera._stream_last_frame_times.pop(sid, None)
        camera._disconnect_events.pop(sid, None)


# ---------------------------------------------------------------------------
# 4. The /proc "nuclear net" now matches USB (v4l2) ffmpeg from prior sessions
# ---------------------------------------------------------------------------


def test_scan_matches_v4l2_ffmpeg(monkeypatch):
    cmdline = b"ffmpeg\x00-f\x00v4l2\x00-i\x00/dev/video0\x00-f\x00mjpeg\x00-\x00"
    monkeypatch.setattr("os.listdir", lambda _p: ["52020"])
    monkeypatch.setattr(camera, "_read_ppid", lambda _pid: 1)
    with patch("builtins.open", mock_open(read_data=cmdline)):
        assert 52020 in camera._scan_bambu_ffmpeg_pids()


def test_scan_ignores_unrelated_ffmpeg(monkeypatch):
    # A transcode of a local file is not ours — must not be reaped.
    cmdline = b"ffmpeg\x00-i\x00/home/user/movie.mp4\x00out.mkv\x00"
    monkeypatch.setattr("os.listdir", lambda _p: ["52021"])
    with patch("builtins.open", mock_open(read_data=cmdline)):
        assert camera._scan_bambu_ffmpeg_pids() == []


# ---------------------------------------------------------------------------
# 5. Only our own USB ffmpeg: another program's camera process is left alone
# ---------------------------------------------------------------------------

OWN_PID = 4242
ONE_SHOT = [b"/usr/bin/ffmpeg", b"-f", b"v4l2", b"-i", b"/dev/video0", b"-frames:v", b"1", b"-f", b"image2pipe", b"-"]
STREAM = [b"ffmpeg", b"-f", b"v4l2", b"-framerate", b"10", b"-i", b"/dev/video2", b"-f", b"mjpeg", b"-"]


def test_our_one_shot_and_stream_commands_are_ours():
    assert camera._is_bambuddy_ffmpeg(ONE_SHOT, OWN_PID, OWN_PID)
    assert camera._is_bambuddy_ffmpeg(STREAM, OWN_PID, OWN_PID)


def test_an_orphan_from_an_earlier_run_is_ours():
    """Reparented to init once the Bambuddy that spawned it died."""
    assert camera._is_bambuddy_ffmpeg(STREAM, 1, OWN_PID)


def test_another_programs_usb_ffmpeg_is_not_ours():
    """motion, an OctoPrint webcam or a recording on a bare-metal install."""
    assert not camera._is_bambuddy_ffmpeg(STREAM, 777, OWN_PID)


def test_a_different_v4l2_command_shape_is_not_ours():
    recording = [b"ffmpeg", b"-f", b"v4l2", b"-input_format", b"h264", b"-i", b"/dev/video0", b"/srv/cam.mkv"]
    assert not camera._is_bambuddy_ffmpeg(recording, OWN_PID, OWN_PID)


def test_bambu_rtsp_is_ours_whatever_the_parent():
    rtsp = [b"ffmpeg", b"-rtsp_transport", b"tcp", b"-i", b"rtsps://bblp:code@10.0.0.5:322/streaming/live/1", b"-"]
    assert camera._is_bambuddy_ffmpeg(rtsp, 777, OWN_PID)


def test_unknown_parent_is_not_ours_for_usb():
    assert not camera._is_bambuddy_ffmpeg(STREAM, None, OWN_PID)