172 lines
6.5 KiB
Python
172 lines
6.5 KiB
Python
from datetime import datetime
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.core.config import settings
|
|
from app.db.sqlite import get_connection, init_db
|
|
from app.main import app
|
|
from app.repositories.alarm_repo import AlarmRepository
|
|
from app.repositories.json_config_repo import JsonConfigRepository
|
|
from app.schemas.platform import AlarmEvent
|
|
from app.services.platform_service import platform_service
|
|
|
|
|
|
client = TestClient(app)
|
|
|
|
|
|
def test_healthcheck() -> None:
|
|
response = client.get("/")
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "ok"
|
|
|
|
|
|
def test_realtime_endpoint() -> None:
|
|
response = client.get("/api/real-time-data")
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["code"] == 200
|
|
assert "line_list" in payload["data"]
|
|
|
|
|
|
def test_alarm_list_filters(tmp_path) -> None:
|
|
settings.alarm_db_path = tmp_path / "alarm.db"
|
|
init_db()
|
|
|
|
with get_connection() as connection:
|
|
connection.execute("DELETE FROM alarm_event")
|
|
connection.commit()
|
|
|
|
repo = AlarmRepository()
|
|
repo.save_alarm(
|
|
AlarmEvent(
|
|
alarm_type="保护报警",
|
|
time=datetime(2026, 5, 16, 10, 0, 0),
|
|
no="L1",
|
|
type="voltage",
|
|
content="A 相过压",
|
|
level="严重",
|
|
)
|
|
)
|
|
repo.save_alarm(
|
|
AlarmEvent(
|
|
alarm_type="状态报警",
|
|
time=datetime(2026, 5, 16, 11, 0, 0),
|
|
no="L2",
|
|
type="current",
|
|
content="B 相过流",
|
|
level="一般",
|
|
)
|
|
)
|
|
|
|
response = client.get(
|
|
"/api/alarm/list",
|
|
params={
|
|
"page": 1,
|
|
"size": 20,
|
|
"no": "L2",
|
|
"type": "current",
|
|
"start_time": "2026-05-16T10:30:00",
|
|
"end_time": "2026-05-16T11:30:00",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["code"] == 200
|
|
assert len(payload["data"]) == 1
|
|
assert payload["data"][0]["no"] == "L2"
|
|
assert payload["data"][0]["type"] == "current"
|
|
|
|
|
|
def test_config_query_endpoints(tmp_path) -> None:
|
|
old_repo = platform_service.config_repo
|
|
platform_service.config_repo = JsonConfigRepository(tmp_path)
|
|
|
|
try:
|
|
platform_service.config_repo.write_device_config(
|
|
{
|
|
"password": "hashed-password",
|
|
"hardware_version": {
|
|
"board_version": "B001.001.002",
|
|
"display_version": "S001.001.001",
|
|
"other_version": "Y001.001.001",
|
|
},
|
|
"software_version": {
|
|
"display_program": "001.001.001",
|
|
"communication_program": "001.001.001",
|
|
"measurement_program": "001.001.001",
|
|
},
|
|
"net": [
|
|
{"nic": "网卡一", "ip": "192.168.1.10", "mask": "255.255.255.0", "gateway": "192.168.1.1", "protocol": "Modbus TCP"}
|
|
],
|
|
"uart": [
|
|
{"port": "COM1", "baud": 9600, "parity": "NONE", "data_bits": 8, "stop_bits": 1, "protocol": "Modbus RTU"}
|
|
],
|
|
}
|
|
)
|
|
platform_service.config_repo.write_channel_config(
|
|
{
|
|
"ai_channel": [{"ch": 1, "singal_type": "4-20mA", "line_no": 1, "type": "UA", "limit_low": 0, "limit_high": 20}],
|
|
"ao_channel": [{"ch": 1, "singal_type": "1-5v", "line_no": 2, "type": "UA", "limit_low": 0, "limit_high": 20}],
|
|
}
|
|
)
|
|
platform_service.config_repo.write_setting_config(
|
|
{
|
|
"line_alarm_setting": [
|
|
{"line_no": 1, "over_limit_alarm": [{"category": "电压", "limit": 180, "delay": 180, "output_node": "开出1", "enabled": True}], "fault_alarm": []},
|
|
{"line_no": 2, "over_limit_alarm": [{"category": "电流", "limit": 200, "delay": 120, "output_node": "开出2", "enabled": True}], "fault_alarm": []},
|
|
],
|
|
"ai_alarm_setting": [{"channel_no": 1, "singal_type": "4-20mA", "limit_low": 0, "limit_high": 20, "delay": 180, "output_node": "开出1", "enabled": True}],
|
|
"system_config": {"time_sync": "manual", "brightness": 83, "screen_saver": 60},
|
|
}
|
|
)
|
|
|
|
headers = {"X-API-Token": settings.auth_password}
|
|
|
|
device_response = client.get("/api/config/device", headers=headers)
|
|
channel_response = client.get("/api/config/channel", headers=headers)
|
|
line_alarm_response = client.get("/api/config/line_alarm_setting", headers=headers, params={"line_no": 2})
|
|
ai_alarm_response = client.get("/api/config/ai_alarm_setting", headers=headers)
|
|
system_response = client.get("/api/config/system", headers=headers)
|
|
|
|
assert device_response.status_code == 200
|
|
assert channel_response.status_code == 200
|
|
assert line_alarm_response.status_code == 200
|
|
assert ai_alarm_response.status_code == 200
|
|
assert system_response.status_code == 200
|
|
|
|
assert device_response.json()["data"]["password"] == ""
|
|
assert device_response.json()["data"]["net"][0]["ip"] == "192.168.1.10"
|
|
assert channel_response.json()["data"]["ai_channel"][0]["singal_type"] == "4-20mA"
|
|
assert line_alarm_response.json()["data"]["line_no"] == 2
|
|
assert line_alarm_response.json()["data"]["over_limit_alarm"][0]["category"] == "电流"
|
|
assert ai_alarm_response.json()["data"][0]["channel_no"] == 1
|
|
assert system_response.json()["data"]["brightness"] == 83
|
|
finally:
|
|
platform_service.config_repo = old_repo
|
|
|
|
|
|
def test_save_line_alarm_setting_stores_list(tmp_path) -> None:
|
|
old_repo = platform_service.config_repo
|
|
platform_service.config_repo = JsonConfigRepository(tmp_path)
|
|
|
|
try:
|
|
headers = {"X-API-Token": settings.auth_password}
|
|
response = client.post(
|
|
"/api/config/line_alarm_setting",
|
|
headers=headers,
|
|
json={
|
|
"line_no": 3,
|
|
"over_limit_alarm": [{"category": "频率", "limit": 51.5, "delay": 30, "output_node": "开出1", "enabled": True}],
|
|
"fault_alarm": [{"category": "PT断线", "delay": 20, "output_node": "开出2", "enabled": False}],
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
|
|
saved = platform_service.config_repo.read_setting_config()
|
|
assert isinstance(saved["line_alarm_setting"], list)
|
|
assert saved["line_alarm_setting"][0]["line_no"] == 3
|
|
finally:
|
|
platform_service.config_repo = old_repo
|