455ed4ad5e
- Replace minimal database.py with full SQLAlchemy async setup - Create Device and DeviceLog models with proper relationships - Add tests for model creation and relationship functionality - Fixed lazy loading issue in tests by avoiding .count() on dynamic relationships
50 lines
1.3 KiB
Python
50 lines
1.3 KiB
Python
import pytest
|
|
from mqtt_home.models import Device, DeviceLog
|
|
|
|
|
|
async def test_create_device(db_session):
|
|
device = Device(
|
|
name="客厅灯",
|
|
type="light",
|
|
protocol="custom",
|
|
mqtt_topic="home/living/light",
|
|
command_topic="home/living/light/set",
|
|
)
|
|
db_session.add(device)
|
|
await db_session.commit()
|
|
await db_session.refresh(device)
|
|
|
|
assert device.id is not None
|
|
assert len(device.id) == 36
|
|
assert device.name == "客厅灯"
|
|
assert device.state is None
|
|
assert device.is_online is False
|
|
|
|
|
|
async def test_create_device_log(db_session):
|
|
device = Device(
|
|
name="温度传感器",
|
|
type="sensor",
|
|
protocol="custom",
|
|
mqtt_topic="home/temperature",
|
|
)
|
|
db_session.add(device)
|
|
await db_session.flush()
|
|
|
|
log = DeviceLog(
|
|
device_id=device.id,
|
|
direction="rx",
|
|
topic="home/temperature",
|
|
payload='{"temperature": 25.5}',
|
|
)
|
|
db_session.add(log)
|
|
await db_session.commit()
|
|
|
|
# Query logs directly instead of using relationship.count()
|
|
from sqlalchemy import select
|
|
result = await db_session.execute(
|
|
select(DeviceLog).where(DeviceLog.device_id == device.id)
|
|
)
|
|
logs = result.scalars().all()
|
|
assert len(logs) == 1
|