mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 11:12:35 +02:00
maziggy
29379e3be7
Reporter @Andlar94 hit a permanent "Build plate not empty" on every print start on an A1 with an external RTSP camera. The runtime auto-check at main.py:1819 called check_plate_empty with use_external=external_camera_enabled, but the manual UI routes (camera.py) declared use_external: bool = False and the frontend client always sent use_external=false. So calibration captured a built-in frame and stamped it as the reference; the runtime check captured an external frame and diffed it against that reference -- a permanent mismatch. Centralise the default on the backend: both routes now take bool | None, deriving the default from the printer's external_camera_enabled + external_camera_url + external_camera_type. The frontend client stops sending the flag unless the caller explicitly sets it, so the existing UI call sites immediately benefit and any future caller gets the right camera automatically. Explicit overrides still win. Adds 4 regression tests pinning the new default for both the external-enabled and external-disabled cases, plus the explicit override path so a future "always built-in" caller stays supported.
731 lines
32 KiB
Python
731 lines
32 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()
|
|
|
|
@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
|
|
|
|
# ------------------------------------------------------------------
|
|
# Regression: #1359 — the manual UI check/calibrate routes must derive
|
|
# use_external from the printer's external_camera_enabled setting when
|
|
# the caller omits the flag. Otherwise the UI calibrates against the
|
|
# built-in camera while the runtime auto-check at print start uses the
|
|
# external one, producing a permanent "build plate not empty".
|
|
# ------------------------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_check_plate_defaults_use_external_when_external_camera_enabled(
|
|
self, async_client: AsyncClient, printer_factory
|
|
):
|
|
"""Omitting use_external on a printer with external camera enabled
|
|
must call the service with use_external=True."""
|
|
printer = await printer_factory(
|
|
external_camera_enabled=True,
|
|
external_camera_url="http://192.168.1.50/mjpeg",
|
|
external_camera_type="mjpeg",
|
|
)
|
|
|
|
mock_result = MagicMock()
|
|
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_result.debug_image = None
|
|
|
|
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
|
|
assert mock_check.await_args.kwargs["use_external"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_check_plate_defaults_use_external_false_when_external_camera_disabled(
|
|
self, async_client: AsyncClient, printer_factory
|
|
):
|
|
"""Omitting use_external on a printer without an external camera
|
|
must call the service with use_external=False (built-in)."""
|
|
printer = await printer_factory() # external_camera_enabled defaults to False
|
|
|
|
mock_result = MagicMock()
|
|
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_result.debug_image = None
|
|
|
|
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
|
|
assert mock_check.await_args.kwargs["use_external"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_calibrate_plate_defaults_use_external_when_external_camera_enabled(
|
|
self, async_client: AsyncClient, printer_factory
|
|
):
|
|
"""Calibrating with use_external omitted on an external-camera-enabled
|
|
printer captures the reference from the external camera — matching
|
|
what the runtime check at print start will compare against (#1359)."""
|
|
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.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
|
|
assert mock_calibrate.await_args.kwargs["use_external"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_calibrate_plate_explicit_use_external_false_overrides_default(
|
|
self, async_client: AsyncClient, printer_factory
|
|
):
|
|
"""An explicit use_external=false from the caller still wins even
|
|
when the printer has an external camera configured, so power users
|
|
can force a built-in-camera reference if they ever need to."""
|
|
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.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?use_external=false"
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert mock_calibrate.await_args.kwargs["use_external"] is False
|
|
|
|
@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"] == []
|