bambuddy/backend/tests/integration/test_camera_api.py

"""Integration tests for Camera API endpoints.

Tests the full request/response cycle for /api/v1/printers/{id}/camera/ endpoints.
"""

from unittest.mock import AsyncMock, MagicMock, patch

import pytest
from httpx import AsyncClient


class TestCameraAPI:
    """Integration tests for /api/v1/printers/{id}/camera/ endpoints."""

    # ========================================================================
    # Camera Stop Endpoint
    # ========================================================================

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_stop_camera_stream_get(self, async_client: AsyncClient, printer_factory):
        """Verify camera stop endpoint works with GET method."""
        printer = await printer_factory()

        response = await async_client.get(f"/api/v1/printers/{printer.id}/camera/stop")

        assert response.status_code == 200
        result = response.json()
        assert "stopped" in result
        assert isinstance(result["stopped"], int)

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_stop_camera_stream_post(self, async_client: AsyncClient, printer_factory):
        """Verify camera stop endpoint works with POST method (sendBeacon compatibility)."""
        printer = await printer_factory()

        response = await async_client.post(f"/api/v1/printers/{printer.id}/camera/stop")

        assert response.status_code == 200
        result = response.json()
        assert "stopped" in result
        assert isinstance(result["stopped"], int)

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_stop_camera_stream_no_active_streams(self, async_client: AsyncClient, printer_factory):
        """Verify stop returns 0 when no active streams exist."""
        printer = await printer_factory()

        response = await async_client.post(f"/api/v1/printers/{printer.id}/camera/stop")

        assert response.status_code == 200
        assert response.json()["stopped"] == 0

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_stop_camera_stream_with_active_stream(self, async_client: AsyncClient, printer_factory):
        """Verify stop terminates active streams for the printer."""
        printer = await printer_factory()

        # Mock an active stream — wait() must be AsyncMock since it's awaited
        mock_process = MagicMock()
        mock_process.returncode = None
        mock_process.pid = 99999
        mock_process.terminate = MagicMock()
        mock_process.wait = AsyncMock()

        with patch("backend.app.api.routes.camera._active_streams", {f"{printer.id}-abc123": mock_process}):
            response = await async_client.post(f"/api/v1/printers/{printer.id}/camera/stop")

        assert response.status_code == 200
        assert response.json()["stopped"] == 1
        mock_process.terminate.assert_called_once()

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_stop_camera_stream_only_stops_matching_printer(self, async_client: AsyncClient, printer_factory):
        """Verify stop only terminates streams for the specified printer."""
        printer1 = await printer_factory(name="Printer 1")
        printer2 = await printer_factory(name="Printer 2")

        # Mock active streams for both printers — wait() must be AsyncMock since it's awaited
        mock_process1 = MagicMock()
        mock_process1.returncode = None
        mock_process1.pid = 99998
        mock_process1.terminate = MagicMock()
        mock_process1.wait = AsyncMock()

        mock_process2 = MagicMock()
        mock_process2.returncode = None
        mock_process2.pid = 99997
        mock_process2.terminate = MagicMock()
        mock_process2.wait = AsyncMock()

        active_streams = {
            f"{printer1.id}-abc123": mock_process1,
            f"{printer2.id}-def456": mock_process2,
        }

        with patch("backend.app.api.routes.camera._active_streams", active_streams):
            response = await async_client.post(f"/api/v1/printers/{printer1.id}/camera/stop")

        assert response.status_code == 200
        assert response.json()["stopped"] == 1
        mock_process1.terminate.assert_called_once()
        mock_process2.terminate.assert_not_called()

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_stop_camera_stream_handles_fanout_stream_id(self, async_client: AsyncClient, printer_factory):
        """Stop must terminate streams keyed with the deterministic
        ``{printer_id}-fanout`` id used by the fan-out broadcaster (#1089).
        Regression guard against the prefix-match drifting away from the
        broadcaster's stream-id convention.
        """
        printer = await printer_factory()
        mock_process = MagicMock()
        mock_process.returncode = None
        mock_process.pid = 99996
        mock_process.terminate = MagicMock()
        mock_process.wait = AsyncMock()

        with patch(
            "backend.app.api.routes.camera._active_streams",
            {f"{printer.id}-fanout": mock_process},
        ):
            response = await async_client.post(f"/api/v1/printers/{printer.id}/camera/stop")

        assert response.status_code == 200
        assert response.json()["stopped"] == 1
        mock_process.terminate.assert_called_once()

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_stop_camera_stream_invokes_broadcaster_shutdown(self, async_client: AsyncClient, printer_factory):
        """Stop must call ``shutdown_broadcaster`` so subscribers wake up via
        the upstream-gone sentinel rather than stalling on the queue (#1089)."""
        printer = await printer_factory()

        with patch(
            "backend.app.api.routes.camera.shutdown_broadcaster",
            AsyncMock(return_value=False),
        ) as mock_shutdown:
            response = await async_client.post(f"/api/v1/printers/{printer.id}/camera/stop")

        assert response.status_code == 200
        mock_shutdown.assert_awaited_once_with(f"printer-{printer.id}")

    # ========================================================================
    # Camera Test Endpoint
    # ========================================================================

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_camera_test_printer_not_found(self, async_client: AsyncClient):
        """Verify 404 when testing camera for non-existent printer."""
        response = await async_client.get("/api/v1/printers/99999/camera/test")

        assert response.status_code == 404
        assert "not found" in response.json()["detail"].lower()

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_camera_test_success(self, async_client: AsyncClient, printer_factory):
        """Verify camera test returns success when camera is accessible."""
        printer = await printer_factory()

        with patch("backend.app.api.routes.camera.test_camera_connection", new_callable=AsyncMock) as mock_test:
            mock_test.return_value = {"success": True, "message": "Camera connected"}

            response = await async_client.get(f"/api/v1/printers/{printer.id}/camera/test")

        assert response.status_code == 200
        result = response.json()
        assert result["success"] is True

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_camera_test_failure(self, async_client: AsyncClient, printer_factory):
        """Verify camera test returns failure when camera is not accessible."""
        printer = await printer_factory()

        with patch("backend.app.api.routes.camera.test_camera_connection", new_callable=AsyncMock) as mock_test:
            mock_test.return_value = {"success": False, "message": "Connection timeout"}

            response = await async_client.get(f"/api/v1/printers/{printer.id}/camera/test")

        assert response.status_code == 200
        result = response.json()
        assert result["success"] is False

    # ========================================================================
    # Camera Snapshot Endpoint
    # ========================================================================

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_camera_snapshot_printer_not_found(self, async_client: AsyncClient):
        """Verify 404 when capturing snapshot for non-existent printer."""
        response = await async_client.get("/api/v1/printers/99999/camera/snapshot")

        assert response.status_code == 404

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_camera_snapshot_success(self, async_client: AsyncClient, printer_factory):
        """Verify snapshot returns JPEG image when successful."""
        printer = await printer_factory()

        # Create a fake JPEG (starts with FFD8)
        fake_jpeg = b"\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x00\x00\x01\x00\x01\x00\x00"

        with patch("backend.app.api.routes.camera.capture_camera_frame", new_callable=AsyncMock) as mock_capture:
            mock_capture.return_value = True

            # Mock the file read
            with patch("builtins.open", create=True) as mock_open:
                mock_open.return_value.__enter__.return_value.read.return_value = fake_jpeg

                with patch("pathlib.Path.exists", return_value=True), patch("pathlib.Path.unlink"):
                    _response = await async_client.get(f"/api/v1/printers/{printer.id}/camera/snapshot")

        # Note: The actual test might fail due to file operations, but this tests the endpoint structure
        # In production tests, we'd mock more comprehensively

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_camera_snapshot_failure(self, async_client: AsyncClient, printer_factory):
        """Verify 503 when camera capture fails."""
        printer = await printer_factory()

        with patch("backend.app.api.routes.camera.capture_camera_frame", new_callable=AsyncMock) as mock_capture:
            mock_capture.return_value = False

            with patch("pathlib.Path.exists", return_value=False), patch("pathlib.Path.unlink"):
                response = await async_client.get(f"/api/v1/printers/{printer.id}/camera/snapshot")

        assert response.status_code == 503
        assert "Failed to capture" in response.json()["detail"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_camera_snapshot_reuses_buffered_frame_when_stream_active(
        self, async_client: AsyncClient, printer_factory
    ):
        """#1271: /camera/snapshot must reuse the broadcaster's buffered frame
        when a live stream is running, instead of opening a second concurrent
        RTSP socket. On printers with strict single-connection enforcement (e.g.
        X2D firmware 01.01.00.00) opening a second socket kicks the live stream.
        """
        printer = await printer_factory()
        fake_jpeg = b"\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x00\x00\x01\x00\x01\x00\x00"

        # Simulate a running broadcaster: one active stream entry + buffered frame.
        active_streams = {f"{printer.id}-fanout": MagicMock()}
        last_frames = {printer.id: fake_jpeg}

        with (
            patch("backend.app.api.routes.camera._active_streams", active_streams),
            patch("backend.app.api.routes.camera._last_frames", last_frames),
            patch("backend.app.api.routes.camera.capture_camera_frame", new_callable=AsyncMock) as mock_capture,
        ):
            response = await async_client.get(f"/api/v1/printers/{printer.id}/camera/snapshot")

        assert response.status_code == 200
        assert response.content == fake_jpeg
        # The fresh-capture path must NOT have been taken — that's the whole point.
        mock_capture.assert_not_called()

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_camera_snapshot_external_camera_success(self, async_client: AsyncClient, printer_factory):
        """Verify snapshot uses external camera when configured."""
        printer = await printer_factory(
            external_camera_enabled=True,
            external_camera_url="http://192.168.1.50/mjpeg",
            external_camera_type="mjpeg",
        )

        fake_jpeg = b"\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x00\x00\x01\x00\x01\x00\x00"

        with patch(
            "backend.app.services.external_camera.capture_frame",
            new_callable=AsyncMock,
            return_value=fake_jpeg,
        ):
            response = await async_client.get(f"/api/v1/printers/{printer.id}/camera/snapshot")

        assert response.status_code == 200
        assert response.headers["content-type"] == "image/jpeg"
        assert response.content == fake_jpeg

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_camera_snapshot_external_camera_failure(self, async_client: AsyncClient, printer_factory):
        """Verify 503 when external camera capture fails."""
        printer = await printer_factory(
            external_camera_enabled=True,
            external_camera_url="http://192.168.1.50/mjpeg",
            external_camera_type="mjpeg",
        )

        with patch(
            "backend.app.services.external_camera.capture_frame",
            new_callable=AsyncMock,
            return_value=None,
        ):
            response = await async_client.get(f"/api/v1/printers/{printer.id}/camera/snapshot")

        assert response.status_code == 503
        assert "external camera" in response.json()["detail"].lower()

    # ========================================================================
    # Camera Stream Endpoint
    # ========================================================================

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_camera_stream_printer_not_found(self, async_client: AsyncClient):
        """Verify 404 when streaming camera for non-existent printer."""
        response = await async_client.get("/api/v1/printers/99999/camera/stream")

        assert response.status_code == 404

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_camera_stream_fps_validation(self, async_client: AsyncClient, printer_factory):
        """Verify FPS parameter is validated and clamped."""
        printer = await printer_factory()

        # FPS should be clamped between 1 and 30
        # Testing that the endpoint accepts various FPS values without error
        # (actual streaming would require mocking ffmpeg)

        with patch("backend.app.api.routes.camera.get_ffmpeg_path", return_value=None):
            # With no ffmpeg, stream should return error message but not crash
            response = await async_client.get(
                f"/api/v1/printers/{printer.id}/camera/stream",
                params={"fps": 100},  # Should be clamped to 30
            )
            # Response will be a streaming response with error
            assert response.status_code == 200

    # ========================================================================
    # Plate Detection Endpoints
    # ========================================================================

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_plate_detection_status_printer_not_found(self, async_client: AsyncClient):
        """Verify 404 when checking plate detection status for non-existent printer."""
        response = await async_client.get("/api/v1/printers/99999/camera/plate-detection/status")

        assert response.status_code == 404

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_plate_detection_status_opencv_not_available(self, async_client: AsyncClient, printer_factory):
        """Verify plate detection status returns unavailable when OpenCV not installed."""
        printer = await printer_factory()

        with patch("backend.app.services.plate_detection.OPENCV_AVAILABLE", False):
            response = await async_client.get(f"/api/v1/printers/{printer.id}/camera/plate-detection/status")

        assert response.status_code == 200
        result = response.json()
        assert result["available"] is False
        assert result["calibrated"] is False

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_plate_detection_status_success(self, async_client: AsyncClient, printer_factory):
        """Verify plate detection status returns correctly when OpenCV available."""
        printer = await printer_factory()

        # OpenCV is available in test environment, just check the response structure
        response = await async_client.get(f"/api/v1/printers/{printer.id}/camera/plate-detection/status")

        assert response.status_code == 200
        result = response.json()
        assert "available" in result
        assert "calibrated" in result

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_check_plate_empty_printer_not_found(self, async_client: AsyncClient):
        """Verify 404 when checking plate for non-existent printer."""
        response = await async_client.get("/api/v1/printers/99999/camera/check-plate")

        assert response.status_code == 404

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_check_plate_empty_success_structure(self, async_client: AsyncClient, printer_factory):
        """Verify check plate returns proper structure when OpenCV available."""
        printer = await printer_factory()

        # Mock PlateDetectionResult to avoid camera timeout
        mock_result = MagicMock()
        mock_result.is_empty = True
        mock_result.confidence = 0.95
        mock_result.difference_percent = 0.5
        mock_result.message = "Plate appears empty"
        mock_result.needs_calibration = False
        mock_result.debug_image = None
        mock_result.to_dict.return_value = {
            "is_empty": True,
            "confidence": 0.95,
            "difference_percent": 0.5,
            "message": "Plate appears empty",
            "has_debug_image": False,
            "needs_calibration": False,
        }

        # Mock PlateDetector for reference count
        mock_detector = MagicMock()
        mock_detector.get_calibration_count.return_value = 0
        mock_detector.MAX_REFERENCES = 5

        with (
            patch("backend.app.services.plate_detection.is_plate_detection_available", return_value=True),
            patch("backend.app.services.plate_detection.check_plate_empty", new_callable=AsyncMock) as mock_check,
            patch("backend.app.services.plate_detection.PlateDetector", return_value=mock_detector),
        ):
            mock_check.return_value = mock_result
            response = await async_client.get(f"/api/v1/printers/{printer.id}/camera/check-plate")

        assert response.status_code == 200
        result = response.json()
        assert "is_empty" in result
        assert "confidence" in result
        assert "message" in result

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_calibrate_plate_printer_not_found(self, async_client: AsyncClient):
        """Verify 404 when calibrating plate for non-existent printer."""
        response = await async_client.post("/api/v1/printers/99999/camera/plate-detection/calibrate")

        assert response.status_code == 404

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_calibrate_plate_success_structure(self, async_client: AsyncClient, printer_factory):
        """Verify calibrate endpoint responds with proper structure."""
        printer = await printer_factory()

        # Mock calibrate_plate at the source module to avoid camera timeout
        with (
            patch("backend.app.services.plate_detection.is_plate_detection_available", return_value=True),
            patch("backend.app.services.plate_detection.calibrate_plate", new_callable=AsyncMock) as mock_calibrate,
        ):
            mock_calibrate.return_value = (True, "Calibration saved (1/5 references)", 0)
            response = await async_client.post(f"/api/v1/printers/{printer.id}/camera/plate-detection/calibrate")

        assert response.status_code == 200
        result = response.json()
        assert result["success"] is True
        assert "index" in result

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_delete_calibration_printer_not_found(self, async_client: AsyncClient):
        """Verify 404 when deleting calibration for non-existent printer."""
        response = await async_client.delete("/api/v1/printers/99999/camera/plate-detection/calibrate")

        assert response.status_code == 404

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_delete_calibration_success(self, async_client: AsyncClient, printer_factory):
        """Verify delete calibration returns proper structure."""
        printer = await printer_factory()

        with patch("backend.app.services.plate_detection.is_plate_detection_available", return_value=True):
            response = await async_client.delete(f"/api/v1/printers/{printer.id}/camera/plate-detection/calibrate")

        assert response.status_code == 200
        result = response.json()
        assert "success" in result
        assert "message" in result

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_get_references_printer_not_found(self, async_client: AsyncClient):
        """Verify 404 when getting references for non-existent printer."""
        response = await async_client.get("/api/v1/printers/99999/camera/plate-detection/references")

        assert response.status_code == 404

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_get_references_opencv_not_available(self, async_client: AsyncClient, printer_factory):
        """Verify get references returns unavailable when OpenCV not installed."""
        printer = await printer_factory()

        with patch("backend.app.services.plate_detection.OPENCV_AVAILABLE", False):
            response = await async_client.get(f"/api/v1/printers/{printer.id}/camera/plate-detection/references")

        assert response.status_code == 503

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_get_references_success(self, async_client: AsyncClient, printer_factory):
        """Verify get references returns proper structure."""
        printer = await printer_factory()

        # Mock OpenCV availability and PlateDetector
        mock_detector = MagicMock()
        mock_detector.get_references.return_value = []
        mock_detector.MAX_REFERENCES = 5

        with (
            patch("backend.app.services.plate_detection.is_plate_detection_available", return_value=True),
            patch("backend.app.services.plate_detection.PlateDetector", return_value=mock_detector),
        ):
            response = await async_client.get(f"/api/v1/printers/{printer.id}/camera/plate-detection/references")

        assert response.status_code == 200
        result = response.json()
        assert "references" in result
        assert "max_references" in result
        assert isinstance(result["references"], list)

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_update_reference_label_printer_not_found(self, async_client: AsyncClient):
        """Verify 404 when updating reference label for non-existent printer."""
        response = await async_client.put(
            "/api/v1/printers/99999/camera/plate-detection/references/0", params={"label": "New Label"}
        )

        assert response.status_code == 404

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_delete_reference_printer_not_found(self, async_client: AsyncClient):
        """Verify 404 when deleting reference for non-existent printer."""
        response = await async_client.delete("/api/v1/printers/99999/camera/plate-detection/references/0")

        assert response.status_code == 404

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_get_reference_thumbnail_printer_not_found(self, async_client: AsyncClient):
        """Verify 404 when getting reference thumbnail for non-existent printer."""
        response = await async_client.get("/api/v1/printers/99999/camera/plate-detection/references/0/thumbnail")

        assert response.status_code == 404

    # ========================================================================
    # USB Camera Endpoint
    # ========================================================================

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_list_usb_cameras_returns_list(self, async_client: AsyncClient):
        """Verify USB cameras endpoint returns a list of cameras."""
        response = await async_client.get("/api/v1/printers/usb-cameras")

        assert response.status_code == 200
        result = response.json()
        assert "cameras" in result
        assert isinstance(result["cameras"], list)

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_list_usb_cameras_structure(self, async_client: AsyncClient):
        """Verify USB cameras endpoint returns proper structure for each camera."""
        with patch("backend.app.services.external_camera.list_usb_cameras") as mock_list:
            mock_list.return_value = [
                {"device": "/dev/video0", "name": "Logitech Webcam C920", "index": 0},
                {"device": "/dev/video2", "name": "USB Camera", "index": 2},
            ]

            response = await async_client.get("/api/v1/printers/usb-cameras")

        assert response.status_code == 200
        result = response.json()
        assert len(result["cameras"]) == 2
        assert result["cameras"][0]["device"] == "/dev/video0"
        assert result["cameras"][0]["name"] == "Logitech Webcam C920"
        assert result["cameras"][1]["device"] == "/dev/video2"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_list_usb_cameras_empty_on_non_linux(self, async_client: AsyncClient):
        """Verify USB cameras endpoint returns empty list on non-Linux systems."""
        with patch("backend.app.services.external_camera.list_usb_cameras") as mock_list:
            # Simulate non-Linux system (no /dev/video* devices)
            mock_list.return_value = []

            response = await async_client.get("/api/v1/printers/usb-cameras")

        assert response.status_code == 200
        result = response.json()
        assert result["cameras"] == []