emcp/backend/tests/test_api.py

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)
        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