36c53e0e7c
Major additions across backend, frontend, and infrastructure: Backend: - IoT collector framework (Modbus TCP, MQTT, HTTP) with manager - Realistic Beijing solar/weather simulator with cloud transients - Alarm auto-checker with demo anomaly injection (3-4 events/hour) - Report generation (PDF/Excel) with sync fallback and E2E testing - Energy data CSV/XLSX export endpoint - WebSocket real-time broadcast at /ws/realtime - Alembic initial migration for all 14 tables - 77 pytest tests across 9 API routers Frontend: - Live notification badge with alarm count (was hardcoded 0) - Sankey energy flow diagram on dashboard - Device photos (SVG illustrations) on all device pages - Report download with status icons - Energy data export buttons (CSV/Excel) - WebSocket hook with auto-reconnect and polling fallback - BigScreen 2D responsive CSS (tablet/mobile) - Error handling improvements across pages Infrastructure: - PostgreSQL + TimescaleDB as primary database - Production docker-compose with nginx reverse proxy - Comprehensive Chinese README - .env.example with documentation - quick-start.sh deployment script - nginx config with gzip, caching, security headers Data: - 30-day realistic backfill (47K rows, weather-correlated) - 18 devices, 6 alarm rules, 15 historical alarm events - Beijing solar position model with seasonal variation Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
120 lines
4.5 KiB
Python
120 lines
4.5 KiB
Python
import pytest
|
|
from conftest import auth_header
|
|
|
|
|
|
class TestListDevices:
|
|
async def test_list_devices(self, client, admin_user, admin_token, seed_devices):
|
|
resp = await client.get("/api/v1/devices", headers=auth_header(admin_token))
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert "total" in body
|
|
assert "items" in body
|
|
assert body["total"] >= 1
|
|
|
|
async def test_list_devices_pagination(self, client, admin_user, admin_token, seed_devices):
|
|
resp = await client.get(
|
|
"/api/v1/devices", params={"page": 1, "page_size": 2},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert len(body["items"]) <= 2
|
|
|
|
async def test_list_devices_filter_by_type(self, client, admin_user, admin_token, seed_devices):
|
|
resp = await client.get(
|
|
"/api/v1/devices", params={"device_type": "pv_inverter"},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 200
|
|
for item in resp.json()["items"]:
|
|
assert item["device_type"] == "pv_inverter"
|
|
|
|
async def test_list_devices_unauthenticated(self, client):
|
|
resp = await client.get("/api/v1/devices")
|
|
assert resp.status_code == 401
|
|
|
|
|
|
class TestDeviceStats:
|
|
async def test_get_device_stats(self, client, admin_user, admin_token, seed_devices):
|
|
resp = await client.get("/api/v1/devices/stats", headers=auth_header(admin_token))
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert "online" in body
|
|
assert "offline" in body
|
|
assert "alarm" in body
|
|
assert "maintenance" in body
|
|
|
|
|
|
class TestDeviceTypes:
|
|
async def test_get_device_types(self, client, admin_user, admin_token, seed_device_types):
|
|
resp = await client.get("/api/v1/devices/types", headers=auth_header(admin_token))
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert isinstance(body, list)
|
|
assert len(body) >= 1
|
|
assert "code" in body[0]
|
|
|
|
|
|
class TestDeviceGroups:
|
|
async def test_get_device_groups(self, client, admin_user, admin_token, seed_device_groups):
|
|
resp = await client.get("/api/v1/devices/groups", headers=auth_header(admin_token))
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert isinstance(body, list)
|
|
|
|
|
|
class TestCreateDevice:
|
|
async def test_create_device_as_admin(self, client, admin_user, admin_token, seed_device_types):
|
|
resp = await client.post(
|
|
"/api/v1/devices",
|
|
json={
|
|
"name": "新设备", "code": "NEW-001", "device_type": "pv_inverter",
|
|
"rated_power": 200.0, "location": "A区屋顶",
|
|
},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["name"] == "新设备"
|
|
assert body["code"] == "NEW-001"
|
|
|
|
async def test_create_device_as_visitor_forbidden(self, client, normal_user, user_token, seed_device_types):
|
|
resp = await client.post(
|
|
"/api/v1/devices",
|
|
json={"name": "Test", "code": "T-001", "device_type": "meter"},
|
|
headers=auth_header(user_token),
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
|
|
class TestUpdateDevice:
|
|
async def test_update_device(self, client, admin_user, admin_token, seed_devices):
|
|
device = seed_devices[0]
|
|
resp = await client.put(
|
|
f"/api/v1/devices/{device.id}",
|
|
json={"location": "新位置"},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["location"] == "新位置"
|
|
|
|
async def test_update_nonexistent_device(self, client, admin_user, admin_token):
|
|
resp = await client.put(
|
|
"/api/v1/devices/99999",
|
|
json={"location": "nowhere"},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
class TestGetDevice:
|
|
async def test_get_single_device(self, client, admin_user, admin_token, seed_devices):
|
|
device = seed_devices[0]
|
|
resp = await client.get(f"/api/v1/devices/{device.id}", headers=auth_header(admin_token))
|
|
assert resp.status_code == 200
|
|
assert resp.json()["id"] == device.id
|
|
|
|
async def test_get_nonexistent_device(self, client, admin_user, admin_token):
|
|
resp = await client.get("/api/v1/devices/99999", headers=auth_header(admin_token))
|
|
assert resp.status_code == 404
|