mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 03:01:21 +02:00
maziggy
06e5114a03
The printer card's AI badge collapsed every class that was not Warning or Failure into green Safe, and the service reported `safe` whenever it had no verdict. The state entry is created when a monitored print is first seen -- before the first snapshot, let alone the first inference -- so a rejected ML API token, an unreachable ML API, a failed capture and an unset External URL all rendered as a healthy watched print: green Safe at score 0.000. For a safety feature that is the worst failure mode available: it asserts the print is being watched exactly when it is not. The reporter read that badge and concluded the loop had never started. It had been calling the ML API every ten seconds and being turned away with a 401 -- invisible because Obico's auth layer rejects a bad token before its request log sees it, and because successful checks log nothing there either. Add two honest states. Not checking (amber) when the last poll produced no result, carrying the reason; Starting while a monitored print waits for its first result. Score and frame count are withheld while not checking, since 0.000 beside "Not checking" reads as a measurement rather than its absence. The reason is per printer, so a card names its own problem rather than whichever printer failed most recently, and stays behind settings:read because it can quote configured URLs -- the badge state does not, because whether a print is watched is not configuration. An unrecognised class now falls back to Starting, not Safe. Test Connection saves the form before probing, so a green result describes the configuration the loop actually runs with rather than what is typed in the boxes.
211 lines
9.3 KiB
Python
211 lines
9.3 KiB
Python
"""Integration tests for Obico API endpoints (#172 follow-up).
|
|
|
|
Verifies the /obico/cached-frame/{nonce} endpoint used by Obico's ML API to fetch
|
|
pre-captured JPEG frames. This endpoint lets the detection loop sidestep Obico's
|
|
hardcoded 5s read timeout by pre-populating a cache before issuing the ML call.
|
|
"""
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
from backend.app.services.obico_detection import _frame_cache, obico_detection_service, stash_frame
|
|
from backend.app.services.obico_smoothing import PrintState
|
|
|
|
FAKE_JPEG = b"\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x00\x00\x01\x00\x01\x00\x00\xff\xd9"
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clear_cache():
|
|
_frame_cache.clear()
|
|
yield
|
|
_frame_cache.clear()
|
|
|
|
|
|
class TestObicoCachedFrame:
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_valid_nonce_returns_jpeg(self, async_client: AsyncClient):
|
|
"""A stashed nonce returns the stored JPEG bytes with image/jpeg."""
|
|
nonce = await stash_frame(FAKE_JPEG)
|
|
response = await async_client.get(f"/api/v1/obico/cached-frame/{nonce}")
|
|
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_unknown_nonce_is_404(self, async_client: AsyncClient):
|
|
"""An unguessable URL must not leak that the endpoint exists — return 404."""
|
|
response = await async_client.get("/api/v1/obico/cached-frame/definitely-not-a-real-nonce")
|
|
assert response.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_nonce_is_single_use(self, async_client: AsyncClient):
|
|
"""A second fetch with the same nonce returns 404 — prevents replay."""
|
|
nonce = await stash_frame(FAKE_JPEG)
|
|
first = await async_client.get(f"/api/v1/obico/cached-frame/{nonce}")
|
|
assert first.status_code == 200
|
|
second = await async_client.get(f"/api/v1/obico/cached-frame/{nonce}")
|
|
assert second.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_endpoint_is_public(self, async_client: AsyncClient):
|
|
"""Obico's ML API can't send auth headers, so the nonce IS the credential.
|
|
The path must be in PUBLIC_API_PATTERNS (no auth wall)."""
|
|
nonce = await stash_frame(FAKE_JPEG)
|
|
# Intentionally omit any auth headers even if the fixture would normally inject them
|
|
response = await async_client.get(
|
|
f"/api/v1/obico/cached-frame/{nonce}",
|
|
headers={}, # no Authorization header
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_response_is_not_cached(self, async_client: AsyncClient):
|
|
"""Browsers/proxies must not hold onto the image after Obico consumes it."""
|
|
nonce = await stash_frame(FAKE_JPEG)
|
|
response = await async_client.get(f"/api/v1/obico/cached-frame/{nonce}")
|
|
assert response.status_code == 200
|
|
assert "no-store" in response.headers.get("cache-control", "")
|
|
|
|
|
|
class TestObicoPrinterStatus:
|
|
"""The lightweight /obico/printer-status endpoint for printer-card badges (#1546)."""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clear_detection_state(self):
|
|
obico_detection_service._states.clear()
|
|
obico_detection_service._last_class.clear()
|
|
obico_detection_service._errors.clear()
|
|
obico_detection_service._last_error = None
|
|
yield
|
|
obico_detection_service._states.clear()
|
|
obico_detection_service._last_class.clear()
|
|
obico_detection_service._errors.clear()
|
|
obico_detection_service._last_error = None
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_returns_per_printer_classification(self, async_client: AsyncClient):
|
|
state = PrintState()
|
|
state.update(0.5)
|
|
obico_detection_service._states[1] = state
|
|
obico_detection_service._last_class[1] = "warning"
|
|
|
|
response = await async_client.get("/api/v1/obico/printer-status")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "enabled" in data
|
|
# None = all printers monitored (no obico_enabled_printers subset configured)
|
|
assert data["monitored_printers"] is None
|
|
entry = data["per_printer"]["1"]
|
|
assert entry["class"] == "warning"
|
|
assert entry["frame_count"] == 1
|
|
assert isinstance(entry["score"], float)
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_empty_when_nothing_monitored(self, async_client: AsyncClient):
|
|
response = await async_client.get("/api/v1/obico/printer-status")
|
|
assert response.status_code == 200
|
|
assert response.json()["per_printer"] == {}
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_monitored_subset_is_returned(self, async_client: AsyncClient):
|
|
"""A configured obico_enabled_printers subset surfaces (as a sorted list) so
|
|
the frontend can show the idle badge only on monitored printers."""
|
|
update = await async_client.put("/api/v1/settings/", json={"obico_enabled_printers": "[3, 1]"})
|
|
assert update.status_code == 200
|
|
try:
|
|
response = await async_client.get("/api/v1/obico/printer-status")
|
|
assert response.json()["monitored_printers"] == [1, 3]
|
|
finally:
|
|
await async_client.put("/api/v1/settings/", json={"obico_enabled_printers": ""})
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_last_error_is_surfaced(self, async_client: AsyncClient):
|
|
"""The badge modal shows the service's last error (auth disabled in the
|
|
test env, so the settings:read gate on the field is open)."""
|
|
obico_detection_service._last_error = "Failed to capture snapshot for printer 1"
|
|
response = await async_client.get("/api/v1/obico/printer-status")
|
|
assert response.json()["last_error"] == "Failed to capture snapshot for printer 1"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_does_not_leak_settings(self, async_client: AsyncClient):
|
|
"""Unlike /obico/status, this endpoint is readable with printers:read only,
|
|
so it must not expose the ML URL or other configuration."""
|
|
response = await async_client.get("/api/v1/obico/printer-status")
|
|
data = response.json()
|
|
for key in ("ml_url", "action", "history", "poll_interval", "external_url_configured"):
|
|
assert key not in data
|
|
|
|
|
|
class TestObicoPrinterStatusNoVerdict:
|
|
"""A printer whose detection is not working must not read as monitored (#2952)."""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clear_detection_state(self):
|
|
obico_detection_service._states.clear()
|
|
obico_detection_service._last_class.clear()
|
|
obico_detection_service._errors.clear()
|
|
obico_detection_service._last_error = None
|
|
yield
|
|
obico_detection_service._states.clear()
|
|
obico_detection_service._last_class.clear()
|
|
obico_detection_service._errors.clear()
|
|
obico_detection_service._last_error = None
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_error_class_and_reason_reach_the_card(self, async_client: AsyncClient):
|
|
obico_detection_service._states[1] = PrintState()
|
|
obico_detection_service._errors[1] = "Obico ML API rejected the token (401)."
|
|
|
|
response = await async_client.get("/api/v1/obico/printer-status")
|
|
entry = response.json()["per_printer"]["1"]
|
|
assert entry["class"] == "error"
|
|
assert entry["error"] == "Obico ML API rejected the token (401)."
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_monitored_but_no_result_yet_is_unknown_not_safe(self, async_client: AsyncClient):
|
|
obico_detection_service._states[1] = PrintState()
|
|
|
|
response = await async_client.get("/api/v1/obico/printer-status")
|
|
entry = response.json()["per_printer"]["1"]
|
|
assert entry["class"] == "unknown"
|
|
assert entry["error"] is None
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_reason_is_withheld_without_settings_read_but_the_class_is_not(self):
|
|
"""The reason can name the ML API base or the External URL, so it stays
|
|
behind settings:read. Whether the print is being watched is not
|
|
configuration, so a printers:read user still gets the class."""
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
from backend.app.api.routes.obico import get_printer_status
|
|
|
|
obico_detection_service._states[1] = PrintState()
|
|
obico_detection_service._errors[1] = "ML API call failed: http://192.168.8.9:3333 refused"
|
|
|
|
user = MagicMock()
|
|
user.has_permission.return_value = False
|
|
|
|
# The route calls _load_settings for the enabled/monitored fields; the
|
|
# redaction under test is independent of them.
|
|
loaded = {"enabled": True, "enabled_printers": None}
|
|
with patch.object(obico_detection_service, "_load_settings", new=AsyncMock(return_value=loaded)):
|
|
data = await get_printer_status(user=user)
|
|
entry = data["per_printer"][1]
|
|
assert entry["class"] == "error"
|
|
assert entry["error"] is None
|
|
assert data["last_error"] is None
|
|
assert "192.168.8.9" not in str(data)
|