第322篇：自动化测试与网络验证
关键词
网络测试、自动化验证、pytest、pyATS、Genie、单元测试、集成测试、回归测试、TDD
一、为什么要做网络自动化测试
1.1 网络变更的风险
一次典型网络变更的"惊险一跃"：
变更前 → 变更中 → 变更后
┌──────────────────────────────────────────┐
│ 变更前（评审阶段） │
│ ┌─ 工程师 review 配置 │
│ ├─ 凭经验判断"应该没问题" │
│ ├─ 异常不在回滚就在变更 │
│ └─ 但回滚本身也有风险 │
│ │
│ 变更中（实施阶段） │
│ ┌─ "commit" 命令敲下去的瞬间 │
│ ├─ 心跳加速 │
│ └─ 祈祷不要出问题 │
│ │
│ 变更后（验证阶段） │
│ ┌─ 人工检查状态（show 命令逐个查） │
│ ├─ 可能漏查某个关键指标 │
│ ├─ 不良影响半小时后才被发现 │
│ └─ "昨晚我们改了啥？" │
└──────────────────────────────────────────┘
自动化测试的解决方案：
┌─ 变更前：自动验证配置语法和约束
├─ 变更中：自动化执行并实时验证
├─ 变更后：自动执行全量验证用例
└─ 回归：已有功能不因变更而破坏
1.2 测试金字塔
网络自动化测试金字塔：
| E2E 测试 ┌───────────┐ | ← 端到端业务验证（少而精） 集成测试 ┌───────┐ | 单元测试 最多 最快 最稳 | ← 多设备交互验证 | ← 单设备/单功能验证（多而快） | |
|---|---|---|---|---|---|
各层特点： ┌─ 单元测试：测试单一函数/模块/单设备配置 │ └─ 执行时间：毫秒-秒级 │ └─ 比例：占总测试 70% │ ├─ 集成测试：测试多设备交互（BGP 建联/OSPF 邻接） │ └─ 执行时间：秒-分钟级 │ └─ 比例：占总测试 20% │ └─ E2E 测试：测试端到端业务（用户→接入→核心→外网） └─ 执行时间：分钟级 └─ 比例：占总测试 10%
二、Python 测试框架
2.1 pytest 基础
#!/usr/bin/env python3
# test_network_basics.py — pytest 网络测试示例
import pytest
from netmiko import ConnectHandler
from typing import Dict, Any
# =============================================
# Fixture：共享设备连接
# =============================================
@pytest.fixture(scope="module")
def device_connection():
"""创建设备连接，测试结束后自动关闭"""
device = {
"device_type": "huawei_vrp",
"host": "192.168.1.1",
"username": "admin",
"password": "admin123",
}
conn = ConnectHandler(**device)
conn.enable()
yield conn
conn.disconnect()
# =============================================
# 单元测试：单条命令验证
# =============================================
def test_ping_gateway(device_connection):
"""测试到网关的可达性"""
output = device_connection.send_command(
"ping 192.168.1.254 -c 3"
)
# 成功率 100% 才算通过
assert "100% packet loss" not in output
assert "0.00% packet loss" not in output or \
"100%" not in output
def test_interface_status(device_connection):
"""测试关键接口状态"""
output = device_connection.send_command(
"display interface brief"
)
# 核心接口必须 UP
assert "GigabitEthernet0/0/1" in output
# 提取接口状态行
for line in output.splitlines():
if "GE0/0/1" in line:
assert "UP" in line.upper(), \
f"接口 GE0/0/1 状态异常: {line}"
def test_bgp_peer_status(device_connection):
"""测试 BGP peer 状态"""
output = device_connection.send_command(
"display bgp peer"
)
# 所有 BGP peer 必须 Established
assert "Established" in output
# 检查是否有非 Established 的 peer
assert "Idle" not in output
assert "Active" not in output
2.2 参数化测试
# =============================================
# 参数化测试：批量验证多个接口
# =============================================
INTERFACES_TO_CHECK = [
"GigabitEthernet0/0/1",
"GigabitEthernet0/0/2",
"GigabitEthernet0/0/3",
]
@pytest.mark.parametrize("interface", INTERFACES_TO_CHECK)
def test_interface_is_up(device_connection, interface):
"""参数化测试：每个接口都要 UP"""
output = device_connection.send_command(
f"display interface {interface}"
)
assert "current state : UP" in output or \
"line protocol is up" in output, \
f"接口 {interface} 未处于 UP 状态"
# 检查是否有 CRC 错误
for line in output.splitlines():
if "CRC" in line and "error" in line:
count = int(line.split()[-1])
assert count == 0, \
f"接口 {interface} 存在 CRC 错误: {count}"
# =============================================
# 自定义 Fixture：测试数据准备
# =============================================
@pytest.fixture
def test_vlan_data() -> Dict[str, Any]:
"""VLAN 测试数据"""
return {
"vlan_id": 100,
"vlan_name": "TEST_VLAN_100",
"interface": "GigabitEthernet0/0/10",
"ip": "10.100.100.1/24",
"expected_state": "UP",
}
def test_vlan_creation(device_connection, test_vlan_data):
"""测试 VLAN 创建后的状态"""
# 配置 VLAN
commands = [
f"vlan batch {test_vlan_data['vlan_id']}",
f"interface {test_vlan_data['interface']}",
f"port link-type access",
f"port default vlan {test_vlan_data['vlan_id']}",
]
output = device_connection.send_config_set(commands)
assert "Error" not in output
# 验证 VLAN 存在
output = device_connection.send_command(
f"display vlan {test_vlan_data['vlan_id']}"
)
assert str(test_vlan_data['vlan_id']) in output
三、pyATS / Genie 测试框架
3.1 框架概述
pyATS / Genie 是 Cisco 开源的网络测试框架：
pyATS（核心框架）：
┌──────────────────────────────────────────┐
│ 测试用例管理 │
│ ┌─ Testcase 定义和执行 │
│ ├─ 测试报告生成 │
│ ├─ 并行执行 │
│ └─ CI/CD 集成 │
└──────────────────────────────────────────┘
Genie（网络领域扩展）：
┌──────────────────────────────────────────┐
│ 网络解析与验证 │
│ ┌─ 设备输出解析（show → 结构化数据） │
│ ├─ 状态验证（operational state） │
│ ├─ 差分对比（diff state between runs） │
│ └─ 网络模型抽象 │
└──────────────────────────────────────────┘
pyATS + Genie 的优势：
┌─ 多厂商支持：Cisco/Huawei/Juniper/Arista
├─ 开箱即用的测试库
├─ YAML 测试用例定义（无需写代码）
├─ 自动解析 show 命令输出
└─ 社区丰富的测试用例模板
3.2 安装与基础使用
# 安装
# python -m pip install pyats genie
# =============================================
# pyATS 测试脚本示例
# =============================================
from pyats import aetest
from genie.testbed import load
import logging
logger = logging.getLogger(__name__)
class CommonSetup(aetest.CommonSetup):
"""通用测试初始化"""
@aetest.subsection
def connect_to_devices(self, testbed):
"""连接到测试床中的所有设备"""
for name, device in testbed.devices.items():
try:
device.connect()
logger.info(f"已连接设备: {name}")
except Exception as e:
logger.error(f"连接设备 {name} 失败: {e}")
self.failed(f"无法连接到 {name}")
class BGPTest(aetest.Testcase):
"""BGP 状态测试"""
@aetest.test
def test_bgp_summary(self, testbed):
"""验证所有设备的 BGP 状态"""
for name, device in testbed.devices.items():
bgp = device.parse("display bgp peer") # Genie 解析
# 遍历所有 BGP peer
for peer_ip, peer_info in bgp.get("peer", {}).items():
state = peer_info.get("state", "")
assert state == "Established", \
f"{name} 的 BGP peer {peer_ip} 状态: {state}"
# 检查 uptime
uptime = peer_info.get("up_time", "")
logger.info(
f"{name} → peer {peer_ip}: "
f"状态={state}, 运行时间={uptime}"
)
@aetest.test
def test_bgp_routes(self, testbed):
"""验证 BGP 路由表"""
for name, device in testbed.devices.items():
bgp_routes = device.parse("display bgp routing-table")
# 必须存在默认路由
default_route = bgp_routes.get("routes", {}).get("0.0.0.0/0")
assert default_route is not None, \
f"{name} 缺少默认 BGP 路由"
class InterfaceTest(aetest.Testcase):
"""接口状态测试"""
@aetest.test
def test_interfaces(self, testbed):
"""验证所有接口状态"""
for name, device in testbed.devices.items():
interfaces = device.parse("display interface brief")
for intf_name, intf_info in interfaces.items():
status = intf_info.get("oper_status", "")
if intf_name.startswith("GE"):
assert status.lower() == "up", \
f"{name} 接口 {intf_name} 状态: {status}"
class CommonCleanup(aetest.CommonCleanup):
"""测试清理"""
@aetest.subsection
def disconnect_devices(self, testbed):
"""断开所有设备连接"""
for name, device in testbed.devices.items():
device.disconnect()
# =============================================
# 主入口
# =============================================
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("--testbed", required=True,
help="测试床 YAML 文件")
args = parser.parse_args()
# 加载测试床
testbed = load(args.testbed)
# 运行测试
aetest.main(testbed=testbed)
3.3 测试床 YAML 定义
# testbed.yaml — pyATS 测试床定义
---
testbed:
name: production_network_testbed
credentials:
default:
username: admin
password: admin123
enable:
password: enable123
devices:
core-sw01:
os: iosxe
type: switch
connections:
cli:
protocol: ssh
ip: 192.168.1.1
port: 22
core-sw02:
os: iosxe
type: switch
connections:
cli:
protocol: ssh
ip: 192.168.1.2
port: 22
acc-sw01:
os: iosxe
type: switch
connections:
cli:
protocol: ssh
ip: 192.168.1.10
port: 22
topology:
# 拓扑关系定义
core-sw01:
interfaces:
GigabitEthernet0/0/1:
link: core-sw02:GigabitEthernet0/0/1
type: trunk
四、配置变更的 CI/CD 测试流水线
4.1 流水线阶段
配置变更 CI/CD 测试流水线：
┌──────────────────────────────────────────┐
│ 阶段 1：语法检查（Syntax Check） │
│ ┌─ Python 语法检查（flake8, mypy） │
│ ├─ YAML/JSON 格式校验 │
│ ├─ Jinja2 模板渲染测试 │
│ └─ CLI 命令语法预检（不支持？） │
│ │
│ 阶段 2：单元测试（Unit Test） │
│ ┌─ 渲染后的配置是否正确？ │
│ ├─ IP 地址是否冲突？ │
│ ├─ VLAN ID 是否在允许范围？ │
│ └─ 参数是否完整？ │
│ │
│ 阶段 3：集成测试（Integration Test） │
│ ┌─ 在沙箱/测试环境部署 │
│ ├─ OSPF 邻接是否建立？ │
│ ├─ BGP peer 是否 Established？ │
│ └─ 路由是否正常？ │
│ │
│ 阶段 4：合规检查（Compliance） │
│ ┌─ 是否符合安全基线？ │
│ ├─ 是否有冗余配置？ │
│ ├─ NTP/DNS/SNMP 是否配置正确？ │
│ └─ 密码复杂度是否达标？ │
│ │
│ 阶段 5：E2E 验证（End-to-End） │
│ ┌─ 端到端连通性测试 │
│ ├─ 业务流量是否正常？ │
│ ├─ 延迟/丢包指标是否达标？ │
│ └─ 回滚方案是否验证？ │
└──────────────────────────────────────────┘
4.2 GitLab CI 集成
# .gitlab-ci.yml — 网络测试 CI 流水线
stages:
- syntax-check
- unit-test
- integration-test
- compliance
- e2e-test
# =============================================
# 阶段 1：语法检查
# =============================================
syntax-check:
stage: syntax-check
image: python:3.10
script:
- python -m pip install pyyaml jsonschema
# YAML 格式检查
- python -c "
import yaml, sys
for f in ['config/devices.yml', 'config/network.yml']:
try:
with open(f) as fp:
yaml.safe_load(fp)
print(f'{f}: OK')
except Exception as e:
print(f'{f}: FAIL - {e}')
sys.exit(1)
"
# Jinja2 模板渲染测试
- python -m pip install jinja2
- python scripts/validate_templates.py
# =============================================
# 阶段 2：单元测试
# =============================================
unit-test:
stage: unit-test
image: python:3.10
script:
- python -m pip install pytest netmiko
- pytest tests/unit/ -v --junitxml=report-unit.xml
artifacts:
reports:
junit: report-unit.xml
# =============================================
# 阶段 3：集成测试（沙箱环境）
# =============================================
integration-test:
stage: integration-test
only:
- master
- develop
script:
- python -m pip install pyats genie
# 在沙箱环境部署配置
- ansible-playbook -i inventory/staging.yml deploy.yml
# 运行集成测试
- pyats run job tests/integration/bgp_test.py --testbed testbeds/staging.yaml
- pyats run job tests/integration/ospf_test.py --testbed testbeds/staging.yaml
artifacts:
paths:
- pyats-logs/
# =============================================
# 阶段 4：合规检查
# =============================================
compliance:
stage: compliance
script:
- python scripts/compliance_check.py
artifacts:
reports:
junit: report-compliance.xml
# =============================================
# 阶段 5：E2E 验证
# =============================================
e2e-test:
stage: e2e-test
only:
- master
when: manual # 手动触发
script:
- python -m pip install requests
- python tests/e2e/business_flow_test.py
- python tests/e2e/latency_test.py
五、测试数据 Mock 与仿真
5.1 设备模拟器
#!/usr/bin/env python3
# mock_device.py — 模拟设备响应用于测试
from unittest.mock import Mock, patch
import pytest
class MockDeviceConnection:
"""模拟设备连接，不依赖真实设备"""
def __init__(self, host: str = "192.168.1.1"):
self.host = host
self._responses = {}
def register_response(self, command: str, output: str):
"""注册命令的模拟返回"""
self._responses[command.strip()] = output
def send_command(self, command: str, **kwargs) -> str:
"""返回预先注册的模拟响应"""
cmd = command.strip()
# 注册默认响应
if cmd not in self._responses:
if "display interface brief" in cmd:
return """
Interface Status Protocol
GigabitEthernet0/0/1 up up
GigabitEthernet0/0/2 up up
GigabitEthernet0/0/3 down down
"""
elif "display bgp peer" in cmd:
return """
BGP Local router identifier: 10.0.0.1
Local AS number: 65001
Peer AS State UpTime
10.0.0.2 65002 Established 01:23:45
10.0.0.3 65003 Established 02:34:56
"""
elif "display version" in cmd:
return "Huawei Versatile Routing Platform Software"
return self._responses.get(cmd, "")
def send_config_set(self, commands: list, **kwargs) -> str:
"""模拟配置下发"""
results = []
for cmd in commands:
if "Error" in cmd:
results.append(f"Error: invalid command - {cmd}")
else:
results.append(f"[OK] {cmd}")
return "\n".join(results)
def disconnect(self):
pass
# =============================================
# 使用 Mock 进行测试
# =============================================
@pytest.fixture
def mock_device():
"""创建模拟设备"""
dev = MockDeviceConnection()
return dev
def test_interface_check_with_mock(mock_device):
"""使用模拟设备测试接口检查"""
from netmiko import ConnectHandler
# 用 Mock 替换真实的 send_command
with patch.object(ConnectHandler, 'send_command',
side_effect=mock_device.send_command):
conn = ConnectHandler(
device_type="huawei_vrp",
host="192.168.1.1",
username="admin",
password="admin",
)
# 验证接口状态
output = conn.send_command("display interface brief")
assert "GigabitEthernet0/0/1" in output
assert "up" in output
def test_config_error_handling(mock_device):
"""测试配置错误处理"""
with patch.object(ConnectHandler, 'send_config_set',
side_effect=mock_device.send_config_set):
conn = ConnectHandler(
device_type="huawei_vrp",
host="192.168.1.1",
username="admin",
password="admin",
)
# 正常配置
result = conn.send_config_set(["vlan 100"])
assert "Error" not in result
# 错误配置
result = conn.send_config_set(["Error: bad command"])
assert "Error" in result
六、最佳实践总结
网络自动化测试最佳实践：
1. 测试先行（Test-Driven Network）
┌─ 先写测试，再写配置
├─ 测试覆盖所有"不能出错"的场景
└─ 变更前跑测试确保基线正常
2. 分层测试
┌─ 70% 单元测试（快速、可靠）
├─ 20% 集成测试（多设备交互）
└─ 10% E2E 测试（关键业务）
3. 自动化集成
┌─ 所有测试集成到 CI/CD 流水线
├─ Git Push 触发自动化测试
└─ 测试不通过禁止合入主分支
4. Mock 外部依赖
┌─ 单元测试不依赖真实设备
├─ 使用 MockDevice 模拟设备响应
└─ 集成测试才使用沙箱环境
5. 持续改进
┌─ 每次故障"复盘"后补充测试
├─ 测试覆盖率持续提升
└─ 定期回顾测试质量
最终目标：
┌──────────────────────────────────────────┐
│ 让网络变更从"心惊胆战" │
│ 变成"信心满满地点击 Merge" │
└──────────────────────────────────────────┘
下篇预告：第323篇 — 网络配置合规性检查自动化，介绍如何自动化检查设备配置是否合规，防范配置漂移风险。
下篇预告：第323篇 — 网络配置合规性检查自动化，介绍如何自动化检查设备配置是否合规，防范配置漂移风险。