bambuddy/backend/tests/integration/test_obico_api.py

"""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)