LGCompanion/tests/test_engine.py

"""Tests for the daemon engine."""

from __future__ import annotations

from unittest.mock import MagicMock, patch

from lgcompanion.config import Config, DeviceProfile
from lgcompanion.engine import Engine


def test_engine_force_surround() -> None:
    config = Config.default()
    with patch("lgcompanion.engine.switch_to") as mock_switch, patch(
        "lgcompanion.engine.route_filter_output_to_stereo"
    ) as mock_route, patch("lgcompanion.engine.sanity_check_sinks"):
        engine = Engine(config, force="surround", once=True)
        code = engine.run()
        assert code == 0
        mock_switch.assert_called_once()
        mock_route.assert_called_once()


def test_engine_force_stereo() -> None:
    config = Config.default()
    with patch("lgcompanion.engine.switch_to") as mock_switch, patch(
        "lgcompanion.engine.sanity_check_sinks"
    ):
        engine = Engine(config, force="stereo", once=True)
        code = engine.run()
        assert code == 0
        mock_switch.assert_called_once()


def test_engine_no_device_found() -> None:
    config = Config.default()
    with patch("lgcompanion.engine.find_hidraw_device", return_value=None), patch(
        "lgcompanion.engine.switch_to"
    ) as mock_switch, patch("lgcompanion.engine.sanity_check_sinks"):
        engine = Engine(config, once=True)
        code = engine.run()
        assert code == 1
        mock_switch.assert_called_once()


def test_engine_reloads_config_before_daemon_iteration() -> None:
    original = Config.default()
    updated = Config.default()
    updated.general.poll_interval = 9.0
    engine = Engine(original)
    with patch("lgcompanion.engine.Config.load", return_value=updated) as mock_load, patch(
        "lgcompanion.engine.sanity_check_sinks"
    ), patch.object(engine, "_iteration", side_effect=engine.stop):
        engine.run()
        mock_load.assert_called_once_with(None)
        assert engine.config is updated


def test_engine_reports_current_device() -> None:
    config = Config.default()
    profile = DeviceProfile(name="Test", vendor_id="0x1038", product_id="0x12AD")
    config.add_device(profile)
    device = MagicMock(path="/dev/hidraw0", profile=profile)
    on_device = MagicMock()
    with patch("lgcompanion.engine.find_hidraw_device", return_value=device), patch(
        "lgcompanion.engine.get_battery_level", return_value=80
    ), patch("lgcompanion.engine.switch_to"), patch(
        "lgcompanion.engine.route_filter_output_to_stereo"
    ), patch("lgcompanion.engine.sanity_check_sinks"):
        engine = Engine(config, once=True, on_device=on_device)
        engine.run()
        on_device.assert_called_once_with("Test")


def test_engine_connected_switch_to_surround() -> None:
    config = Config.default()
    profile = DeviceProfile(name="Test", vendor_id="0x1038", product_id="0x12AD")
    config.add_device(profile)
    device = MagicMock(path="/dev/hidraw0", profile=profile)
    with patch("lgcompanion.engine.find_hidraw_device", return_value=device), patch(
        "lgcompanion.engine.get_battery_level", return_value=80
    ), patch("lgcompanion.engine.switch_to") as mock_switch, patch(
        "lgcompanion.engine.route_filter_output_to_stereo"
    ) as mock_route, patch("lgcompanion.engine.sanity_check_sinks"):
        engine = Engine(config, once=True)
        code = engine.run()
        assert code == 0
        mock_switch.assert_called_once()
        mock_route.assert_called_once()


def _auto_engine(config: Config, device) -> Engine:
    engine = Engine(config, once=True)
    engine._previous_connected = True  # simulate already-connected state
    return engine


def test_engine_single_timeout_keeps_surround() -> None:
    config = Config.default()
    profile = DeviceProfile(name="Test", vendor_id="0x1038", product_id="0x12AD")
    config.add_device(profile)
    device = MagicMock(path="/dev/hidraw0", profile=profile)
    with patch("lgcompanion.engine.find_hidraw_device", return_value=device), patch(
        "lgcompanion.engine.get_battery_level", return_value=None
    ), patch("lgcompanion.engine.switch_to") as mock_switch:
        engine = _auto_engine(config, device)
        engine._iteration()
        mock_switch.assert_not_called()
        assert engine._missed_polls == 1


def test_engine_switches_to_stereo_after_repeated_timeouts() -> None:
    from lgcompanion.engine import MAX_MISSED_POLLS

    config = Config.default()
    profile = DeviceProfile(name="Test", vendor_id="0x1038", product_id="0x12AD")
    config.add_device(profile)
    device = MagicMock(path="/dev/hidraw0", profile=profile)
    with patch("lgcompanion.engine.find_hidraw_device", return_value=device), patch(
        "lgcompanion.engine.get_battery_level", return_value=None
    ), patch("lgcompanion.engine.switch_to") as mock_switch:
        engine = _auto_engine(config, device)
        for _ in range(MAX_MISSED_POLLS - 1):
            engine._iteration()
        mock_switch.assert_not_called()
        engine._iteration()
        mock_switch.assert_called_once_with(config.sinks.stereo, "беспроводная связь с наушниками потеряна")


def test_engine_successful_poll_resets_missed_counter() -> None:
    config = Config.default()
    profile = DeviceProfile(name="Test", vendor_id="0x1038", product_id="0x12AD")
    config.add_device(profile)
    device = MagicMock(path="/dev/hidraw0", profile=profile)
    with patch("lgcompanion.engine.find_hidraw_device", return_value=device), patch(
        "lgcompanion.engine.get_battery_level", side_effect=[None, 80]
    ), patch("lgcompanion.engine.switch_to"), patch(
        "lgcompanion.engine.route_filter_output_to_stereo"
    ):
        engine = _auto_engine(config, device)
        engine._iteration()
        assert engine._missed_polls == 1
        engine._iteration()
        assert engine._missed_polls == 0