Files
bambuddy/backend/tests/integration/test_camera_api.py
T
maziggy 1eeeb37b42 Updated commit message (now includes the test fix):
Fix camera button permissions & ffmpeg process leak (#550)

  Camera button on printer card was clickable without camera:view
  permission. ffmpeg processes (~240MB each) accumulated after closing
  camera streams because: (1) stop endpoint called terminate() without
  wait()/kill(), (2) HTTP disconnect detection only ran between frames
  so was blocked when the generator was stuck on stdout read, and
  (3) no mechanism caught processes orphaned by generator abandonment
  or app restarts.

  - Add camera:view permission check + tooltip to camera button
  - Fix stop endpoint: terminate() → wait(2s) → kill() → wait()
  - Add background disconnect monitor (polls every 2s, kills ffmpeg
    directly on disconnect)
  - Add periodic /proc scan (every 60s) that SIGKILLs any ffmpeg
    with rtsps://bblp: not in an active stream
  - Add noCamera i18n key to all 6 locales
  - Fix camera API test mocks for async wait() and pid attribute
2026-02-28 12:58:31 +01:00

529 lines
23 KiB
Python

"""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()
# ========================================================================
# 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_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"] == []