H3ConuMS-v2/backend/tests/test_ssh_parser.py

"""SSH 输出解析测试"""
import re
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


class TestDetectLoopback:
    """环路检测输出解析测试"""

    def _parse(self, output: str):
        """模拟 detect_loopback 中的解析逻辑"""
        has_loop = "Loop is detected on following interfaces" in output
        interfaces = []
        if has_loop:
            for line in output.splitlines():
                m = re.match(r'\s+(Onu\S+)', line)
                if m:
                    interfaces.append(m.group(1))
        return has_loop, interfaces

    def test_no_loop(self):
        output = """
Loopback detection is enabled.
Loopback detection interval is 30 second(s).
No loopback is detected.
"""
        has_loop, interfaces = self._parse(output)
        assert not has_loop
        assert interfaces == []

    def test_has_loop_single(self):
        output = """
Loopback detection is enabled.
Loopback detection interval is 30 second(s).
Loop is detected on following interfaces:
        Onu1/0/1:1
"""
        has_loop, interfaces = self._parse(output)
        assert has_loop
        assert interfaces == ["Onu1/0/1:1"]

    def test_has_loop_multiple(self):
        output = """
Loop is detected on following interfaces:
        Onu1/0/1:1
        Onu1/0/2:3
        Onu2/0/5:10
"""
        has_loop, interfaces = self._parse(output)
        assert has_loop
        assert interfaces == ["Onu1/0/1:1", "Onu1/0/2:3", "Onu2/0/5:10"]

    def test_has_loop_with_extra_whitespace(self):
        """接口行有多余空白字符"""
        output = """
Loop is detected on following interfaces:
            Onu1/0/1:1
"""
        has_loop, interfaces = self._parse(output)
        assert has_loop
        assert interfaces == ["Onu1/0/1:1"]

    def test_no_false_positive_on_prompt(self):
        """确保设备提示符不被误识别为接口"""
        output = """
Loop is detected on following interfaces:
        Onu1/0/1:1
<H3C_Device>
"""
        has_loop, interfaces = self._parse(output)
        assert has_loop
        assert interfaces == ["Onu1/0/1:1"]