wled-screen-controller-mixed/server/tests/test_processor_manager.py

"""Tests for processor manager."""

import pytest

from wled_controller.core.processing.processor_manager import ProcessorDependencies, ProcessorManager


@pytest.fixture
def wled_url():
    """Provide test WLED device URL."""
    return "http://192.168.1.100"


@pytest.fixture
def mock_wled_responses():
    """Provide mock WLED API responses."""
    return {
        "info": {
            "name": "Test WLED",
            "ver": "0.14.0",
            "leds": {"count": 150},
            "brand": "WLED",
            "product": "FOSS",
            "mac": "AA:BB:CC:DD:EE:FF",
            "ip": "192.168.1.100",
        },
        "state": {"on": True, "bri": 255},
    }


@pytest.fixture
def processor_manager():
    """Provide processor manager instance."""
    return ProcessorManager(deps=ProcessorDependencies())


def test_processor_manager_init():
    """Test processor manager initialization."""
    manager = ProcessorManager(deps=ProcessorDependencies())
    assert manager is not None
    assert manager.get_all_devices() == []


def test_add_device(processor_manager):
    """Test adding a device."""
    processor_manager.add_device(
        device_id="test_device",
        device_url="http://192.168.1.100",
        led_count=150,
    )

    devices = processor_manager.get_all_devices()
    assert "test_device" in devices


def test_add_device_duplicate(processor_manager):
    """Test adding duplicate device fails."""
    processor_manager.add_device(
        device_id="test_device",
        device_url="http://192.168.1.100",
        led_count=150,
    )

    with pytest.raises(ValueError, match="already registered"):
        processor_manager.add_device(
            device_id="test_device",
            device_url="http://192.168.1.100",
            led_count=150,
        )


@pytest.mark.asyncio
async def test_remove_device(processor_manager):
    """Test removing a device."""
    processor_manager.add_device(
        device_id="test_device",
        device_url="http://192.168.1.100",
        led_count=150,
    )

    await processor_manager.remove_device("test_device")

    assert "test_device" not in processor_manager.get_all_devices()


@pytest.mark.asyncio
async def test_remove_device_not_found(processor_manager):
    """Test removing non-existent device fails."""
    with pytest.raises(ValueError, match="not found"):
        await processor_manager.remove_device("nonexistent")


def test_add_target(processor_manager):
    """Test adding a WLED target."""
    processor_manager.add_device(
        device_id="test_device",
        device_url="http://192.168.1.100",
        led_count=150,
    )

    processor_manager.add_target(
        target_id="target_1",
        device_id="test_device",
        fps=60,
    )

    state = processor_manager.get_target_state("target_1")
    assert state["target_id"] == "target_1"


def test_add_target_duplicate(processor_manager):
    """Test adding duplicate target fails."""
    processor_manager.add_device(
        device_id="test_device",
        device_url="http://192.168.1.100",
        led_count=150,
    )

    processor_manager.add_target(
        target_id="target_1",
        device_id="test_device",
    )

    with pytest.raises(ValueError, match="already registered"):
        processor_manager.add_target(
            target_id="target_1",
            device_id="test_device",
        )


def test_remove_target(processor_manager):
    """Test removing a target."""
    processor_manager.add_device(
        device_id="test_device",
        device_url="http://192.168.1.100",
        led_count=150,
    )

    processor_manager.add_target(
        target_id="target_1",
        device_id="test_device",
    )

    processor_manager.remove_target("target_1")

    with pytest.raises(ValueError, match="not found"):
        processor_manager.get_target_state("target_1")


def test_update_target_settings(processor_manager):
    """Test updating target settings."""
    processor_manager.add_device(
        device_id="test_device",
        device_url="http://192.168.1.100",
        led_count=150,
    )

    processor_manager.add_target(
        target_id="target_1",
        device_id="test_device",
    )

    processor_manager.update_target_settings("target_1", {"fps": 60})

    state = processor_manager.get_target_state("target_1")
    assert state["fps_target"] == 60


def test_update_device_info(processor_manager):
    """Test updating device info after registration."""
    processor_manager.add_device(
        device_id="test_device",
        device_url="http://192.168.1.100",
        led_count=150,
    )

    processor_manager.update_device_info(
        "test_device", led_count=200, device_url="http://192.168.1.101"
    )
    dev = processor_manager._devices["test_device"]
    assert dev.led_count == 200


def test_get_target_state(processor_manager):
    """Test getting target state."""
    processor_manager.add_device(
        device_id="test_device",
        device_url="http://192.168.1.100",
        led_count=150,
    )

    processor_manager.add_target(
        target_id="target_1",
        device_id="test_device",
        fps=30,
    )

    state = processor_manager.get_target_state("target_1")

    assert state["target_id"] == "target_1"
    assert state["processing"] is False


def test_get_target_state_not_found(processor_manager):
    """Test getting state for non-existent target."""
    with pytest.raises(ValueError, match="not found"):
        processor_manager.get_target_state("nonexistent")


def test_get_target_metrics(processor_manager):
    """Test getting target metrics."""
    processor_manager.add_device(
        device_id="test_device",
        device_url="http://192.168.1.100",
        led_count=150,
    )

    processor_manager.add_target(
        target_id="target_1",
        device_id="test_device",
    )

    metrics = processor_manager.get_target_metrics("target_1")

    assert metrics["target_id"] == "target_1"
    assert metrics["processing"] is False
    assert metrics["frames_processed"] == 0
    assert metrics["errors_count"] == 0


def test_target_type_detection(processor_manager):
    """Test target type detection via processor instances."""
    from wled_controller.storage.key_colors_output_target import KeyColorsSettings
    from wled_controller.core.processing.kc_target_processor import KCTargetProcessor
    from wled_controller.core.processing.wled_target_processor import WledTargetProcessor

    processor_manager.add_device(
        device_id="test_device",
        device_url="http://192.168.1.100",
        led_count=150,
    )

    processor_manager.add_target(
        target_id="wled_target",
        device_id="test_device",
    )

    processor_manager.add_kc_target(
        target_id="kc_target",
        picture_source_id="src_1",
        settings=KeyColorsSettings(),
    )

    assert isinstance(processor_manager._processors["kc_target"], KCTargetProcessor)
    assert isinstance(processor_manager._processors["wled_target"], WledTargetProcessor)


@pytest.mark.asyncio
async def test_stop_all(processor_manager):
    """Test stopping all processors."""
    processor_manager.add_device(
        device_id="test_device1",
        device_url="http://192.168.1.100",
        led_count=150,
    )
    processor_manager.add_device(
        device_id="test_device2",
        device_url="http://192.168.1.101",
        led_count=150,
    )

    processor_manager.add_target(
        target_id="target_1",
        device_id="test_device1",
    )
    processor_manager.add_target(
        target_id="target_2",
        device_id="test_device2",
    )

    await processor_manager.stop_all()

    assert processor_manager.is_target_processing("target_1") is False
    assert processor_manager.is_target_processing("target_2") is False