第326篇：自动化测试：用 pytest 验证网络状态
关键词：pytest、网络测试、合同测试（Contract Testing）、状态验证、接口测试、路由测试、ACL 测试
一、引言
在传统运维模式中，配置变更后是否"正确生效"，往往依赖工程师手工登录设备执行 display 命令逐一核对。这种方式不仅效率低下，而且容易因疲劳或疏忽遗漏检查项。
pytest 是 Python 生态中最流行的测试框架。将 pytest 引入网络运维，意味着网络配置的状态验证可以像软件代码测试一样：编写测试用例 → 自动化执行 → 生成报告 → 持续集成中阻断不合格变更。
二、pytest 网络测试框架搭建
2.1 环境配置
# 安装依赖
pip install pytest pytest-html pytest-metadata
pip install pyntc napalm netmiko
# 项目测试目录结构
tests/
- conftest.py # 全局 Fixtures
- test_interfaces.py # 接口状态测试
- test_bgp.py # BGP 协议测试
- test_ospf.py # OSPF 协议测试
- test_acls.py # ACL 策略测试
- test_connectivity.py # 连通性测试
- test_routing.py # 路由表测试
- fixtures/
- devices.yaml # 设备连接信息
- expected_states.yaml # 期望状态定义
2.2 conftest.py 全局 Fixtures
# tests/conftest.py
import pytest
import yaml
import napalm
from pyntc import ntc_helpers
@pytest.fixture(scope='session')
def device_list():
"""加载设备清单"""
with open('tests/fixtures/devices.yaml') as f:
return yaml.safe_load(f)
@pytest.fixture(scope='session')
def expected_states():
"""加载期望状态"""
with open('tests/fixtures/expected_states.yaml') as f:
return yaml.safe_load(f)
@pytest.fixture
def device_connection(request, device_list):
"""
创建设备连接 Fixture
用法: def test_xxx(device_connection): ...
参数: request.param 指定设备名
"""
dev_name = request.param
dev_info = next(d for d in device_list['devices']
if d['name'] == dev_name)
driver = napalm.get_network_driver(dev_info['os'])
device = driver(
dev_info['hostname'],
dev_info['username'],
dev_info['password'],
optional_args={'timeout': 60}
)
device.open()
yield device
device.close()
# 参数化：所有 PE 设备
@pytest.fixture(params=['PE-BJ-01', 'PE-BJ-02', 'PE-SH-01'])
def pe_device(request, device_connection):
"""PE 设备 Fixture"""
return device_connection
2.3 期望状态定义
# tests/fixtures/expected_states.yaml
interfaces:
PE-BJ-01:
GigabitEthernet1/0/0:
status: up
description: "UPLINK_TO_PE-BJ-02"
ip_address: "10.1.1.1/30"
mtu: 9216
GigabitEthernet1/0/1:
status: up
description: "UPLINK_TO_P-BJ-01"
ip_address: "10.1.1.5/30"
mtu: 9216
PE-BJ-02:
GigabitEthernet1/0/0:
status: up
description: "UPLINK_TO_PE-BJ-01"
ip_address: "10.1.1.2/30"
mtu: 9216
bgp:
PE-BJ-01:
as_number: 65001
router_id: "10.100.1.1"
peers:
- ip: "10.1.1.2"
remote_as: 65002
state: "Established"
af: "ipv4 unicast"
keepalive: 10
holdtime: 30
acl:
PE-BJ-01:
advanced_acl_3001:
rules:
- "permit tcp any 192.168.1.0 0.0.0.255 eq 443"
- "deny ip any any"
三、核心测试用例编写
3.1 接口状态测试
# tests/test_interfaces.py
import pytest
def test_interface_status(pe_device, expected_states):
"""验证接口 Up/Down 状态"""
dev_name = pe_device._name # 从 fixture 获取设备名
interfaces = pe_device.get_interfaces()
expected = expected_states['interfaces'].get(dev_name, {})
for intf_name, expected_attrs in expected.items():
if intf_name in interfaces:
actual = interfaces[intf_name]
assert actual['is_up'] == (expected_attrs.get('status') == 'up'), \
f"接口 {intf_name} 状态异常: 期望 {expected_attrs.get('status')}, " \
f"实际 {'up' if actual['is_up'] else 'down'}"
else:
pytest.fail(f"接口 {intf_name} 不存在于设备 {dev_name}")
def test_interface_description(pe_device, expected_states):
"""验证接口描述（用于核对连接关系）"""
dev_name = pe_device._name
interfaces = pe_device.get_interfaces()
expected = expected_states['interfaces'].get(dev_name, {})
for intf_name, expected_attrs in expected.items():
if 'description' in expected_attrs:
actual_desc = interfaces[intf_name].get('description', '')
assert expected_attrs['description'].lower() in actual_desc.lower(), \
f"接口 {intf_name} 描述不匹配: " \
f"期望包含 '{expected_attrs['description']}', " \
f"实际 '{actual_desc}'"
def test_interface_mtu(pe_device, expected_states):
"""验证接口 MTU"""
dev_name = pe_device._name
interfaces = pe_device.get_interfaces()
expected = expected_states['interfaces'].get(dev_name, {})
for intf_name, expected_attrs in expected.items():
if 'mtu' in expected_attrs:
actual_mtu = interfaces[intf_name].get('mtu', 1500)
assert actual_mtu == expected_attrs['mtu'], \
f"接口 {intf_name} MTU 不匹配: " \
f"期望 {expected_attrs['mtu']}, 实际 {actual_mtu}"
def test_interface_ip_address(pe_device, expected_states):
"""验证接口 IP 地址"""
dev_name = pe_device._name
facts = pe_device.get_interfaces_ip()
expected = expected_states['interfaces'].get(dev_name, {})
for intf_name, expected_attrs in expected.items():
if 'ip_address' in expected_attrs:
ip_network = expected_attrs['ip_address']
# 解析期望 IP 和掩码
if '/' in ip_network:
ip, mask = ip_network.split('/')
else:
ip = ip_network
mask = '24'
assert ip in str(facts.get(intf_name, {}).get('ipv4', {})), \
f"接口 {intf_name} IP 地址不匹配"
3.2 BGP 协议测试
# tests/test_bgp.py
import pytest
def test_bgp_as_number(pe_device, expected_states):
"""验证 BGP AS 号"""
dev_name = pe_device._name
bgp_config = pe_device.get_bgp_config()
expected = expected_states['bgp'].get(dev_name, {})
actual_as = bgp_config.get('global', {}).get('as', 0)
expected_as = expected.get('as_number', 0)
assert actual_as == expected_as, \
f"BGP AS 不匹配: 期望 {expected_as}, 实际 {actual_as}"
def test_bgp_router_id(pe_device, expected_states):
"""验证 BGP Router ID"""
dev_name = pe_device._name
bgp_config = pe_device.get_bgp_config()
expected = expected_states['bgp'].get(dev_name, {})
actual_rid = bgp_config.get('global', {}).get('router_id', '')
expected_rid = expected.get('router_id', '')
assert actual_rid == expected_rid, \
f"BGP Router ID 不匹配: 期望 {expected_rid}, 实际 {actual_rid}"
def test_bgp_peer_state(pe_device, expected_states):
"""验证 BGP Peer 会话状态"""
dev_name = pe_device._name
bgp_fact = pe_device.get_bgp_facts()
expected = expected_states['bgp'].get(dev_name, {})
for peer_cfg in expected.get('peers', []):
peer_ip = peer_cfg['ip']
expected_state = peer_cfg['state']
# 在 BGP facts 中查找对应 peer
peer_found = False
for peer_key, peer_info in bgp_fact.get('peers', {}).items():
if peer_ip in str(peer_key):
peer_found = True
actual_state = peer_info.get('state', 'Idle')
assert actual_state == expected_state, \
f"BGP Peer {peer_ip} 状态异常: " \
f"期望 {expected_state}, 实际 {actual_state}"
break
assert peer_found, f"BGP Peer {peer_ip} 未在运行状态中找到"
def test_bgp_peer_count(pe_device, expected_states):
"""验证 BGP Peer 数量"""
dev_name = pe_device._name
bgp_fact = pe_device.get_bgp_facts()
expected = expected_states['bgp'].get(dev_name, {})
expected_count = len(expected.get('peers', []))
established_count = sum(
1 for p in bgp_fact.get('peers', {}).values()
if p.get('state') == 'Established'
)
assert established_count == expected_count, \
f"BGP Established Peer 数量不匹配: " \
f"期望 {expected_count}, 实际 {established_count}"
def test_bgp_route_count(pe_device):
"""验证 BGP 路由表条目数不为 0"""
bgp_fact = pe_device.get_bgp_facts()
total_routes = bgp_fact.get('routes', {}).get('total', 0)
assert total_routes > 0, \
"BGP 路由表为空，可能存在路由泄露或配置错误"
3.3 ACL 策略测试
# tests/test_acls.py
import pytest
def test_acl_rules_exist(device_connection, expected_states):
"""验证 ACL 规则已应用"""
dev_name = device_connection._name
acls = device_connection.get_acls()
expected_acls = expected_states['acl'].get(dev_name, {})
for acl_name in expected_acls:
assert acl_name in acls, \
f"ACL {acl_name} 未找到"
actual_rules = acls[acl_name].get('rules', [])
expected_rules = expected_acls[acl_name].get('rules', [])
for exp_rule in expected_rules:
found = any(exp_rule.lower() in str(rule).lower()
for rule in actual_rules)
assert found, \
f"ACL {acl_name} 中未找到规则: {exp_rule}"
def test_default_deny_applied(device_connection, expected_states):
"""验证默认拒绝策略已生效"""
dev_name = device_connection._name
acls = device_connection.get_acls()
expected_acls = expected_states['acl'].get(dev_name, {})
for acl_name, acl_cfg in expected_acls.items():
rules = acl_cfg.get('rules', [])
# 检查最后一条规则是否为 deny
if rules and 'deny' in rules[-1].lower():
actual_rules = acls[acl_name].get('rules', [])
assert actual_rules, \
f"ACL {acl_name} 规则列表为空"
def test_acl_applied_to_interface(pe_device, expected_states):
"""验证 ACL 已应用到接口"""
dev_name = pe_device._name
interfaces = pe_device.get_interfaces()
# 检查入方向 ACL 应用
for intf_name, intf_info in interfaces.items():
applied_acls = intf_info.get('ipv4', {}).get('acl', '')
# 根据业务规则验证特定接口是否绑定了 ACL
if 'UPLINK' in intf_info.get('description', '').upper():
assert applied_acls, \
f"上联接口 {intf_name} 未绑定 ACL"
3.4 连通性测试
# tests/test_connectivity.py
import pytest
import subprocess
def test_ping_reachability(pe_device):
"""从设备执行 Ping 测试"""
# 测试到核心路由器的连通性
targets = ['10.1.1.3', '10.1.1.4', '10.1.1.5']
for target in targets:
result = subprocess.run(
['ping', '-c', '3', '-W', '2', target],
capture_output=True, text=True
)
assert result.returncode == 0, \
f"Ping {target} 失败: {result.stderr}"
def test_traceroute_path(pe_device):
"""验证 Traceroute 路径符合预期"""
# 通过 SSH 执行 traceroute
result = pe_device.cli(['traceroute 10.100.0.1 max-hops 10'])
hops = result.strip().split('\n')
# 验证路径不经过非预期的跳
assert len(hops) <= 5, \
"Traceroute 路径过长，可能存在次优路由"
四、测试执行与报告
4.1 执行命令
# 执行全部测试
pytest tests/ -v --html=reports/test_report.html
# 执行特定测试文件
pytest tests/test_bgp.py -v
# 执行特定设备测试
pytest tests/test_interfaces.py -v -k "PE-BJ-01"
# 仅执行失败的测试（调试用）
pytest tests/ --lf
# 带详细输出
pytest tests/ -v -s --tb=long
4.2 CI/CD 集成
# .gitlab-ci.yml 中的测试阶段
network_test:
stage: integration-test
image: python:3.11-slim
script:
- pip install -r requirements-test.txt
- pytest tests/ -v
--html=reports/test_report.html
--junitxml=reports/junit.xml
artifacts:
when: always
reports:
junit: reports/junit.xml
paths:
- reports/
expire_in: 30 days
rules:
- if: $CI_PIPELINE_SOURCE == "merge_request_event"
4.3 测试输出示例
=================================== test session starts ===================================
platform linux -- Python 3.11.6, pytest-7.4.3
plugins: html-3.2.0, metadata-3.0.0
collected 24 items
tests/test_interfaces.py::test_interface_status[PE-BJ-01] PASSED [ 4%]
tests/test_interfaces.py::test_interface_status[PE-BJ-02] PASSED [ 8%]
tests/test_interfaces.py::test_interface_status[PE-SH-01] PASSED [ 12%]
tests/test_interfaces.py::test_interface_description[PE-BJ-01] PASSED [ 16%]
tests/test_interfaces.py::test_interface_description[PE-BJ-02] PASSED [ 20%]
tests/test_bgp.py::test_bgp_as_number[PE-BJ-01] PASSED [ 25%]
tests/test_bgp.py::test_bgp_peer_state[PE-BJ-01] FAILED [ 29%]
tests/test_bgp.py::test_bgp_peer_count[PE-BJ-01] FAILED [ 33%]
tests/test_acls.py::test_acl_rules_exist[PE-BJ-01] PASSED [ 37%]
==================================== FAILURES =============================================
____________________ test_bgp_peer_state[PE-BJ-01] ____________________
E AssertionError: BGP Peer 10.1.1.2 状态异常: 期望 Established, 实际 Idle
tests/test_bgp.py:45: AssertionError
=========================== short test summary info ===========================
FAILED tests/test_bgp.py::test_bgp_peer_state[PE-BJ-01] - AssertionError...
FAILED tests/test_bgp.py::test_bgp_peer_count[PE-BJ-01] - AssertionError...
========================= 2 failed, 22 passed in 45.32s ==========================
五、Network Tools 工具集：SNMP采集器/配置备份/批量执行
5.1 工具集架构设计
5.2 SNMP 设备指标采集器
# tools/snmp_collector.py
"""
SNMP 批量采集器
支持：CPU/内存利用率、接口流量/错误计数、温度/风扇状态
协议：SNMPv2c / SNMPv3（AES-128 认证 + 加密）
"""
import pysnmp.hlapi as snmp
from concurrent.futures import ThreadPoolExecutor, as_completed
import logging
from dataclasses import dataclass, field, asdict
from typing import List, Dict, Optional
from datetime import datetime
logger = logging.getLogger(__name__)
# 常用 OID 定义
SNMP_OIDS = {
'system_uptime': '1.3.6.1.2.1.1.3.0',
'hostname': '1.3.6.1.2.1.1.5.0',
# 华为 VRP CPU/内存
'cpu_usage': '1.3.6.1.4.1.2011.5.25.31.1.1.1.1.1',
'memory_usage': '1.3.6.1.4.1.2011.5.25.31.1.1.1.1.2',
# 标准接口表
'if_table_entries': '1.3.6.1.2.1.2.2.1.2',
'if_in_octets': '1.3.6.1.2.1.2.2.1.10',
'if_out_octets': '1.3.6.1.2.1.2.2.1.16',
'if_in_errors': '1.3.6.1.2.1.2.2.1.14',
'if_out_errors': '1.3.6.1.2.1.2.2.1.20',
'if_admin_status': '1.3.6.1.2.1.2.2.1.7',
'if_oper_status': '1.3.6.1.2.1.2.2.1.8',
}
@dataclass
class DeviceMetrics:
"""设备指标数据模型"""
device_name: str
ip_address: str
timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
cpu_usage: Optional[float] = None
memory_usage: Optional[float] = None
interfaces: List[Dict] = field(default_factory=list)
errors: List[str] = field(default_factory=list)
class SNMPCollector:
"""SNMP 批量采集器"""
def __init__(self, community='public', version=2, timeout=5, retries=2):
self.community = community
self.version = version
self.timeout = timeout
self.retries = retries
def _build_snmp_version(self):
if self.version == 3:
from pysnmp.hlapi import UsmUserData
return UsmUserData('admin', 'authkey', 'privkey',
authProtocol='usmHMACSHAAuthProtocol',
privProtocol='usmAesCBCCipher')
return snmp.CommunityData(self.community)
def collect_single(self, host: str, device_name: str) -> DeviceMetrics:
"""采集单台设备的指标"""
metrics = DeviceMetrics(device_name=device_name, ip_address=host)
creds = self._build_snmp_version()
try:
# 采集标量 OID
for metric_name, oid in SNMP_OIDS.items():
if metric_name in ('if_table_entries', 'if_in_octets',
'if_out_octets', 'if_in_errors',
'if_out_errors', 'if_admin_status',
'if_oper_status'):
continue # 表格OID单独处理
iterator = snmp.getCmd(
snmp.SnmpEngine(),
creds,
snmp.UdpTransportTarget((host, 161),
timeout=self.timeout,
retries=self.retries),
snmp.ObjectType(snmp.ObjectIdentifier(oid))
)
error_indication, error_status, error_index, var_binds = next(iterator)
if error_indication:
metrics.errors.append(f"SNMP Error: {error_indication}")
elif var_binds:
value = var_binds[0][1]
if metric_name == 'cpu_usage':
metrics.cpu_usage = float(value) / 10.0 # 转换为百分比
elif metric_name == 'memory_usage':
metrics.memory_usage = float(value) / 10.0
# 采集接口表格数据
interfaces = self._collect_interfaces(host, creds)
metrics.interfaces = interfaces
except Exception as e:
metrics.errors.append(f"采集异常: {str(e)}")
logger.error(f"设备 {host} 采集失败: {e}")
return metrics
def _collect_interfaces(self, host: str, creds) -> List[Dict]:
"""采集接口表格数据"""
interfaces = []
# 遍历接口索引（假设最多128个接口）
for idx in range(1, 129):
name_oid = f"1.3.6.1.2.1.2.2.1.2.{idx}"
iterator = snmp.getCmd(
snmp.SnmpEngine(),
creds,
snmp.UdpTransportTarget((host, 161),
timeout=self.timeout,
retries=self.retries),
snmp.ObjectType(snmp.ObjectIdentifier(name_oid))
)
_, _, _, var_binds = next(iterator)
if not var_binds:
break # 到达接口表末尾
iface_name = str(var_binds[0][1])
in_octets = self._get_oid(host, creds, f"1.3.6.1.2.1.2.2.1.10.{idx}")
out_octets = self._get_oid(host, creds, f"1.3.6.1.2.1.2.2.1.16.{idx}")
in_errors = self._get_oid(host, creds, f"1.3.6.1.2.1.2.2.1.14.{idx}")
oper_status = self._get_oid(host, creds, f"1.3.6.1.2.1.2.2.1.8.{idx}")
interfaces.append({
'name': iface_name,
'index': idx,
'in_octets': int(in_octets) if in_octets else 0,
'out_octets': int(out_octets) if out_octets else 0,
'in_errors': int(in_errors) if in_errors else 0,
'status': 'up' if oper_status == '1' else 'down'
})
return interfaces
def _get_oid(self, host: str, creds, oid: str) -> Optional[str]:
"""获取单个 OID 值"""
try:
iterator = snmp.getCmd(
snmp.SnmpEngine(),
creds,
snmp.UdpTransportTarget((host, 161),
timeout=self.timeout,
retries=self.retries),
snmp.ObjectType(snmp.ObjectIdentifier(oid))
)
_, _, _, var_binds = next(iterator)
return str(var_binds[0][1]) if var_binds else None
except:
return None
def collect_batch(self, devices: List[Dict], max_workers: int = 20) -> List[DeviceMetrics]:
"""批量采集多台设备"""
results = []
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {
executor.submit(self.collect_single, d['ip'], d['name']): d
for d in devices
}
for future in as_completed(futures):
try:
result = future.result()
results.append(result)
except Exception as e:
dev = futures[future]
logger.error(f"设备 {dev['name']} 采集异常: {e}")
return results
# 使用示例
if __name__ == '__main__':
devices = [
{'name': 'PE-BJ-01', 'ip': '10.1.1.1'},
{'name': 'PE-BJ-02', 'ip': '10.1.1.2'},
{'name': 'PE-SH-01', 'ip': '10.1.2.1'},
]
collector = SNMPCollector(community='public')
results = collector.collect_batch(devices)
for m in results:
print(f"{m.device_name}: CPU={m.cpu_usage}%, MEM={m.memory_usage}%")
5.3 配置备份工具
# tools/config_backup.py
"""
网络配置备份工具
功能：批量采集设备运行配置、生成SHA256校验、增量检测、自动归档
"""
import os
import hashlib
import logging
from netmiko import ConnectHandler
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional
logger = logging.getLogger(__name__)
class ConfigBackupManager:
"""配置备份管理器"""
def __init__(self, backup_root: str = './backups',
device_credentials: Optional[Dict] = None):
self.backup_root = Path(backup_root)
self.backup_root.mkdir(parents=True, exist_ok=True)
self.creds = device_credentials or {}
# 存储上次备份的哈希值，用于增量检测
self.last_hashes: Dict[str, str] = {}
self._load_last_hashes()
def _load_last_hashes(self):
"""加载上次备份的哈希映射"""
hash_file = self.backup_root / '.last_hashes.json'
if hash_file.exists():
import json
with open(hash_file) as f:
self.last_hashes = json.load(f)
def _save_hashes(self):
"""保存当前哈希映射"""
import json
hash_file = self.backup_root / '.last_hashes.json'
with open(hash_file, 'w') as f:
json.dump(self.last_hashes, f, indent=2, ensure_ascii=False)
def _get_device_params(self, device: Dict) -> Dict:
"""构建 Netmiko 连接参数"""
base = {
'device_type': device.get('platform', 'huawei_vrpv8'),
'host': device['ip'],
'username': device.get('username', self.creds.get('username', 'admin')),
'password': device.get('password', self.creds.get('password', '')),
'timeout': 30,
'session_timeout': 60,
}
if device.get('port'):
base['port'] = device['port']
return base
def backup_single(self, device: Dict) -> Dict:
"""备份单台设备配置"""
dev_name = device['name']
today = datetime.now().strftime('%Y/%m/%d')
backup_dir = self.backup_root / dev_name / today
backup_dir.mkdir(parents=True, exist_ok=True)
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
backup_file = backup_dir / f"{dev_name}_{timestamp}.cfg"
change_file = backup_dir / f"{dev_name}_{timestamp}.changes"
result = {
'device': dev_name,
'status': 'success',
'backup_file': str(backup_file),
'has_changes': False,
'errors': []
}
try:
with ConnectHandler(**self._get_device_params(device)) as net:
# 获取运行配置
running_config = net.send_command(
'display current-configuration',
use_textfsession=True
)
# 写入备份文件
with open(backup_file, 'w', encoding='utf-8') as f:
f.write(running_config)
# 计算哈希
current_hash = hashlib.sha256(
running_config.encode()
).hexdigest()
# 增量检测
if dev_name in self.last_hashes:
if self.last_hashes[dev_name] != current_hash:
result['has_changes'] = True
# 生成 diff 文件
self._generate_diff(
dev_name,
self.last_hashes[dev_name],
current_hash,
change_file
)
logger.info(f"设备 {dev_name} 配置已变更")
else:
result['has_changes'] = True
logger.info(f"设备 {dev_name} 首次备份")
self.last_hashes[dev_name] = current_hash
except Exception as e:
result['status'] = 'failed'
result['errors'].append(str(e))
logger.error(f"设备 {dev_name} 备份失败: {e}")
return result
def _generate_diff(self, device_name: str, old_hash: str,
new_hash: str, diff_file: Path):
"""生成配置差异报告"""
import difflib
# 读取上次备份
last_backup = self._find_last_backup(device_name)
if last_backup:
with open(last_backup, 'r', encoding='utf-8') as f:
old_config = f.read()
with open(diff_file.parent / f"{device_name}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.cfg",
'r', encoding='utf-8') as f:
new_config = f.read()
diff = difflib.unified_diff(
old_config.splitlines(),
new_config.splitlines(),
fromfile='previous',
tofile='current',
lineterm=''
)
with open(diff_file, 'w', encoding='utf-8') as f:
f.write('\n'.join(diff))
def _find_last_backup(self, device_name: str) -> Optional[Path]:
"""查找设备上次备份文件"""
dev_dir = self.backup_root / device_name
if not dev_dir.exists():
return None
cfg_files = sorted(dev_dir.rglob('*.cfg'), reverse=True)
for cfg in cfg_files:
if '.changes' not in cfg.name:
return cfg
return None
def backup_all(self, devices: List[Dict], max_workers: int = 10) -> List[Dict]:
"""批量备份所有设备"""
results = []
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {
executor.submit(self.backup_single, d): d
for d in devices
}
for future in as_completed(futures):
results.append(future.result())
# 保存哈希映射
self._save_hashes()
# 汇总报告
changed = sum(1 for r in results if r['has_changes'])
failed = sum(1 for r in results if r['status'] == 'failed')
logger.info(f"备份完成: 总计{len(results)}台, "
f"变更{changed}台, 失败{failed}台")
return results
# 配合 crontab 定时执行
# 0 2 * * * /usr/bin/python3 /opt/tools/config_backup.py --all
5.4 批量命令执行器
# tools/batch_executor.py
"""
批量命令执行器
支持：单命令批量执行、多命令顺序/并行执行、结果聚合、超时控制
"""
import logging
from netmiko import ConnectHandler
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from datetime import datetime
logger = logging.getLogger(__name__)
@dataclass
class ExecResult:
"""单台设备执行结果"""
device_name: str
ip_address: str
command: str
output: str = ""
success: bool = True
error: str = ""
duration_ms: float = 0.0
timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
class BatchCommandExecutor:
"""批量命令执行器"""
def __init__(self, default_credentials: Optional[Dict] = None,
connect_timeout: int = 30, session_timeout: int = 120):
self.default_creds = default_credentials or {}
self.connect_timeout = connect_timeout
self.session_timeout = session_timeout
def execute_single(self, device: Dict, commands: List[str],
delay_factor: int = 1) -> List[ExecResult]:
"""在单台设备上顺序执行多条命令"""
results = []
params = {
'device_type': device.get('platform', 'huawei_vrpv8'),
'host': device['ip'],
'username': device.get('username', self.default_creds.get('username')),
'password': device.get('password', self.default_creds.get('password')),
'timeout': self.connect_timeout,
'session_timeout': self.session_timeout,
'delay_factor': delay_factor,
}
try:
with ConnectHandler(**params) as net:
for cmd in commands:
import time
start = time.time()
try:
output = net.send_command(cmd, read_timeout=60)
duration = (time.time() - start) * 1000
results.append(ExecResult(
device_name=device['name'],
ip_address=device['ip'],
command=cmd,
output=output,
success=True,
duration_ms=duration
))
except Exception as e:
duration = (time.time() - start) * 1000
results.append(ExecResult(
device_name=device['name'],
ip_address=device['ip'],
command=cmd,
success=False,
error=str(e),
duration_ms=duration
))
except Exception as e:
# 连接失败，所有命令标记失败
for cmd in commands:
results.append(ExecResult(
device_name=device['name'],
ip_address=device['ip'],
command=cmd,
success=False,
error=f"连接失败: {str(e)}"
))
return results
def execute_parallel(self, devices: List[Dict], commands: List[str],
max_workers: int = 20) -> List[ExecResult]:
"""并行执行：所有设备同时执行指定命令"""
all_results = []
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {
executor.submit(self.execute_single, d, commands): d['name']
for d in devices
}
for future in as_completed(futures):
dev_name = futures[future]
try:
results = future.result()
all_results.extend(results)
except Exception as e:
logger.error(f"设备 {dev_name} 执行异常: {e}")
return all_results
def execute_sequential_with_check(self, devices: List[Dict],
commands: List[Dict],
check_interval: int = 60) -> List[ExecResult]:
"""
顺序执行带健康检查
commands: [{'cmd': '...', 'check': 'keyword', 'critical': True}]
如果关键命令检查结果不匹配，停止后续设备执行
"""
all_results = []
for device in devices:
results = self.execute_single(device,
[c['cmd'] for c in commands])
all_results.extend(results)
# 检查结果
for i, cmd_info in enumerate(commands):
if cmd_info.get('critical') and i < len(results):
result = results[i]
if result.success:
check_keyword = cmd_info.get('check', '')
if check_keyword and check_keyword not in result.output:
logger.warning(
f"设备 {device['name']} 检查结果异常: "
f"期望 '{check_keyword}' 不在输出中"
)
# 关键检查失败，是否中断？
if cmd_info.get('stop_on_fail'):
logger.error("关键检查失败，停止后续设备执行")
break
# 等待检查间隔
if check_interval > 0:
import time
time.sleep(check_interval)
return all_results
def generate_report(self, results: List[ExecResult]) -> str:
"""生成执行报告"""
lines = ["## 批量命令执行报告", ""]
lines.append(f"| 设备 | 命令 | 状态 | 耗时(ms) | 错误 |")
lines.append(f"|------|------|------|----------|------|")
for r in results:
status = '✅' if r.success else '❌'
lines.append(
f"| {r.device_name} | `{r.command[:40]}...` "
f"| {status} | {r.duration_ms:.0f} | {r.error[:30]} |"
)
total = len(results)
failed = sum(1 for r in results if not r.success)
lines.extend([
"",
f"**统计**: 总计 {total} 条, 成功 {total - failed} 条, "
f"失败 {failed} 条, 成功率 {(total-failed)/total*100:.1f}%"
])
return '\n'.join(lines)
5.5 工具集单元测试示例
# tests/test_network_tools.py
"""Network Tools 工具集单元测试"""
import pytest
from unittest.mock import MagicMock, patch
from tools.snmp_collector import SNMPCollector, DeviceMetrics
from tools.config_backup import ConfigBackupManager
from tools.batch_executor import BatchCommandExecutor, ExecResult
class TestSNMPCollector:
"""SNMP 采集器测试"""
@pytest.fixture
def collector(self):
return SNMPCollector(community='public', version=2)
def test_collect_single_success(self, collector):
"""测试单设备采集成功"""
with patch('tools.snmp_collector.snmp') as mock_snmp:
# Mock SNMP 返回值
mock_iterator = iter([(None, 0, 0,
[(snmp.ObjectIdentifier('1.3.6.1.4.1.2011.5.25.31.1.1.1.1.1'),
MagicMock().__str__=MagicMock(return_value='45'))])])
mock_snmp.getCmd.return_value = mock_iterator
result = collector.collect_single('10.1.1.1', 'PE-BJ-01')
assert result.device_name == 'PE-BJ-01'
assert result.ip_address == '10.1.1.1'
assert isinstance(result.timestamp, str)
def test_collect_single_timeout(self, collector):
"""测试采集超时处理"""
with patch('tools.snmp_collector.snmp') as mock_snmp:
mock_snmp.getCmd.side_effect = TimeoutError()
result = collector.collect_single('10.1.1.1', 'PE-BJ-01')
assert len(result.errors) > 0
def test_device_metrics_defaults(self):
"""测试 DeviceMetrics 默认值"""
m = DeviceMetrics(device_name='test', ip_address='1.2.3.4')
assert m.cpu_usage is None
assert m.memory_usage is None
assert m.interfaces == []
assert m.errors == []
class TestConfigBackupManager:
"""配置备份工具测试"""
@pytest.fixture
def backup_mgr(self, tmp_path):
mgr = ConfigBackupManager(backup_root=str(tmp_path / 'backups'))
return mgr
def test_backup_single_success(self, backup_mgr):
"""测试单设备备份成功"""
with patch('tools.config_backup.ConnectHandler') as mock_connect:
mock_net = MagicMock()
mock_net.send_command.return_value = 'sysname test-device\n'
mock_connect.return_value.__enter__ = MagicMock(return_value=mock_net)
mock_connect.return_value.__exit__ = MagicMock(return_value=False)
result = backup_mgr.backup_single({
'name': 'PE-BJ-01', 'ip': '10.1.1.1'
})
assert result['status'] == 'success'
assert result['has_changes'] is True # 首次备份
def test_backup_detects_no_change(self, backup_mgr):
"""测试无变更时不标记变更"""
device = {'name': 'PE-BJ-01', 'ip': '10.1.1.1'}
config = 'sysname test-device\n'
with patch('tools.config_backup.ConnectHandler') as mock_connect:
mock_net = MagicMock()
mock_net.send_command.return_value = config
mock_connect.return_value.__enter__ = MagicMock(return_value=mock_net)
mock_connect.return_value.__exit__ = MagicMock(return_value=False)
# 首次备份
backup_mgr.backup_single(device)
# 再次备份（配置未变）
result = backup_mgr.backup_single(device)
assert result['has_changes'] is False
class TestBatchExecutor:
"""批量命令执行器测试"""
@pytest.fixture
def executor(self):
return BatchCommandExecutor(
default_credentials={'username': 'admin', 'password': 'test'}
)
def test_generate_report(self, executor):
"""测试报告生成"""
results = [
ExecResult('PE-BJ-01', '10.1.1.1', 'display version',
'VRP version 8.x', True, '', 120.5),
ExecResult('PE-BJ-02', '10.1.1.2', 'display version',
'', False, 'Connection timeout', 30000),
]
report = executor.generate_report(results)
assert '## 批量命令执行报告' in report
assert '✅' in report
assert '❌' in report
assert '成功率 50.0%' in report
5.6 CI/CD 集成方案
# .gitlab-ci.yml - Network Tools CI/CD 集成
stages:
- unit-test
- lint
- integration-test
- deploy-tools
variables:
PYTHONPATH: "${PYTHONPATH}:."
# 单元测试
unit_test_tools:
stage: unit-test
image: python:3.11-slim
script:
- pip install -r requirements.txt
- pip install -r requirements-test.txt
- pytest tests/test_network_tools.py -v --junitxml=report.xml
--cov=tools --cov-report=xml --cov-report=html
coverage: '/TOTAL.*\s+(\d+%)/'
artifacts:
reports:
junit: report.xml
coverage_report:
coverage_format: cobertura
path: coverage.xml
paths:
- htmlcov/
expire_in: 30 days
# 代码规范检查
lint_tools:
stage: lint
image: python:3.11-slim
script:
- pip install ruff mypy black
- ruff check tools/
- black --check tools/
- mypy tools/ --ignore-missing-imports
# 集成测试（连接仿真环境）
integration_test:
stage: integration-test
image: python:3.11-slim
script:
- pip install -r requirements.txt
- pytest tests/integration/ -v
--devices-config=tests/fixtures/simulator_devices.yaml
needs:
- unit_test_tools
- lint_tools
rules:
- if: $CI_COMMIT_BRANCH == "main"
# 打包部署工具
deploy_tools:
stage: deploy-tools
image: python:3.11-slim
script:
- pip install build
- python -m build
- echo "构建完成，产物: dist/*"
artifacts:
paths:
- dist/
expire_in: 7 days
rules:
- if: $CI_COMMIT_BRANCH == "main"
六、总结
A["单元测试<br/>配置语法检查<br/>YAML 格式校验<br/>模板渲染验证"] --> B
B["集成测试<br/>单设备状态验证<br/>BGP/OSPF/ACL<br/>接口/路由表"] --> C
C["端到端测试<br/>多设备连通性<br/>业务流验证<br/>故障切换演练"] --> D
D["混沌测试<br/>随机端口 Down<br/>链路中断模拟<br/>设备重启演练"]
style A fill:#E8F5E9
style B fill:#E3F2FD
style C fill:#FFF3E0
style D fill:#FFEBEE
```
故障案例：pytest 拦截 BGP Timer 变更回归
在 CI/CD Pipeline（第322篇）中，工程师提交了 BGP Timer 调整的配置变更。pytest 集成测试阶段发现了问题：
test_bgp_peer_state检测到 PE-BJ-01 的 BGP 会话从 Established 降至 Idletest_bgp_peer_count检测到 Established Peer 数从 4 降至 2- Pipeline 自动标记为 Failed，阻止了后续的生产部署
- 工程师分析发现：对端设备的 holdtime 未同步调整，导致 BGP Timer 不匹配
- 补充对端设备的 Timer 配置后，测试全部通过，变更成功部署
思考题
- pytest 网络测试中的"期望状态"文件（expected_states.yaml）由谁维护？当网络变更频繁时，如何确保期望状态的时效性？
- 对于涉及多台设备协同工作的测试（如 BGP 对等会话、OSPF 邻接关系），如何设计测试用例以准确验证端到端状态？
下篇预告：第327篇《ChatOps：用飞书/钉钉机器人操作网络》——通过即时通讯工具机器人实现网络操作的对话式交互。