Files
tianpu-ems/backend/tests/test_users.py
Du Wenbo 36c53e0e7c feat: complete platform build-out to 95% benchmark-ready
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>
2026-04-02 18:46:42 +08:00

79 lines
2.8 KiB
Python

import pytest
from conftest import auth_header
class TestListUsers:
async def test_list_users_as_admin(self, client, admin_user, admin_token):
resp = await client.get("/api/v1/users", 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_users_as_visitor_forbidden(self, client, normal_user, user_token):
resp = await client.get("/api/v1/users", headers=auth_header(user_token))
assert resp.status_code == 403
async def test_list_users_unauthenticated(self, client):
resp = await client.get("/api/v1/users")
assert resp.status_code == 401
class TestCreateUser:
async def test_create_user_as_admin(self, client, admin_user, admin_token):
resp = await client.post(
"/api/v1/users",
json={"username": "newuser", "password": "newpass123", "full_name": "New User", "role": "visitor"},
headers=auth_header(admin_token),
)
assert resp.status_code == 200
body = resp.json()
assert body["username"] == "newuser"
assert "id" in body
async def test_create_user_as_visitor_forbidden(self, client, normal_user, user_token):
resp = await client.post(
"/api/v1/users",
json={"username": "another", "password": "pass123"},
headers=auth_header(user_token),
)
assert resp.status_code == 403
async def test_create_duplicate_user(self, client, admin_user, admin_token):
resp = await client.post(
"/api/v1/users",
json={"username": "testadmin", "password": "pass123"},
headers=auth_header(admin_token),
)
assert resp.status_code == 400
class TestUpdateUser:
async def test_update_user_as_admin(self, client, admin_user, normal_user, admin_token):
resp = await client.put(
f"/api/v1/users/{normal_user.id}",
json={"full_name": "Updated Name"},
headers=auth_header(admin_token),
)
assert resp.status_code == 200
async def test_update_nonexistent_user(self, client, admin_user, admin_token):
resp = await client.put(
"/api/v1/users/99999",
json={"full_name": "Ghost"},
headers=auth_header(admin_token),
)
assert resp.status_code == 404
class TestRoles:
async def test_list_roles(self, client, admin_user, admin_token, seed_roles):
resp = await client.get("/api/v1/users/roles", headers=auth_header(admin_token))
assert resp.status_code == 200
body = resp.json()
assert isinstance(body, list)
assert len(body) >= 1
assert "name" in body[0]
assert "display_name" in body[0]