feat(alerts): 添加微信告警、WebSocket实时推送和地图功能

- 新增微信告警服务(wechat_service)和告警任务(alert_tasks)
- 新增 WebSocket 实时推送端点
- 新增监控管理模块(monitor)
- 增强统计仪表板:趋势图、区域分布、光功率历史
- 设备管理:添加坐标信息、标签系统、复合索引优化
- 前端:重构Dashboard/Charts页面,新增业务组件
- 新增5个数据库迁移(坐标、复合索引、标签、光功率历史、显示名)
- 更新部署配置和脚本
- 新增测试框架基础结构
This commit is contained in:
2026-06-11 15:16:29 +08:00
parent 3453441754
commit b2c20ec43d
45 changed files with 2089 additions and 107 deletions
+93
View File
@@ -0,0 +1,93 @@
"""SSH 输出解析测试"""
import pytest
from app.services.ssh_service import SSHService
def _make_svc():
return SSHService("10.0.0.1", "admin", "pass")
class TestParseOnuInfo:
def test_single_online(self):
svc = _make_svc()
output = """
Flags: S-Switched L-Loopback N-Not exist U-Up D-Down
Port MAC Status OAM State LOID Model Distance
0/0/1 1484-778f-aa60 Up OAM_Up test_loid H3C_ET704 1234m
"""
onu_dict, unknown = svc.parse_onu_info(output)
assert len(onu_dict) == 1
assert "1484-778f-aa60" in onu_dict
assert onu_dict["1484-778f-aa60"].status == "online"
def test_mixed_online_offline(self):
svc = _make_svc()
output = """
Flags: S-Switched L-Loopback N-Not exist U-Up D-Down
Port MAC Status OAM State LOID Model Distance
0/0/1 1484-778f-aa60 Up OAM_Up loid_a H3C_ET704 500m
0/0/2 1484-778f-bb70 Down OAM_Down loid_b Unknown <1000m
"""
onu_dict, unknown = svc.parse_onu_info(output)
mac_a = "1484-778f-aa60"
mac_b = "1484-778f-bb70"
assert onu_dict[mac_a].status == "online"
assert onu_dict[mac_b].status == "offline"
def test_more_marker_removal(self):
svc = _make_svc()
output = """
Flags: S-Switched L-Loopback N-Not exist U-Up D-Down
Port MAC Status OAM State LOID Model Distance
---- More ----
0/0/1 1484-778f-aa60 Up OAM_Up loid H3C_ET704 500m
---- More ----
0/0/2 1484-778f-bb70 Up OAM_Up loid2 H3C_ET704 800m
"""
onu_dict, _ = svc.parse_onu_info(output)
assert len(onu_dict) == 2
def test_more_inline_with_device_line(self):
"""More 标记与下一条设备数据同行时,不应丢弃该行"""
svc = _make_svc()
output = """
---- More ---- 1484-778f-aa60 Up OAM_Up loid H3C_ET704 500m
"""
onu_dict, _ = svc.parse_onu_info(output)
mac = "1484-778f-aa60"
assert mac in onu_dict
def test_empty_output(self):
svc = _make_svc()
onu_dict, unknown = svc.parse_onu_info("")
assert len(onu_dict) == 0
def test_header_only(self):
svc = _make_svc()
output = " Flags: S-Switched L-Loopback N-Not exist U-Up D-Down\n Port MAC Status"
onu_dict, _ = svc.parse_onu_info(output)
assert len(onu_dict) == 0
class TestCleanOutput:
def test_strips_ansi_codes(self):
svc = _make_svc()
cleaned = svc._clean_output("\x1b[37D\x1b[K 1484-778f-aa60 Up")
assert "\x1b[37D" not in cleaned
assert "\x1b[K" not in cleaned
assert "1484-778f-aa60" in cleaned
def test_removes_more_marker(self):
svc = _make_svc()
output = "---- More ----\n1484-778f-aa60 Up"
cleaned = svc._clean_output(output)
assert "---- More ----" not in cleaned
assert "1484-778f-aa60" in cleaned
def test_preserves_device_line_after_more(self):
"""More 标记同行后续设备数据不应被删除"""
svc = _make_svc()
output = "---- More ----\r\r 1484-778f-aa60 Up"
cleaned = svc._clean_output(output)
assert "1484-778f-aa60" in cleaned
assert "---- More ----" not in cleaned