393 lines
15 KiB
Python
393 lines
15 KiB
Python
from datetime import datetime
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.core.config import settings
|
|
from app.core.security import hash_password
|
|
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, LineData, RealtimeData, ValueGroup
|
|
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_realtime_precision_normalization() -> None:
|
|
realtime = RealtimeData(
|
|
line_list=[
|
|
LineData(
|
|
line_no=1,
|
|
pri_val=ValueGroup(
|
|
Ua=5765.439,
|
|
Ia=61.728,
|
|
),
|
|
sec_val=ValueGroup(
|
|
Ua=57.65439,
|
|
Ia=1.23456,
|
|
Pa=68.245,
|
|
Qa=9.876,
|
|
Sa=69.995,
|
|
PFa=0.98123,
|
|
Uab=100.98765,
|
|
frq=49.99994,
|
|
),
|
|
)
|
|
],
|
|
switch={"di1": 1},
|
|
ai_collect={"ai1": 4.5678},
|
|
)
|
|
|
|
normalized = platform_service._normalize_realtime_precision(realtime)
|
|
line = normalized.line_list[0]
|
|
|
|
assert line.pri_val.Ua == 5765.439
|
|
assert line.pri_val.Ia == 61.728
|
|
assert line.pri_val.Pa == 341225.0
|
|
assert line.pri_val.Qa == 49380.0
|
|
assert line.pri_val.Sa == 349975.0
|
|
assert line.pri_val.PFa == 0.981
|
|
assert line.pri_val.Uab == 10098.765
|
|
assert line.pri_val.frq == 50.0
|
|
|
|
assert line.sec_val.Ua == 57.654
|
|
assert line.sec_val.Ia == 1.235
|
|
assert line.sec_val.Pa == 68.25
|
|
assert line.sec_val.Qa == 9.88
|
|
assert line.sec_val.Sa == 70.0
|
|
assert line.sec_val.PFa == 0.981
|
|
assert line.sec_val.Uab == 100.988
|
|
assert line.sec_val.frq == 50.0
|
|
|
|
assert normalized.ai_collect["ai1"] == 4.57
|
|
|
|
|
|
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)
|
|
net_response = client.get("/api/config/device/net", headers=headers, params={"nic": "网卡一"})
|
|
uart_response = client.get("/api/config/device/uart", headers=headers, params={"port": "COM1"})
|
|
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 net_response.status_code == 200
|
|
assert uart_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 net_response.json()["data"]["nic"] == "网卡一"
|
|
assert uart_response.json()["data"]["port"] == "COM1"
|
|
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
|
|
|
|
|
|
def test_save_device_net_and_uart_by_key(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.001",
|
|
"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"},
|
|
{"nic": "网卡二", "ip": "192.168.1.56", "mask": "255.255.255.255", "gateway": "192.168.1.56", "protocol": "Modbus TCP"},
|
|
],
|
|
"uart": [
|
|
{"port": "COM1", "baud": 9600, "parity": "NONE", "data_bits": 8, "stop_bits": 1, "protocol": ""},
|
|
{"port": "COM2", "baud": 115200, "parity": "NONE", "data_bits": 8, "stop_bits": 1, "protocol": "Modbus RTU"},
|
|
],
|
|
}
|
|
)
|
|
|
|
headers = {"X-API-Token": settings.auth_password}
|
|
|
|
net_save = client.post(
|
|
"/api/config/device/net",
|
|
headers=headers,
|
|
json={"nic": "网卡二", "ip": "10.10.10.2", "mask": "255.255.255.0", "gateway": "10.10.10.1", "protocol": "IEC104"},
|
|
)
|
|
uart_save = client.post(
|
|
"/api/config/device/uart",
|
|
headers=headers,
|
|
json={"port": "COM2", "baud": 4800, "parity": "EVEN", "data_bits": 8, "stop_bits": 1, "protocol": "DLT645"},
|
|
)
|
|
full_save = client.post(
|
|
"/api/config/device",
|
|
headers=headers,
|
|
json={
|
|
"password": "",
|
|
"hardware_version": {
|
|
"board_version": "B001.001.003",
|
|
"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": "172.16.1.10", "mask": "255.255.255.0", "gateway": "172.16.1.1", "protocol": "Modbus TCP"}
|
|
],
|
|
"uart": [
|
|
{"port": "COM1", "baud": 19200, "parity": "ODD", "data_bits": 8, "stop_bits": 1, "protocol": "Modbus RTU"}
|
|
],
|
|
},
|
|
)
|
|
|
|
assert net_save.status_code == 200
|
|
assert uart_save.status_code == 200
|
|
assert full_save.status_code == 200
|
|
|
|
saved = platform_service.config_repo.read_device_config()
|
|
net_map = {item["nic"]: item for item in saved["net"]}
|
|
uart_map = {item["port"]: item for item in saved["uart"]}
|
|
|
|
assert net_map["网卡一"]["ip"] == "172.16.1.10"
|
|
assert net_map["网卡二"]["ip"] == "10.10.10.2"
|
|
assert net_map["网卡二"]["protocol"] == "IEC104"
|
|
assert uart_map["COM1"]["baud"] == 19200
|
|
assert uart_map["COM2"]["baud"] == 4800
|
|
assert uart_map["COM2"]["protocol"] == "DLT645"
|
|
assert saved["password"] == "hashed-password"
|
|
finally:
|
|
platform_service.config_repo = old_repo
|
|
|
|
|
|
def test_save_device_password_only_updates_password(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": "old-hash",
|
|
"hardware_version": {
|
|
"board_version": "B001.001.001",
|
|
"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"}
|
|
],
|
|
}
|
|
)
|
|
|
|
headers = {"X-API-Token": settings.auth_password}
|
|
response = client.post(
|
|
"/api/config/device/password",
|
|
headers=headers,
|
|
json={"password": "654321"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
saved = platform_service.config_repo.read_device_config()
|
|
assert saved["password"] == hash_password("654321")
|
|
assert saved["net"][0]["nic"] == "网卡一"
|
|
assert saved["uart"][0]["port"] == "COM1"
|
|
assert saved["hardware_version"]["board_version"] == "B001.001.001"
|
|
finally:
|
|
platform_service.config_repo = old_repo
|
|
|
|
|
|
def test_verify_password_reads_device_config_hash(tmp_path) -> None:
|
|
old_repo = platform_service.config_repo
|
|
old_config_dir = settings.config_dir
|
|
settings.config_dir = tmp_path
|
|
platform_service.config_repo = JsonConfigRepository(tmp_path)
|
|
|
|
try:
|
|
platform_service.config_repo.write_device_config(
|
|
{
|
|
"password": hash_password("654321"),
|
|
"hardware_version": {
|
|
"board_version": "B001.001.001",
|
|
"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": [],
|
|
"uart": [],
|
|
}
|
|
)
|
|
|
|
ok_response = client.post("/api/auth/verify-password", json={"password": "654321"})
|
|
fail_response = client.post("/api/auth/verify-password", json={"password": "admin123"})
|
|
|
|
assert ok_response.status_code == 200
|
|
assert ok_response.json()["data"] is True
|
|
assert fail_response.status_code == 200
|
|
assert fail_response.json()["data"] is False
|
|
finally:
|
|
settings.config_dir = old_config_dir
|
|
platform_service.config_repo = old_repo
|