2614ae8880
🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
99 lines
3.7 KiB
Python
99 lines
3.7 KiB
Python
import pytest
|
|
from mqtt_home.device_registry import (
|
|
create_device, get_device, list_devices, delete_device,
|
|
update_device, send_command, handle_state_update, get_device_logs,
|
|
get_dashboard_stats,
|
|
)
|
|
from mqtt_home.schemas import DeviceCreate, DeviceUpdate
|
|
|
|
|
|
async def test_create_and_get_device(db_session):
|
|
data = DeviceCreate(name="客厅灯", type="light", mqtt_topic="home/light")
|
|
device = await create_device(db_session, data)
|
|
assert device.name == "客厅灯"
|
|
|
|
fetched = await get_device(db_session, device.id)
|
|
assert fetched is not None
|
|
assert fetched.id == device.id
|
|
|
|
|
|
async def test_list_devices(db_session):
|
|
await create_device(db_session, DeviceCreate(name="设备1", type="switch", mqtt_topic="t1"))
|
|
await create_device(db_session, DeviceCreate(name="设备2", type="sensor", mqtt_topic="t2"))
|
|
devices = await list_devices(db_session)
|
|
assert len(devices) == 2
|
|
|
|
|
|
async def test_update_device(db_session):
|
|
device = await create_device(db_session, DeviceCreate(name="灯", type="light", mqtt_topic="t"))
|
|
updated = await update_device(db_session, device.id, DeviceUpdate(name="新名字"))
|
|
assert updated.name == "新名字"
|
|
|
|
|
|
async def test_delete_device(db_session):
|
|
device = await create_device(db_session, DeviceCreate(name="灯", type="light", mqtt_topic="t"))
|
|
assert await delete_device(db_session, device.id) is True
|
|
assert await get_device(db_session, device.id) is None
|
|
|
|
|
|
async def test_delete_nonexistent_device(db_session):
|
|
assert await delete_device(db_session, "nonexistent") is False
|
|
|
|
|
|
async def test_send_command(db_session):
|
|
device = await create_device(
|
|
db_session, DeviceCreate(name="灯", type="light", mqtt_topic="t", command_topic="t/set")
|
|
)
|
|
published = {}
|
|
|
|
async def mock_publish(topic, payload):
|
|
published[topic] = payload
|
|
|
|
log = await send_command(db_session, device.id, '{"state":"on"}', mock_publish)
|
|
assert log is not None
|
|
assert log.direction == "tx"
|
|
assert published["t/set"] == '{"state":"on"}'
|
|
|
|
|
|
async def test_send_command_no_command_topic(db_session):
|
|
device = await create_device(db_session, DeviceCreate(name="传感器", type="sensor", mqtt_topic="t"))
|
|
with pytest.raises(ValueError, match="no command_topic"):
|
|
await send_command(db_session, device.id, '{"value":1}')
|
|
|
|
|
|
async def test_handle_state_update(db_session):
|
|
device = await create_device(db_session, DeviceCreate(name="灯", type="light", mqtt_topic="home/light"))
|
|
updated = await handle_state_update(db_session, "home/light", '{"state":"on"}')
|
|
assert updated is not None
|
|
assert updated.state == '{"state":"on"}'
|
|
assert updated.is_online is True
|
|
|
|
|
|
async def test_handle_state_update_unknown_topic(db_session):
|
|
result = await handle_state_update(db_session, "unknown/topic", "on")
|
|
assert result is None
|
|
|
|
|
|
async def test_get_device_logs(db_session):
|
|
device = await create_device(
|
|
db_session, DeviceCreate(name="灯", type="light", mqtt_topic="t", command_topic="t/set")
|
|
)
|
|
async def noop(t, p):
|
|
pass
|
|
await send_command(db_session, device.id, '{"state":"on"}', noop)
|
|
await send_command(db_session, device.id, '{"state":"off"}', noop)
|
|
|
|
logs = await get_device_logs(db_session, device.id, limit=10)
|
|
assert len(logs) == 2
|
|
|
|
|
|
async def test_dashboard_stats(db_session):
|
|
await create_device(db_session, DeviceCreate(name="在线设备", type="switch", mqtt_topic="t1"))
|
|
await create_device(db_session, DeviceCreate(name="离线设备", type="sensor", mqtt_topic="t2"))
|
|
await handle_state_update(db_session, "t1", "online")
|
|
|
|
stats = await get_dashboard_stats(db_session)
|
|
assert stats["total_devices"] == 2
|
|
assert stats["online_devices"] == 1
|
|
assert stats["offline_devices"] == 1
|