matthias.lotz
ddd1e05f55
- parsers/registry.py: PARSER_REGISTRY mit shelly_pm-Eintrag; get_parser(), get_schema(), list_parser_types() als Public API - parsers/__init__.py: Registry-Funktionen exportiert - core/device_manager.py: globalen ShellyParser entfernt; DeviceManager verwaltet jetzt ein eigenes parser-Dict pro Device (per get_parser()) - api/server.py: GET /parsers Endpoint hinzugefügt (gibt get_schema() zurück) - tests/unit/test_parser_registry.py: 17 neue Tests (Registry-API, PARSER_REGISTRY-Integrität, DeviceManager-Integration) - tests/unit/test_device_manager.py: Test auf neues API angepasst (patch statt parser=-Argument) Tests: 63/63 passed
85 lines
2.7 KiB
Python
85 lines
2.7 KiB
Python
"""Unit tests for DeviceManager lifecycle operations."""
|
|
|
|
from unittest.mock import Mock, patch
|
|
|
|
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()
|
|
|
|
mock_parser = Mock()
|
|
mock_parser.parse_message.return_value = {"apower": 123.4}
|
|
|
|
manager = DeviceManager(
|
|
mqtt_client=mqtt_client,
|
|
event_queue=event_queue,
|
|
status_monitor=status_monitor,
|
|
)
|
|
|
|
with patch("core.device_manager.get_parser", return_value=mock_parser):
|
|
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")
|
|
mock_parser.parse_message.assert_called_once()
|