mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 19:21:33 +02:00
maziggy
09d8e7d682
The snapshot endpoint always used the internal printer camera even when an external camera was configured. Now checks for external camera first, matching the stream endpoint pattern. Also added retry logic (3 attempts, 2s delay) to MJPEG and RTSP stream generators so they reconnect on timeout instead of silently ending the stream.
523 lines
22 KiB
Python
523 lines
22 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
|
|
mock_process = MagicMock()
|
|
mock_process.returncode = None
|
|
mock_process.terminate = MagicMock()
|
|
|
|
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
|
|
mock_process1 = MagicMock()
|
|
mock_process1.returncode = None
|
|
mock_process1.terminate = MagicMock()
|
|
|
|
mock_process2 = MagicMock()
|
|
mock_process2.returncode = None
|
|
mock_process2.terminate = MagicMock()
|
|
|
|
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"] == []
|