mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 11:12:35 +02:00
maziggy
b99536cc33
- Remove 28 unused imports across 22 test files - Prefix 4 unused local variables with _ in app code (archives, bambu_mqtt, main) and remove 1 dead store - Consolidate import/import-from in test_plate_detection.py - Fix unreachable statement in test_archive_service.py - Simplify redundant comparison in timelapse_processor.py Resolves ~50 CodeQL py/unused-import, py/unused-local-variable, py/import-and-import-from, py/unreachable-statement, and py/redundant-comparison findings.
480 lines
21 KiB
Python
480 lines
21 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"]
|
|
|
|
# ========================================================================
|
|
# 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"] == []
|