b2c20ec43d
- 新增微信告警服务(wechat_service)和告警任务(alert_tasks) - 新增 WebSocket 实时推送端点 - 新增监控管理模块(monitor) - 增强统计仪表板:趋势图、区域分布、光功率历史 - 设备管理:添加坐标信息、标签系统、复合索引优化 - 前端:重构Dashboard/Charts页面,新增业务组件 - 新增5个数据库迁移(坐标、复合索引、标签、光功率历史、显示名) - 更新部署配置和脚本 - 新增测试框架基础结构
94 lines
3.4 KiB
Python
94 lines
3.4 KiB
Python
"""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
|