mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-10-07 06:31:22 +02:00
maziggy
475e34ebda
The ML API previously called back into Bambuddy to fetch snapshots, which failed behind reverse proxies with external auth (Authelia, etc.). Now the detection loop captures the JPEG locally and POSTs it directly as multipart form data — no callback URL, no nonce cache, no external_url dependency.
300 lines
12 KiB
Python
300 lines
12 KiB
Python
"""Unit tests for Obico detection service (#172)."""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from backend.app.schemas.settings import AppSettingsUpdate
|
|
from backend.app.services.obico_detection import ObicoDetectionService
|
|
from backend.app.services.obico_smoothing import WARMUP_FRAMES
|
|
|
|
FAKE_JPEG = b"\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x00\x00\x01\x00\x01\x00\x00\xff\xd9"
|
|
|
|
|
|
class TestSettingsSchemaValidators:
|
|
"""Guard rails on the new obico_* AppSettings fields."""
|
|
|
|
def test_sensitivity_accepts_valid_values(self):
|
|
for value in ("low", "medium", "high"):
|
|
u = AppSettingsUpdate(obico_sensitivity=value)
|
|
assert u.obico_sensitivity == value
|
|
|
|
def test_sensitivity_rejects_garbage(self):
|
|
with pytest.raises(ValueError, match="obico_sensitivity"):
|
|
AppSettingsUpdate(obico_sensitivity="extreme")
|
|
|
|
def test_action_accepts_valid_values(self):
|
|
for value in ("notify", "pause", "pause_and_off"):
|
|
assert AppSettingsUpdate(obico_action=value).obico_action == value
|
|
|
|
def test_action_rejects_garbage(self):
|
|
with pytest.raises(ValueError, match="obico_action"):
|
|
AppSettingsUpdate(obico_action="explode")
|
|
|
|
def test_enabled_printers_accepts_empty(self):
|
|
assert AppSettingsUpdate(obico_enabled_printers="").obico_enabled_printers == ""
|
|
assert AppSettingsUpdate(obico_enabled_printers=None).obico_enabled_printers is None
|
|
|
|
def test_enabled_printers_accepts_int_array(self):
|
|
u = AppSettingsUpdate(obico_enabled_printers="[1, 2, 3]")
|
|
assert u.obico_enabled_printers == "[1, 2, 3]"
|
|
|
|
def test_enabled_printers_rejects_non_json(self):
|
|
with pytest.raises(ValueError, match="valid JSON"):
|
|
AppSettingsUpdate(obico_enabled_printers="1,2,3")
|
|
|
|
def test_enabled_printers_rejects_non_list(self):
|
|
with pytest.raises(ValueError, match="JSON array"):
|
|
AppSettingsUpdate(obico_enabled_printers='{"1": true}')
|
|
|
|
def test_enabled_printers_rejects_non_int_elements(self):
|
|
with pytest.raises(ValueError, match="JSON array"):
|
|
AppSettingsUpdate(obico_enabled_printers='[1, "two"]')
|
|
|
|
def test_poll_interval_bounds(self):
|
|
with pytest.raises(ValueError):
|
|
AppSettingsUpdate(obico_poll_interval=4)
|
|
with pytest.raises(ValueError):
|
|
AppSettingsUpdate(obico_poll_interval=121)
|
|
assert AppSettingsUpdate(obico_poll_interval=10).obico_poll_interval == 10
|
|
|
|
|
|
class TestGetStatus:
|
|
def test_empty_initial_status(self):
|
|
svc = ObicoDetectionService()
|
|
s = svc.get_status()
|
|
assert s["is_running"] is False
|
|
assert s["per_printer"] == {}
|
|
assert s["history"] == []
|
|
assert "low" in s["thresholds"] and "high" in s["thresholds"]
|
|
|
|
|
|
class TestTestConnection:
|
|
@pytest.mark.asyncio
|
|
async def test_empty_url_via_route(self):
|
|
"""Service does not special-case empty URL — the route does."""
|
|
svc = ObicoDetectionService()
|
|
# This will fail DNS/connect, but should return ok=False
|
|
result = await svc.test_connection("http://nonexistent-obico-host-xyz.invalid:3333")
|
|
assert result["ok"] is False
|
|
assert result["error"] is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_healthy_response_is_ok(self):
|
|
svc = ObicoDetectionService()
|
|
mock_response = MagicMock(status_code=200, text="ok")
|
|
mock_client = MagicMock()
|
|
mock_client.get = AsyncMock(return_value=mock_response)
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
with patch("backend.app.services.obico_detection.httpx.AsyncClient", return_value=mock_client):
|
|
result = await svc.test_connection("http://obico:3333")
|
|
assert result["ok"] is True
|
|
assert result["status_code"] == 200
|
|
assert result["body"] == "ok"
|
|
assert result["error"] is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_non_ok_body_is_not_ok(self):
|
|
svc = ObicoDetectionService()
|
|
mock_response = MagicMock(status_code=200, text="something else")
|
|
mock_client = MagicMock()
|
|
mock_client.get = AsyncMock(return_value=mock_response)
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
with patch("backend.app.services.obico_detection.httpx.AsyncClient", return_value=mock_client):
|
|
result = await svc.test_connection("http://obico:3333/")
|
|
assert result["ok"] is False
|
|
assert result["body"] == "something else"
|
|
|
|
|
|
class TestPollOneStateLifecycle:
|
|
"""Confirms per-printer state is reset when a new print starts."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_task_name_resets_state(self):
|
|
svc = ObicoDetectionService()
|
|
# Seed a state that has been running for a while
|
|
from backend.app.services.obico_smoothing import PrintState
|
|
|
|
seeded = PrintState()
|
|
for _ in range(WARMUP_FRAMES + 5):
|
|
seeded.update(0.5)
|
|
svc._states[1] = seeded
|
|
svc._state_keys[1] = "old_task"
|
|
svc._action_fired[1] = True
|
|
|
|
settings = {
|
|
"enabled": True,
|
|
"ml_url": "http://obico:3333",
|
|
"sensitivity": "medium",
|
|
"action": "notify",
|
|
"poll_interval": 10,
|
|
"enabled_printers": None,
|
|
}
|
|
status = MagicMock(state="RUNNING", task_name="new_task", subtask_name="")
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"detections": []}
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_client = MagicMock()
|
|
mock_client.post = AsyncMock(return_value=mock_response)
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
with (
|
|
patch("backend.app.services.obico_detection.httpx.AsyncClient", return_value=mock_client),
|
|
patch.object(svc, "_capture_frame", new=AsyncMock(return_value=FAKE_JPEG)),
|
|
):
|
|
await svc._check_printer(1, status, settings)
|
|
|
|
# State was reset (frame_count is 1 after the single update, not 36)
|
|
assert svc._states[1].frame_count == 1
|
|
assert svc._state_keys[1] == "new_task"
|
|
assert svc._action_fired[1] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ml_api_error_does_not_crash(self):
|
|
svc = ObicoDetectionService()
|
|
settings = {
|
|
"enabled": True,
|
|
"ml_url": "http://obico:3333",
|
|
"sensitivity": "medium",
|
|
"action": "notify",
|
|
"poll_interval": 10,
|
|
"enabled_printers": None,
|
|
}
|
|
status = MagicMock(state="RUNNING", task_name="job", subtask_name="")
|
|
|
|
mock_client = MagicMock()
|
|
mock_client.post = AsyncMock(side_effect=RuntimeError("connection refused"))
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
with (
|
|
patch("backend.app.services.obico_detection.httpx.AsyncClient", return_value=mock_client),
|
|
patch.object(svc, "_capture_frame", new=AsyncMock(return_value=FAKE_JPEG)),
|
|
):
|
|
await svc._check_printer(1, status, settings)
|
|
|
|
assert svc._last_error is not None
|
|
assert "connection refused" in svc._last_error
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_failure_fires_action_only_once(self):
|
|
"""Once a failure has fired for a print, subsequent failures should not re-fire."""
|
|
svc = ObicoDetectionService()
|
|
settings = {
|
|
"enabled": True,
|
|
"ml_url": "http://obico:3333",
|
|
"sensitivity": "medium",
|
|
"action": "notify",
|
|
"poll_interval": 10,
|
|
"enabled_printers": None,
|
|
}
|
|
status = MagicMock(state="RUNNING", task_name="job", subtask_name="")
|
|
|
|
# Seed state so the next frame crosses HIGH immediately
|
|
from backend.app.services.obico_smoothing import PrintState
|
|
|
|
seeded = PrintState()
|
|
for _ in range(WARMUP_FRAMES + 500):
|
|
seeded.update(1.0)
|
|
svc._states[1] = seeded
|
|
svc._state_keys[1] = "job"
|
|
svc._action_fired[1] = False
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"detections": [["failure", 0.9, [0, 0, 1, 1]]]}
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_client = MagicMock()
|
|
mock_client.post = AsyncMock(return_value=mock_response)
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
with (
|
|
patch("backend.app.services.obico_detection.httpx.AsyncClient", return_value=mock_client),
|
|
patch("backend.app.services.obico_actions.execute_action", new=AsyncMock()) as mock_action,
|
|
patch.object(svc, "_capture_frame", new=AsyncMock(return_value=FAKE_JPEG)),
|
|
):
|
|
await svc._check_printer(1, status, settings)
|
|
assert mock_action.call_count == 1
|
|
await svc._check_printer(1, status, settings)
|
|
# Second call must not dispatch again
|
|
assert mock_action.call_count == 1
|
|
|
|
|
|
class TestCheckPrinterPostsImageDirectly:
|
|
"""The detection loop must POST JPEG bytes directly to the ML API."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ml_api_called_with_post_and_image_bytes(self):
|
|
svc = ObicoDetectionService()
|
|
settings = {
|
|
"enabled": True,
|
|
"ml_url": "http://obico:3333",
|
|
"sensitivity": "medium",
|
|
"action": "notify",
|
|
"poll_interval": 10,
|
|
"enabled_printers": None,
|
|
}
|
|
status = MagicMock(state="RUNNING", task_name="job", subtask_name="")
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"detections": []}
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_client = MagicMock()
|
|
mock_client.post = AsyncMock(return_value=mock_response)
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
with (
|
|
patch("backend.app.services.obico_detection.httpx.AsyncClient", return_value=mock_client),
|
|
patch.object(svc, "_capture_frame", new=AsyncMock(return_value=FAKE_JPEG)),
|
|
):
|
|
await svc._check_printer(1, status, settings)
|
|
|
|
# ML API was called via POST
|
|
mock_client.post.assert_called_once()
|
|
_args, kwargs = mock_client.post.call_args
|
|
# URL is the ML API /p/ endpoint
|
|
assert _args[0] == "http://obico:3333/p/"
|
|
# Image bytes sent as multipart file upload
|
|
files = kwargs["files"]
|
|
assert "img" in files
|
|
filename, data, content_type = files["img"]
|
|
assert filename == "snapshot.jpg"
|
|
assert data == FAKE_JPEG
|
|
assert content_type == "image/jpeg"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_capture_failure_skips_ml_call(self):
|
|
"""If we can't capture a frame, don't bother the ML API."""
|
|
svc = ObicoDetectionService()
|
|
settings = {
|
|
"enabled": True,
|
|
"ml_url": "http://obico:3333",
|
|
"sensitivity": "medium",
|
|
"action": "notify",
|
|
"poll_interval": 10,
|
|
"enabled_printers": None,
|
|
}
|
|
status = MagicMock(state="RUNNING", task_name="job", subtask_name="")
|
|
|
|
mock_client = MagicMock()
|
|
mock_client.post = AsyncMock()
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
with (
|
|
patch("backend.app.services.obico_detection.httpx.AsyncClient", return_value=mock_client),
|
|
patch.object(svc, "_capture_frame", new=AsyncMock(return_value=None)),
|
|
):
|
|
await svc._check_printer(1, status, settings)
|
|
|
|
mock_client.post.assert_not_called()
|
|
assert svc._last_error is not None
|
|
assert "Failed to capture snapshot" in svc._last_error
|