mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 19:21:33 +02:00
maziggy
510005f043
streams when one viewer closes
1) Offline tiles now show OFF (not LIVE)
CameraWall.modeByPrinter assigned 'live' to any visible printer
without considering status.connected, so a disconnected X1C wasted
a live-budget slot AND rendered the red LIVE chip on top of the
WifiOff placeholder. Disconnected printers now map to 'paused' and
don't decrement liveBudget — the existing WifiOff + Off chip
rendering takes over.
2) /camera/stop no longer kills other viewers' streams
The cam-wall tile, EmbeddedCameraViewer, and the /camera/:id popup
all subscribe to the same fan-out broadcaster for a printer.
/camera/stop used to unconditionally shutdown_broadcaster() + kill
every ffmpeg process for the printer, so closing the embedded viewer
while the cam-wall tile of the same printer was live force-killed
the source the tile was pulling from — the tile's <img> errored.
New get_subscriber_count(key) accessor in camera_fanout.py exposes
the broadcaster's subscriber list length. /camera/stop now reads
that first; when >= 1 subscriber is still attached, return
{stopped: 0, skipped: true} and leave the broadcaster + ffmpeg
processes alone. The leaving viewer's HTTP teardown still runs the
natural iter_subscriber.finally -> unsubscribe path, so its slot is
released; the broadcaster keeps serving the other viewers. Single-
viewer close still hits the immediate force-teardown (count is 0).
814 lines
35 KiB
Python
814 lines
35 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}")
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_stop_camera_stream_skips_shutdown_when_subscribers_remain(
|
|
self, async_client: AsyncClient, printer_factory
|
|
):
|
|
"""Reference-count guard: when other viewers are still subscribed to the
|
|
broadcaster, /camera/stop must NOT force-shutdown — otherwise closing
|
|
the embedded viewer kills the cam-wall tile of the same printer.
|
|
Natural cleanup tears it down when the last HTTP connection closes.
|
|
"""
|
|
printer = await printer_factory()
|
|
|
|
mock_shutdown = AsyncMock(return_value=True)
|
|
mock_process = MagicMock()
|
|
mock_process.returncode = None
|
|
mock_process.pid = 88888
|
|
mock_process.terminate = MagicMock()
|
|
mock_process.wait = AsyncMock()
|
|
|
|
with (
|
|
patch("backend.app.api.routes.camera.get_subscriber_count", return_value=2),
|
|
patch("backend.app.api.routes.camera.shutdown_broadcaster", mock_shutdown),
|
|
patch("backend.app.api.routes.camera._active_streams", {f"{printer.id}-abc": mock_process}),
|
|
):
|
|
response = await async_client.post(f"/api/v1/printers/{printer.id}/camera/stop")
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["stopped"] == 0
|
|
assert result.get("skipped") is True
|
|
mock_shutdown.assert_not_awaited()
|
|
mock_process.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 Diagnose Endpoint (#1395 follow-up)
|
|
# ========================================================================
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_camera_diagnose_printer_not_found(self, async_client: AsyncClient):
|
|
response = await async_client.post("/api/v1/printers/99999/camera/diagnose")
|
|
assert response.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_camera_diagnose_returns_structured_result(self, async_client: AsyncClient, printer_factory):
|
|
"""Endpoint returns the per-stage shape the frontend modal renders."""
|
|
from backend.app.services.camera_diagnose import (
|
|
CameraDiagnoseResult,
|
|
CameraDiagnoseStage,
|
|
)
|
|
|
|
printer = await printer_factory()
|
|
|
|
fake = CameraDiagnoseResult(
|
|
printer_id=printer.id,
|
|
protocol="rtsp",
|
|
port=322,
|
|
profile="P2S",
|
|
overall_status="failed",
|
|
stages=[
|
|
CameraDiagnoseStage(name="tcp_reachable", status="ok", duration_ms=12),
|
|
CameraDiagnoseStage(name="first_frame", status="failed", duration_ms=15123, code="no_frame"),
|
|
],
|
|
summary_code="no_frame",
|
|
)
|
|
with patch(
|
|
"backend.app.services.camera_diagnose.diagnose_camera",
|
|
new_callable=AsyncMock,
|
|
return_value=fake,
|
|
):
|
|
response = await async_client.post(f"/api/v1/printers/{printer.id}/camera/diagnose")
|
|
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["printer_id"] == printer.id
|
|
assert body["protocol"] == "rtsp"
|
|
assert body["profile"] == "P2S"
|
|
assert body["overall_status"] == "failed"
|
|
assert body["summary_code"] == "no_frame"
|
|
assert [s["name"] for s in body["stages"]] == ["tcp_reachable", "first_frame"]
|
|
assert body["stages"][1]["code"] == "no_frame"
|
|
|
|
# ========================================================================
|
|
# 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"] == []
|