abb170ace6
🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
79 lines
2.5 KiB
Python
79 lines
2.5 KiB
Python
import json
|
|
import pytest
|
|
from unittest.mock import AsyncMock
|
|
|
|
from mqtt_home.discovery import parse_discovery_topic, handle_discovery_message
|
|
from mqtt_home.config import Settings
|
|
from mqtt_home.mqtt_client import MqttClient
|
|
|
|
|
|
@pytest.fixture
|
|
def settings():
|
|
return Settings(
|
|
mqtt_host="localhost",
|
|
emqx_api_url="http://localhost:18083/api/v5",
|
|
emqx_api_key="test-key",
|
|
emqx_api_secret="test-secret",
|
|
database_url="sqlite+aiosqlite:///:memory:",
|
|
)
|
|
|
|
|
|
def test_parse_discovery_topic_valid():
|
|
result = parse_discovery_topic("homeassistant/light/abc123/config")
|
|
assert result is not None
|
|
assert result["component"] == "light"
|
|
assert result["node_id"] == "abc123"
|
|
|
|
|
|
def test_parse_discovery_topic_nested_node():
|
|
result = parse_discovery_topic("homeassistant/sensor/room/temperature/config")
|
|
assert result is not None
|
|
assert result["component"] == "sensor"
|
|
assert result["node_id"] == "room/temperature"
|
|
|
|
|
|
def test_parse_discovery_topic_invalid():
|
|
assert parse_discovery_topic("other/topic/config") is None
|
|
assert parse_discovery_topic("homeassistant/light/abc") is None
|
|
assert parse_discovery_topic("homeassistant/light/abc/status") is None
|
|
|
|
|
|
async def test_handle_discovery_creates_device(db_session, settings):
|
|
mqtt_client = MqttClient(settings)
|
|
mqtt_client.subscribe = AsyncMock()
|
|
|
|
payload = json.dumps({
|
|
"name": "客厅灯",
|
|
"state_topic": "home/living/light/status",
|
|
"command_topic": "home/living/light/set",
|
|
"device_class": "light",
|
|
})
|
|
|
|
device = await handle_discovery_message(
|
|
"homeassistant/light/living_room/config",
|
|
payload,
|
|
db_session,
|
|
mqtt_client,
|
|
)
|
|
assert device is not None
|
|
assert device.name == "客厅灯"
|
|
assert device.protocol == "ha_discovery"
|
|
assert device.mqtt_topic == "home/living/light/status"
|
|
mqtt_client.subscribe.assert_called_once_with("home/living/light/status", qos=1)
|
|
|
|
|
|
async def test_handle_discovery_duplicate(db_session, settings):
|
|
mqtt_client = MqttClient(settings)
|
|
mqtt_client.subscribe = AsyncMock()
|
|
|
|
payload = json.dumps({
|
|
"name": "灯",
|
|
"state_topic": "home/light",
|
|
})
|
|
|
|
await handle_discovery_message("homeassistant/light/test/config", payload, db_session, mqtt_client)
|
|
device2 = await handle_discovery_message("homeassistant/light/test/config", payload, db_session, mqtt_client)
|
|
# Same discovery_topic should not create duplicate
|
|
assert device2 is not None
|
|
mqtt_client.subscribe.assert_called_once() # Only called once
|