107 lines
3.4 KiB
Python
107 lines
3.4 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 数据"
|