matthias.lotz
8766597a6a
- Add tests/unit/test_config.py for config loading and validation behavior - Add tests/unit/test_device_manager.py for device add/remove/routing lifecycle - Keep existing bootstrap/exceptions tests as completed items - Update OPTIMIZATION_PLAN.md: Phase 4 status in progress, 4.1 test tasks checked Validation: - pytest tests/unit/test_config.py tests/unit/test_device_manager.py tests/unit/test_bootstrap.py tests/unit/test_exceptions.py - mypy tests/unit/test_config.py tests/unit/test_device_manager.py
81 lines
2.5 KiB
Python
81 lines
2.5 KiB
Python
"""Unit tests for DeviceManager lifecycle operations."""
|
|
|
|
from unittest.mock import Mock
|
|
|
|
from api.models import BridgeConfig, DeviceConfig, SessionConfig
|
|
from core.device_manager import DeviceManager
|
|
|
|
|
|
def _device_config(device_id: str, topic: str) -> DeviceConfig:
|
|
return DeviceConfig(
|
|
device_id=device_id,
|
|
machine_name=f"Machine {device_id}",
|
|
mqtt_topic=topic,
|
|
parser_type="shelly_pm",
|
|
device_status_timeout_s=120,
|
|
session_config=SessionConfig(
|
|
standby_threshold_w=20,
|
|
working_threshold_w=100,
|
|
start_debounce_s=3,
|
|
stop_debounce_s=15,
|
|
message_timeout_s=20,
|
|
heartbeat_interval_s=60,
|
|
),
|
|
)
|
|
|
|
|
|
def test_apply_config_adds_device_and_subscribes() -> None:
|
|
mqtt_client = Mock()
|
|
event_queue = Mock()
|
|
|
|
manager = DeviceManager(mqtt_client=mqtt_client, event_queue=event_queue)
|
|
|
|
config = BridgeConfig(mqtt=None, devices=[_device_config("dev-1", "dev-1/status/pm1:0")], version="1.0")
|
|
manager.apply_config(config)
|
|
|
|
mqtt_client.subscribe.assert_called_once_with("dev-1/status/pm1:0")
|
|
assert "dev-1" in manager.session_detectors
|
|
assert manager.device_map["dev-1/status/pm1:0"] == "dev-1"
|
|
|
|
|
|
def test_apply_config_removes_device_and_unsubscribes() -> None:
|
|
mqtt_client = Mock()
|
|
event_queue = Mock()
|
|
|
|
manager = DeviceManager(mqtt_client=mqtt_client, event_queue=event_queue)
|
|
start_config = BridgeConfig(mqtt=None, devices=[_device_config("dev-1", "dev-1/status/pm1:0")], version="1.0")
|
|
manager.apply_config(start_config)
|
|
|
|
mqtt_client.reset_mock()
|
|
|
|
manager.apply_config(BridgeConfig(mqtt=None, devices=[], version="1.0"))
|
|
|
|
mqtt_client.unsubscribe.assert_called_once_with("dev-1/status/pm1:0")
|
|
assert "dev-1" not in manager.session_detectors
|
|
assert "dev-1/status/pm1:0" not in manager.device_map
|
|
|
|
|
|
def test_route_message_updates_status_monitor_and_processes_power() -> None:
|
|
mqtt_client = Mock()
|
|
event_queue = Mock()
|
|
status_monitor = Mock()
|
|
|
|
parser = Mock()
|
|
parser.parse_message.return_value = {"apower": 123.4}
|
|
|
|
manager = DeviceManager(
|
|
mqtt_client=mqtt_client,
|
|
event_queue=event_queue,
|
|
parser=parser,
|
|
status_monitor=status_monitor,
|
|
)
|
|
|
|
manager.apply_config(
|
|
BridgeConfig(mqtt=None, devices=[_device_config("dev-1", "dev-1/status/pm1:0")], version="1.0")
|
|
)
|
|
|
|
manager.route_message("dev-1/status/pm1:0", {"apower": 123.4})
|
|
|
|
status_monitor.update_last_seen.assert_called_once_with("dev-1")
|
|
parser.parse_message.assert_called_once()
|