957f489a0d
🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
94 lines
2.8 KiB
Python
94 lines
2.8 KiB
Python
import pytest
|
|
from httpx import AsyncClient, ASGITransport
|
|
|
|
from mqtt_home.main import app
|
|
from mqtt_home.database import Base, get_engine
|
|
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
|
|
from sqlalchemy.pool import StaticPool
|
|
from mqtt_home.database import get_db
|
|
|
|
TEST_DB = "sqlite+aiosqlite:///:memory:"
|
|
|
|
|
|
@pytest.fixture
|
|
async def client():
|
|
engine = create_async_engine(TEST_DB, connect_args={"check_same_thread": False}, poolclass=StaticPool)
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
|
|
test_factory = async_sessionmaker(engine, expire_on_commit=False)
|
|
|
|
async def override_get_db():
|
|
async with test_factory() as session:
|
|
yield session
|
|
|
|
app.dependency_overrides[get_db] = override_get_db
|
|
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as ac:
|
|
yield ac
|
|
|
|
app.dependency_overrides.clear()
|
|
await engine.dispose()
|
|
|
|
|
|
async def test_get_devices_empty(client):
|
|
resp = await client.get("/api/devices")
|
|
assert resp.status_code == 200
|
|
assert resp.json() == []
|
|
|
|
|
|
async def test_create_device(client):
|
|
resp = await client.post("/api/devices", json={
|
|
"name": "客厅灯",
|
|
"type": "light",
|
|
"mqtt_topic": "home/light",
|
|
"command_topic": "home/light/set",
|
|
})
|
|
assert resp.status_code == 201
|
|
data = resp.json()
|
|
assert data["name"] == "客厅灯"
|
|
assert data["type"] == "light"
|
|
|
|
|
|
async def test_get_device_detail(client):
|
|
create_resp = await client.post("/api/devices", json={
|
|
"name": "灯", "type": "switch", "mqtt_topic": "t"
|
|
})
|
|
device_id = create_resp.json()["id"]
|
|
|
|
resp = await client.get(f"/api/devices/{device_id}")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["name"] == "灯"
|
|
|
|
|
|
async def test_get_device_not_found(client):
|
|
resp = await client.get("/api/devices/nonexistent")
|
|
assert resp.status_code == 404
|
|
|
|
|
|
async def test_update_device(client):
|
|
create_resp = await client.post("/api/devices", json={
|
|
"name": "灯", "type": "switch", "mqtt_topic": "t"
|
|
})
|
|
device_id = create_resp.json()["id"]
|
|
|
|
resp = await client.put(f"/api/devices/{device_id}", json={"name": "新名字"})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["name"] == "新名字"
|
|
|
|
|
|
async def test_delete_device(client):
|
|
create_resp = await client.post("/api/devices", json={
|
|
"name": "灯", "type": "switch", "mqtt_topic": "t"
|
|
})
|
|
device_id = create_resp.json()["id"]
|
|
|
|
resp = await client.delete(f"/api/devices/{device_id}")
|
|
assert resp.status_code == 204
|
|
|
|
|
|
async def test_delete_device_not_found(client):
|
|
resp = await client.delete("/api/devices/nonexistent")
|
|
assert resp.status_code == 404
|