639 lines
26 KiB
Python
639 lines
26 KiB
Python
import asyncio
|
||
from datetime import datetime, timedelta
|
||
|
||
from fastapi.testclient import TestClient
|
||
|
||
from app.adapters.device_client import MockDeviceClient
|
||
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, DeviceStatus, LineData, RealtimeData, ValueGroup
|
||
from app.services.platform_service import platform_service
|
||
from app.ws.manager import ws_manager
|
||
|
||
|
||
# 测试固定使用 Mock，避免受本地 .env 中 RTU/TCP 真机配置影响。
|
||
platform_service.device_client = MockDeviceClient()
|
||
|
||
client = TestClient(app)
|
||
|
||
|
||
def _assert_time_sync_in_window(value: str, before: datetime, after: datetime) -> None:
|
||
parsed = datetime.strptime(value, "%Y-%m-%d %H:%M:%S")
|
||
assert before - timedelta(seconds=1) <= parsed <= after + timedelta(seconds=1)
|
||
|
||
|
||
def test_healthcheck() -> None:
|
||
response = client.get("/")
|
||
assert response.status_code == 200
|
||
assert response.json()["status"] == "ok"
|
||
|
||
|
||
def test_device_status_endpoint() -> None:
|
||
response = client.get("/api/device-status")
|
||
assert response.status_code == 200
|
||
payload = response.json()
|
||
assert payload["code"] == 200
|
||
assert "self_check" in payload["data"]
|
||
assert "net4" in payload["data"]
|
||
assert "uart4" in payload["data"]
|
||
assert "frequency_limit" in payload["data"]
|
||
assert "ct_break" in payload["data"]
|
||
assert "pt_break" in payload["data"]
|
||
assert "status_word_00e6" in payload["data"]
|
||
assert "status_word_00e7" in payload["data"]
|
||
|
||
|
||
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_status_websocket_receives_payload() -> None:
|
||
with client.websocket_connect("/ws/status") as websocket:
|
||
asyncio.run(platform_service.poll_device_once())
|
||
payload = websocket.receive_json()
|
||
|
||
assert payload["type"] == "status"
|
||
assert "self_check" in payload["data"]
|
||
assert "net4" in payload["data"]
|
||
assert "uart4" in payload["data"]
|
||
assert "frequency_limit" in payload["data"]
|
||
|
||
|
||
def test_startup_self_check_saves_alarm_when_status_is_abnormal(tmp_path) -> None:
|
||
class AbnormalStatusMockClient(MockDeviceClient):
|
||
def startup_self_check(self) -> bool:
|
||
return True
|
||
|
||
def read_device_status(self) -> DeviceStatus:
|
||
return DeviceStatus(
|
||
self_check="异常",
|
||
net1="断开",
|
||
net2="正常",
|
||
net3="正常",
|
||
net4="正常",
|
||
uart1="正常",
|
||
uart2="断开",
|
||
uart3="正常",
|
||
uart4="正常",
|
||
)
|
||
|
||
old_db_path = settings.alarm_db_path
|
||
old_client = platform_service.device_client
|
||
settings.alarm_db_path = tmp_path / "alarm.db"
|
||
init_db()
|
||
with get_connection() as connection:
|
||
connection.execute("DELETE FROM alarm_event")
|
||
connection.commit()
|
||
platform_service.device_client = AbnormalStatusMockClient()
|
||
|
||
try:
|
||
result = platform_service.startup_self_check()
|
||
alarms = AlarmRepository().list_alarms(page=1, size=20)
|
||
|
||
assert result["success"] is True
|
||
assert result["status"]["self_check"] == "异常"
|
||
assert len(alarms) == 1
|
||
assert alarms[0]["event_type"] == 3
|
||
assert alarms[0]["event_code"] == 99
|
||
assert "装置自检=异常" in alarms[0]["content"]
|
||
assert "网口1=断开" in alarms[0]["content"]
|
||
assert "串口2=断开" in alarms[0]["content"]
|
||
finally:
|
||
platform_service.device_client = old_client
|
||
settings.alarm_db_path = old_db_path
|
||
|
||
|
||
def test_get_line_alarm_setting_saves_read_failure_event(tmp_path) -> None:
|
||
class FailingLineAlarmClient(MockDeviceClient):
|
||
def read_line_alarm_setting(self, line_no: int) -> dict:
|
||
raise RuntimeError("线路定值回读失败")
|
||
|
||
old_db_path = settings.alarm_db_path
|
||
old_client = platform_service.device_client
|
||
settings.alarm_db_path = tmp_path / "alarm.db"
|
||
init_db()
|
||
with get_connection() as connection:
|
||
connection.execute("DELETE FROM alarm_event")
|
||
connection.commit()
|
||
platform_service.device_client = FailingLineAlarmClient()
|
||
|
||
try:
|
||
result = platform_service.get_line_alarm_setting(2)
|
||
alarms = AlarmRepository().list_alarms(page=1, size=20)
|
||
|
||
assert result["line_no"] == 2
|
||
assert len(alarms) == 1
|
||
assert alarms[0]["event_type"] == 4
|
||
assert alarms[0]["event_code"] == 2
|
||
assert alarms[0]["line_code"] == 2
|
||
assert "读定值失败" in alarms[0]["content"]
|
||
assert "线路定值回读失败" in alarms[0]["content"]
|
||
finally:
|
||
platform_service.device_client = old_client
|
||
settings.alarm_db_path = old_db_path
|
||
|
||
|
||
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(
|
||
event_time=datetime(2026, 5, 16, 10, 0, 0),
|
||
line_code=1,
|
||
event_type=3,
|
||
event_code=1,
|
||
event_value=230,
|
||
content="A 相过压",
|
||
)
|
||
)
|
||
repo.save_alarm(
|
||
AlarmEvent(
|
||
event_time=datetime(2026, 5, 16, 11, 0, 0),
|
||
line_code=2,
|
||
event_type=5,
|
||
event_code=2,
|
||
event_value=456,
|
||
content="B 相过流",
|
||
)
|
||
)
|
||
|
||
response = client.get(
|
||
"/api/alarm/list",
|
||
params={
|
||
"page": 1,
|
||
"size": 20,
|
||
"line_code": 2,
|
||
"event_type": 5,
|
||
"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]["line_code"] == 2
|
||
assert payload["data"][0]["event_type"] == 5
|
||
assert payload["data"][0]["event_code"] == 2
|
||
assert payload["data"][0]["event_value"] == 456
|
||
|
||
|
||
def test_config_query_endpoints(tmp_path) -> None:
|
||
old_repo = platform_service.config_repo
|
||
old_client = platform_service.device_client
|
||
platform_service.config_repo = JsonConfigRepository(tmp_path)
|
||
platform_service.device_client = MockDeviceClient()
|
||
|
||
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"}
|
||
],
|
||
}
|
||
)
|
||
|
||
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": 1})
|
||
ai_alarm_response = client.get("/api/config/ai_alarm_setting", headers=headers)
|
||
system_response = client.get("/api/config/system", headers=headers)
|
||
before_time_sync = datetime.now()
|
||
time_sync_response = client.get("/api/config/time-sync", headers=headers)
|
||
after_time_sync = datetime.now()
|
||
|
||
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 time_sync_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 net_response.json()["data"]["ip"] == "192.168.1.10"
|
||
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"] == 1
|
||
assert line_alarm_response.json()["data"]["transformer_change"][0]["category"] == "PT变化"
|
||
assert ai_alarm_response.json()["data"][0]["channel_no"] == 1
|
||
assert system_response.json()["data"]["time_sync"] == "2026-05-25 12:34:56"
|
||
_assert_time_sync_in_window(time_sync_response.json()["data"]["time_sync"], before_time_sync, after_time_sync)
|
||
finally:
|
||
platform_service.device_client = old_client
|
||
platform_service.config_repo = old_repo
|
||
|
||
|
||
def test_protocol_config_endpoints_read_and_write_device_only(tmp_path) -> None:
|
||
old_repo = platform_service.config_repo
|
||
old_client = platform_service.device_client
|
||
platform_service.config_repo = JsonConfigRepository(tmp_path)
|
||
platform_service.device_client = MockDeviceClient()
|
||
|
||
try:
|
||
headers = {"X-API-Token": settings.auth_password}
|
||
channel_response = client.post(
|
||
"/api/config/channel",
|
||
headers=headers,
|
||
json={
|
||
"ai_channel": [{"ch": 1, "singal_type": "4-20mA", "line_no": 3, "type": "UA", "limit_low": 0, "limit_high": 20}],
|
||
"ao_channel": [{"ch": 2, "singal_type": "1~5V", "line_no": 2, "type": "Q", "limit_low": 1, "limit_high": 8}],
|
||
},
|
||
)
|
||
line_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}],
|
||
"transformer_change": [{"category": "PT变化", "value": 20, "enabled": True}, {"category": "CT变化", "value": 30, "enabled": True}],
|
||
},
|
||
)
|
||
ai_alarm_response = client.post(
|
||
"/api/config/ai_alarm_setting",
|
||
headers=headers,
|
||
json=[
|
||
{
|
||
"channel_no": 1,
|
||
"singal_type": "4-20mA",
|
||
"limit_low": 0,
|
||
"limit_high": 20,
|
||
"delay": 30,
|
||
"output_node": "开出1",
|
||
"enabled": True,
|
||
}
|
||
],
|
||
)
|
||
system_response = client.post(
|
||
"/api/config/system",
|
||
headers=headers,
|
||
json={"time_sync": "2026-05-25 13:00:00", "brightness": 88, "screen_saver": 90},
|
||
)
|
||
net_response = client.post(
|
||
"/api/config/device/net",
|
||
headers=headers,
|
||
json={"nic": "网卡二", "ip": "172.16.2.10", "mask": "255.255.255.0", "gateway": "172.16.2.1", "protocol": "Modbus TCP"},
|
||
)
|
||
query_channel = client.get("/api/config/channel", headers=headers)
|
||
query_line = client.get("/api/config/line_alarm_setting", headers=headers, params={"line_no": 3})
|
||
query_ai_alarm = client.get("/api/config/ai_alarm_setting", headers=headers)
|
||
query_system = client.get("/api/config/system", headers=headers)
|
||
query_net = client.get("/api/config/device/net", headers=headers, params={"nic": "网卡二"})
|
||
time_sync_response = client.post(
|
||
"/api/config/time-sync",
|
||
headers=headers,
|
||
json={"time_sync": "2026-05-25 14:00:00"},
|
||
)
|
||
before_query_time_sync = datetime.now()
|
||
query_time_sync = client.get("/api/config/time-sync", headers=headers)
|
||
after_query_time_sync = datetime.now()
|
||
|
||
assert channel_response.status_code == 200
|
||
assert line_response.status_code == 200
|
||
assert ai_alarm_response.status_code == 200
|
||
assert system_response.status_code == 200
|
||
assert net_response.status_code == 200
|
||
assert time_sync_response.status_code == 200
|
||
assert query_channel.json()["data"]["ai_channel"][0]["line_no"] == 3
|
||
assert query_line.json()["data"]["transformer_change"][0]["value"] == 20.0
|
||
assert query_ai_alarm.json()["data"][0]["delay"] == 30
|
||
assert query_system.json()["data"]["brightness"] == 88
|
||
assert query_net.json()["data"]["ip"] == "172.16.2.10"
|
||
assert query_net.json()["data"]["gateway"] == "172.16.2.1"
|
||
assert time_sync_response.json()["data"]["time_sync"] == "2026-05-25 14:00:00"
|
||
_assert_time_sync_in_window(query_time_sync.json()["data"]["time_sync"], before_query_time_sync, after_query_time_sync)
|
||
assert platform_service.config_repo.read_setting_config() == {}
|
||
assert platform_service.config_repo.read_device_config() == {}
|
||
finally:
|
||
platform_service.device_client = old_client
|
||
platform_service.config_repo = old_repo
|
||
|
||
|
||
def test_mock_time_sync_ws_payload_uses_system_time() -> None:
|
||
old_client = platform_service.device_client
|
||
platform_service.device_client = MockDeviceClient()
|
||
captured: dict = {}
|
||
|
||
async def fake_broadcast(channel: str, payload: dict) -> None:
|
||
captured["channel"] = channel
|
||
captured["payload"] = payload
|
||
|
||
original_broadcast = ws_manager.broadcast
|
||
ws_manager.broadcast = fake_broadcast
|
||
|
||
try:
|
||
before = datetime.now()
|
||
asyncio.run(platform_service.poll_time_sync_once())
|
||
after = datetime.now()
|
||
finally:
|
||
ws_manager.broadcast = original_broadcast
|
||
platform_service.device_client = old_client
|
||
|
||
assert captured["channel"] == "time-sync"
|
||
assert captured["payload"]["type"] == "time_sync"
|
||
_assert_time_sync_in_window(captured["payload"]["data"]["time_sync"], before, after)
|
||
|
||
|
||
def test_save_device_net_and_uart_by_key(tmp_path) -> None:
|
||
old_repo = platform_service.config_repo
|
||
old_client = platform_service.device_client
|
||
platform_service.config_repo = JsonConfigRepository(tmp_path)
|
||
platform_service.device_client = MockDeviceClient()
|
||
|
||
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"]}
|
||
net_query = client.get("/api/config/device/net", headers=headers, params={"nic": "网卡二"})
|
||
|
||
assert net_map["网卡一"]["ip"] == "172.16.1.10"
|
||
assert net_map["网卡二"]["ip"] == "192.168.1.56"
|
||
assert net_map["网卡二"]["protocol"] == "Modbus TCP"
|
||
assert uart_map["COM1"]["baud"] == 19200
|
||
assert uart_map["COM2"]["baud"] == 4800
|
||
assert uart_map["COM2"]["protocol"] == "DLT645"
|
||
assert net_query.status_code == 200
|
||
assert net_query.json()["data"]["ip"] == "10.10.10.2"
|
||
assert net_query.json()["data"]["gateway"] == "10.10.10.1"
|
||
assert saved["password"] == "hashed-password"
|
||
finally:
|
||
platform_service.device_client = old_client
|
||
platform_service.config_repo = old_repo
|
||
|
||
|
||
def test_save_device_net_and_uart_batch(tmp_path) -> None:
|
||
old_repo = platform_service.config_repo
|
||
old_client = platform_service.device_client
|
||
platform_service.config_repo = JsonConfigRepository(tmp_path)
|
||
platform_service.device_client = MockDeviceClient()
|
||
|
||
try:
|
||
headers = {"X-API-Token": settings.auth_password}
|
||
|
||
net_save = client.post(
|
||
"/api/config/device/net",
|
||
headers=headers,
|
||
json=[
|
||
{"nic": "网卡一", "ip": "192.168.1.10", "mask": "255.255.255.0", "gateway": "192.168.1.1", "protocol": "Modbus TCP"},
|
||
{"nic": "网卡二", "ip": "192.168.2.10", "mask": "255.255.255.0", "gateway": "192.168.2.1", "protocol": "Modbus TCP"},
|
||
],
|
||
)
|
||
uart_save = client.post(
|
||
"/api/config/device/uart",
|
||
headers=headers,
|
||
json=[
|
||
{"port": "COM1", "baud": 9600, "parity": "NONE", "data_bits": 8, "stop_bits": 1, "protocol": "Modbus RTU"},
|
||
{"port": "COM2", "baud": 115200, "parity": "EVEN", "data_bits": 8, "stop_bits": 2, "protocol": "DLT645"},
|
||
],
|
||
)
|
||
net_one = client.get("/api/config/device/net", headers=headers, params={"nic": "网卡一"})
|
||
net_two = client.get("/api/config/device/net", headers=headers, params={"nic": "网卡二"})
|
||
uart_one = client.get("/api/config/device/uart", headers=headers, params={"port": "COM1"})
|
||
uart_two = client.get("/api/config/device/uart", headers=headers, params={"port": "COM2"})
|
||
|
||
assert net_save.status_code == 200
|
||
assert uart_save.status_code == 200
|
||
assert net_save.json()["data"]["target"] == "net_batch"
|
||
assert net_save.json()["data"]["items"] == 2
|
||
assert uart_save.json()["data"]["target"] == "uart_batch"
|
||
assert uart_save.json()["data"]["items"] == 2
|
||
assert net_one.json()["data"]["ip"] == "192.168.1.10"
|
||
assert net_two.json()["data"]["gateway"] == "192.168.2.1"
|
||
assert uart_one.json()["data"]["protocol"] == "Modbus RTU"
|
||
assert uart_two.json()["data"]["baud"] == 115200
|
||
assert uart_two.json()["data"]["stop_bits"] == 2
|
||
finally:
|
||
platform_service.device_client = old_client
|
||
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
|