duck-preview/tests/test_inference_manager.py

"""Tests for InferenceManager — drop-if-busy, restart counter, model validation."""

from __future__ import annotations

import sys
from unittest.mock import MagicMock, patch

import pytest
from PySide6.QtWidgets import QApplication

from app.inference.worker_manager import InferenceManager

# Ensure a QApplication exists for tests that create Qt objects
_app = QApplication.instance() or QApplication(sys.argv)


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def _make_manager() -> InferenceManager:
    """Return an InferenceManager without starting any process."""
    mgr = InferenceManager.__new__(InferenceManager)
    mgr._model_path = None
    mgr._process = None
    mgr._input_queue = None
    mgr._output_queue = None
    mgr._stop_event = None
    mgr._busy = False
    mgr._frame_id = 0
    mgr._restart_count = 0
    mgr._last_result_time = 0.0
    mgr._paused = False
    return mgr


# ---------------------------------------------------------------------------
# Model path validation
# ---------------------------------------------------------------------------

class TestModelPathValidation:
    def test_start_emits_error_for_missing_file(self, tmp_path) -> None:
        """start() with non-existent path must NOT spawn a process."""
        mgr = InferenceManager()
        errors: list[str] = []
        mgr.inference_error.connect(errors.append)

        mgr.start(str(tmp_path / "nonexistent.pt"))

        assert errors, "Expected inference_error signal"
        assert mgr._process is None

    def test_start_does_not_raise_for_existing_file(self, tmp_path) -> None:
        """start() with existing file should attempt to start (we mock _start_worker)."""
        model_file = tmp_path / "model.pt"
        model_file.write_bytes(b"fake")

        mgr = InferenceManager()
        with patch.object(mgr, "_start_worker") as mock_start:
            mgr.start(str(model_file))
            mock_start.assert_called_once()


# ---------------------------------------------------------------------------
# Drop-if-busy logic
# ---------------------------------------------------------------------------

class TestDropIfBusy:
    def test_submit_frame_drops_when_busy(self) -> None:
        """submit_frame must not enqueue when _busy is True."""
        mgr = _make_manager()
        mgr._busy = True
        mgr._process = MagicMock()
        mgr._process.is_alive.return_value = True
        mgr._input_queue = MagicMock()

        frame = MagicMock()
        frame.isValid.return_value = True
        mgr.submit_frame(frame)

        mgr._input_queue.put_nowait.assert_not_called()

    def test_submit_frame_drops_when_paused(self) -> None:
        mgr = _make_manager()
        mgr._paused = True
        mgr._process = MagicMock()
        mgr._process.is_alive.return_value = True
        mgr._input_queue = MagicMock()

        frame = MagicMock()
        frame.isValid.return_value = True
        mgr.submit_frame(frame)

        mgr._input_queue.put_nowait.assert_not_called()

    def test_submit_frame_drops_when_not_running(self) -> None:
        mgr = _make_manager()
        mgr._process = None
        mgr._input_queue = MagicMock()

        frame = MagicMock()
        frame.isValid.return_value = True
        mgr.submit_frame(frame)

        mgr._input_queue.put_nowait.assert_not_called()

    def test_submit_frame_drops_invalid_frame(self) -> None:
        mgr = _make_manager()
        mgr._process = MagicMock()
        mgr._process.is_alive.return_value = True
        mgr._input_queue = MagicMock()

        frame = MagicMock()
        frame.isValid.return_value = False
        mgr.submit_frame(frame)

        mgr._input_queue.put_nowait.assert_not_called()


# ---------------------------------------------------------------------------
# Pause / resume
# ---------------------------------------------------------------------------

class TestPauseResume:
    def test_pause_sets_flag(self) -> None:
        mgr = _make_manager()
        assert mgr._paused is False
        mgr.pause()
        assert mgr._paused is True

    def test_resume_clears_flag(self) -> None:
        mgr = _make_manager()
        mgr.pause()
        mgr.resume()
        assert mgr._paused is False

    def test_is_paused_property(self) -> None:
        mgr = _make_manager()
        assert mgr.is_paused is False
        mgr.pause()
        assert mgr.is_paused is True


# ---------------------------------------------------------------------------
# Restart counter
# ---------------------------------------------------------------------------

class TestRestartCounter:
    def test_handle_crash_increments_counter(self) -> None:
        mgr = InferenceManager()
        mgr._model_path = "fake.pt"
        mgr._restart_count = 0

        with (
            patch.object(mgr, "_start_worker"),
            patch.object(mgr._poll_timer, "stop"),
            patch.object(mgr._watchdog_timer, "stop"),
        ):
            mgr._handle_crash("test crash")

        assert mgr._restart_count == 1

    def test_handle_crash_emits_error_after_max_restarts(self) -> None:
        from app.config import INFERENCE_MAX_RESTARTS

        mgr = InferenceManager()
        mgr._model_path = "fake.pt"
        mgr._restart_count = INFERENCE_MAX_RESTARTS

        errors: list[str] = []
        mgr.inference_error.connect(errors.append)

        with (
            patch.object(mgr, "_start_worker") as mock_start,
            patch.object(mgr._poll_timer, "stop"),
            patch.object(mgr._watchdog_timer, "stop"),
        ):
            mgr._handle_crash("final crash")

        assert errors, "Expected inference_error signal after max restarts"
        mock_start.assert_not_called()

    def test_stop_resets_restart_count(self) -> None:
        mgr = InferenceManager()
        mgr._restart_count = 2

        with patch.object(mgr, "_stop_worker"):
            mgr.stop()

        assert mgr._restart_count == 0


# ---------------------------------------------------------------------------
# is_running property
# ---------------------------------------------------------------------------

class TestIsRunning:
    def test_not_running_when_process_is_none(self) -> None:
        mgr = _make_manager()
        assert mgr.is_running is False

    def test_not_running_when_process_dead(self) -> None:
        mgr = _make_manager()
        proc = MagicMock()
        proc.is_alive.return_value = False
        mgr._process = proc
        assert mgr.is_running is False

    def test_running_when_process_alive(self) -> None:
        mgr = _make_manager()
        proc = MagicMock()
        proc.is_alive.return_value = True
        mgr._process = proc
        assert mgr.is_running is True


# ---------------------------------------------------------------------------
# Worker data structures
# ---------------------------------------------------------------------------

class TestWorkerDataStructures:
    def test_frame_packet_is_immutable(self) -> None:
        from app.inference.worker import FramePacket
        pkt = FramePacket(1, b"", 640, 480, 3)
        with pytest.raises(AttributeError):
            pkt.frame_id = 2  # type: ignore[misc]

    def test_result_packet_is_immutable(self) -> None:
        from app.inference.worker import ResultPacket
        pkt = ResultPacket(1, [], 640, 480)
        with pytest.raises(AttributeError):
            pkt.frame_id = 2  # type: ignore[misc]

    def test_select_device_returns_string(self) -> None:
        from app.inference.worker import _select_device
        device = _select_device()
        assert isinstance(device, str)
        assert device in ("cpu", "mps", "cuda")