287 lines
10 KiB
Python
287 lines
10 KiB
Python
from __future__ import annotations
|
||
|
||
"""
|
||
与真实 H3C 设备联调的“活体”测试用例。
|
||
|
||
说明：
|
||
- 连接参数参考 exp/yangcli/run_yangcli.sh：
|
||
- server=127.0.0.1
|
||
- ncport=8830
|
||
- user=netconf_user
|
||
- password='...'
|
||
- 为避免在仓库中硬编码明文密码，本测试通过环境变量注入密码：
|
||
- H3C_NETCONF_PASSWORD
|
||
- 可选：H3C_NETCONF_HOST / H3C_NETCONF_PORT / H3C_NETCONF_USER
|
||
|
||
默认行为：
|
||
- 若未设置 H3C_NETCONF_PASSWORD，或无法建立到指定 host:port 的 TCP 连接，
|
||
则使用 pytest.skip() 自动跳过，不影响普通单元测试/CI。
|
||
- 仅在本地联调时、显式设置上述环境变量后，此测试才会真正访问设备。
|
||
"""
|
||
|
||
import os
|
||
import re
|
||
import socket
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from ncclient import manager
|
||
|
||
from exporter.netconf_client import build_transceiver_filter, parse_netconf_response
|
||
|
||
|
||
H3C_HOST = os.getenv("H3C_NETCONF_HOST", "127.0.0.1")
|
||
H3C_PORT = int(os.getenv("H3C_NETCONF_PORT", "8830"))
|
||
H3C_USER = os.getenv("H3C_NETCONF_USER", "netconf_user")
|
||
|
||
|
||
def _load_password_from_script() -> str:
|
||
"""
|
||
尝试从 exp/yangcli 下的脚本中解析 --password='...'.
|
||
|
||
这样可以重用你已经验证过的 yangcli 参数，而不在测试里硬编码密码。
|
||
若脚本不存在或未找到 password，则返回空字符串。
|
||
"""
|
||
root = Path(__file__).resolve().parents[1]
|
||
candidates = [
|
||
root / "exp" / "yangcli" / "run_yangcli_h3c.sh",
|
||
root / "exp" / "yangcli" / "run_yangcli.sh",
|
||
]
|
||
pattern = re.compile(r"--password='([^']*)'")
|
||
for path in candidates:
|
||
if not path.exists():
|
||
continue
|
||
text = path.read_text(encoding="utf-8")
|
||
match = pattern.search(text)
|
||
if match:
|
||
return match.group(1)
|
||
return ""
|
||
|
||
|
||
H3C_PASSWORD = os.getenv("H3C_NETCONF_PASSWORD") or _load_password_from_script()
|
||
|
||
|
||
def _can_connect(host: str, port: int, timeout: float = 2.0) -> bool:
|
||
"""快速探测 host:port 是否可连，用于决定是否跳过 live 测试。"""
|
||
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||
try:
|
||
sock.settimeout(timeout)
|
||
sock.connect((host, port))
|
||
return True
|
||
except OSError:
|
||
return False
|
||
finally:
|
||
sock.close()
|
||
|
||
|
||
@pytest.mark.h3c_live
|
||
def test_h3c_live_transceiver_rpc_and_parse() -> None:
|
||
"""
|
||
使用真实 H3C 设备验证：
|
||
- ncclient 能与设备建立 NETCONF 会话；
|
||
- build_transceiver_filter() 构造的 subtree filter 在设备上可用；
|
||
- parse_netconf_response() 能正确解析设备返回的 XML，并得到非空结果。
|
||
"""
|
||
if not _can_connect(H3C_HOST, H3C_PORT):
|
||
pytest.skip(f"H3C NETCONF {H3C_HOST}:{H3C_PORT} 不可达，跳过 live 测试")
|
||
|
||
flt = build_transceiver_filter()
|
||
|
||
with manager.connect(
|
||
host=H3C_HOST,
|
||
port=H3C_PORT,
|
||
username=H3C_USER,
|
||
password=H3C_PASSWORD,
|
||
hostkey_verify=False,
|
||
timeout=30,
|
||
allow_agent=False,
|
||
look_for_keys=False,
|
||
) as m:
|
||
reply = m.get(filter=("subtree", flt))
|
||
xml_str = str(reply)
|
||
|
||
transceivers, channels = parse_netconf_response(xml_str, device_name=f"h3c-{H3C_HOST}")
|
||
|
||
# 只要返回非空结果，就说明"连接 + filter + 解析"在真实设备上可以工作
|
||
assert transceivers or channels, "H3C 设备未返回任何 transceiver/channel 数据"
|
||
|
||
|
||
@pytest.mark.h3c_live
|
||
def test_h3c_config_description_consistency() -> None:
|
||
"""
|
||
验证 H3C 设备上所有 transceiver 的 physical channel 的 config/description 是否正确。
|
||
|
||
检查规则：
|
||
- component 名称格式如: "67.TwoHundredGigE1/0/2:1"
|
||
- 从中提取物理端口: "1/0/2"
|
||
- channel config/description 应该是: "1/0/2:1", "1/0/2:2", 等
|
||
- 即：config/description 的端口部分应与 component 名称中的端口部分一致
|
||
|
||
目的：发现 H3C 设备上 config/description 配置错误的情况
|
||
"""
|
||
if not _can_connect(H3C_HOST, H3C_PORT):
|
||
pytest.skip(f"H3C NETCONF {H3C_HOST}:{H3C_PORT} 不可达，跳过 live 测试")
|
||
|
||
flt = build_transceiver_filter()
|
||
|
||
with manager.connect(
|
||
host=H3C_HOST,
|
||
port=H3C_PORT,
|
||
username=H3C_USER,
|
||
password=H3C_PASSWORD,
|
||
hostkey_verify=False,
|
||
timeout=30,
|
||
allow_agent=False,
|
||
look_for_keys=False,
|
||
) as m:
|
||
reply = m.get(filter=("subtree", flt))
|
||
xml_str = str(reply)
|
||
|
||
# 解析 XML 提取详细的 config 和 state 信息
|
||
import xml.etree.ElementTree as ET
|
||
|
||
NS = {
|
||
"oc-platform": "http://openconfig.net/yang/platform",
|
||
"oc-transceiver": "http://openconfig.net/yang/platform/transceiver",
|
||
}
|
||
|
||
root = ET.fromstring(xml_str)
|
||
components_path = ".//oc-platform:components/oc-platform:component"
|
||
|
||
inconsistencies = []
|
||
total_channels_checked = 0
|
||
|
||
for comp in root.findall(components_path, NS):
|
||
# 获取 component 名称
|
||
name_elem = comp.find("oc-platform:name", NS)
|
||
if name_elem is None or not name_elem.text:
|
||
continue
|
||
component_name = name_elem.text.strip()
|
||
|
||
# 只检查 TRANSCEIVER 类型
|
||
type_elem = comp.find("oc-platform:state/oc-platform:type", NS)
|
||
if type_elem is None or "TRANSCEIVER" not in type_elem.text:
|
||
continue
|
||
|
||
# 从 component 名称中提取端口
|
||
# 格式: "67.TwoHundredGigE1/0/2:1" -> 提取 "1/0/2"
|
||
expected_port = _extract_port_from_component_name(component_name)
|
||
if not expected_port:
|
||
# 无法从 component 名称中提取端口，跳过
|
||
continue
|
||
|
||
# 遍历所有 physical channels
|
||
channels_path = (
|
||
"oc-transceiver:transceiver/oc-transceiver:physical-channels/"
|
||
"oc-transceiver:channel"
|
||
)
|
||
for ch in comp.findall(channels_path, NS):
|
||
total_channels_checked += 1
|
||
|
||
# 获取 channel index
|
||
idx_elem = ch.find("oc-transceiver:index", NS)
|
||
channel_index = idx_elem.text if idx_elem is not None else "?"
|
||
|
||
# 获取 config/description
|
||
config_desc_elem = ch.find("oc-transceiver:config/oc-transceiver:description", NS)
|
||
config_desc = config_desc_elem.text.strip() if config_desc_elem is not None and config_desc_elem.text else None
|
||
|
||
# 获取 state/description
|
||
state_desc_elem = ch.find("oc-transceiver:state/oc-transceiver:description", NS)
|
||
state_desc = state_desc_elem.text.strip() if state_desc_elem is not None and state_desc_elem.text else None
|
||
|
||
# 检查 config/description
|
||
if config_desc:
|
||
# config/description 应该是 "端口:通道号" 格式，如 "1/0/2:1"
|
||
if ":" in config_desc:
|
||
config_port = config_desc.split(":", 1)[0]
|
||
if config_port != expected_port:
|
||
inconsistencies.append({
|
||
"component": component_name,
|
||
"channel_index": channel_index,
|
||
"expected_port": expected_port,
|
||
"config_desc": config_desc,
|
||
"config_port": config_port,
|
||
"issue": "config/description 端口部分与 component 名称不一致"
|
||
})
|
||
else:
|
||
# config/description 缺失
|
||
inconsistencies.append({
|
||
"component": component_name,
|
||
"channel_index": channel_index,
|
||
"expected_port": expected_port,
|
||
"config_desc": None,
|
||
"issue": "config/description 缺失"
|
||
})
|
||
|
||
# 同时检查 state/description（仅作信息收集，不作为测试失败条件）
|
||
# 但我们会记录这些差异，以便了解 state 数据质量
|
||
pass # state 检查移至单独的报告
|
||
|
||
# 生成报告
|
||
print(f"\n=== H3C Config/Description 一致性检查报告 ===")
|
||
print(f"检查的 channel 总数: {total_channels_checked}")
|
||
print(f"发现不一致的 channel 数: {len(inconsistencies)}")
|
||
|
||
if inconsistencies:
|
||
print("\n发现的不一致情况:")
|
||
for item in inconsistencies:
|
||
print(f" Component: {item['component']}")
|
||
print(f" Channel Index: {item['channel_index']}")
|
||
print(f" 期望端口: {item['expected_port']}")
|
||
print(f" 实际 config/description: {item.get('config_desc', 'N/A')}")
|
||
if 'config_port' in item:
|
||
print(f" config 中的端口: {item['config_port']}")
|
||
print(f" 问题: {item['issue']}")
|
||
print()
|
||
|
||
# 测试失败，config/description 应该是准确的
|
||
pytest.fail(
|
||
f"发现 {len(inconsistencies)} 个 channel 的 config/description 与 component 名称不一致，"
|
||
"详见上方输出"
|
||
)
|
||
else:
|
||
print("\n✅ 所有 channel 的 config/description 都与 component 名称一致")
|
||
|
||
|
||
def _extract_port_from_component_name(component_name: str) -> str | None:
|
||
"""
|
||
从 component 名称中提取物理端口。
|
||
|
||
例如:
|
||
- "67.TwoHundredGigE1/0/2:1" -> "1/0/2"
|
||
- "63.TwoHundredGigE1/0/1:1" -> "1/0/1"
|
||
- "323.FourHundredGigE1/0/66" -> "1/0/66"
|
||
|
||
返回: 端口字符串，如 "1/0/2"，若无法提取则返回 None
|
||
"""
|
||
import re
|
||
|
||
# 匹配 "数字/数字/数字" 格式的端口
|
||
# 可能在 GigE 或 FourHundredGigE 等后面，可能带或不带 ":通道号"
|
||
pattern = r"(\d+/\d+/\d+)"
|
||
match = re.search(pattern, component_name)
|
||
if match:
|
||
return match.group(1)
|
||
return None
|
||
|
||
|
||
def test_extract_port_from_component_name():
|
||
"""测试从 component 名称中提取端口的辅助函数（不依赖真实设备）"""
|
||
# 测试各种可能的命名格式
|
||
test_cases = [
|
||
("67.TwoHundredGigE1/0/2:1", "1/0/2"),
|
||
("63.TwoHundredGigE1/0/1:1", "1/0/1"),
|
||
("323.FourHundredGigE1/0/66", "1/0/66"),
|
||
("64.TwoHundredGigE10/20/30:2", "10/20/30"),
|
||
("SomePrefix99/88/77", "99/88/77"),
|
||
("no_port_here", None),
|
||
("", None),
|
||
]
|
||
for component_name, expected_port in test_cases:
|
||
result = _extract_port_from_component_name(component_name)
|
||
assert result == expected_port, (
|
||
f"从 '{component_name}' 提取端口失败: "
|
||
f"期望 '{expected_port}', 实际 '{result}'"
|
||
)
|