告别手动点击！用Python API让SPX测试仪自动跑起来
作者：诺居科技产品团队
上期我们介绍了如何通过图形界面操作SPX测试仪，跑通第一个测试。但在实际工作中，手动点GUI会遇到两个痛点：
- 2. 批量测试：要测100种流量组合，手动操作根本不现实
这就是SPX Python API的用武之地。本文手把手带你写出第一个自动化测试脚本，从此让测试仪"听你的指挥"。
一、为什么选择Python API？
SPX基于DPDK构建，同时提供了完整的Python SDK，让你可以用代码完整描述一次测试：
配置端口 → 构造流量 → 启动发包 → 收集统计 → 输出报告
与GUI相比的优势：
二、环境准备
2.1 安装依赖
SPX的Python SDK已内置于测试仪系统，在控制PC上安装客户端库：
pip install scapy
💡 说明：SPX使用Scapy构建数据包模板，这是它兼具灵活性和高性能的关键所在。
2.2 确认测试仪已启动
确保SPX测试仪已上电运行，并确认：
- • 管理网络可达（ping
192.168.188.188 通）
2.3 项目结构
创建一个工作目录，结构如下：
spx-test/
├── simple_test.py # 今天的第一个脚本
├── utils/
│ └── stats_parser.py # 后续扩展用
└── results/ # 测试结果输出
三、第一个Python脚本：发送UDP流
3.1 完整代码
#!/usr/bin/env python3
"""
SPX Python API 入门示例
功能：向DUT发送20条UDP流，统计丢包率
"""
from trex_stl_lib.api import *
# ============================================================
# 第一步：构造数据包
# ============================================================
defcreate_stream():
"""用Scapy构造一个UDP数据包，并配置IP变化"""
# 使用Scapy定义数据包结构
pkt = (
Ether() /
IP(src="16.0.0.1", dst="48.0.0.1") /
UDP(sport=10000, dport=20000) /
Raw(load=b'\x00' * 64) # 64字节payload
)
# 配置Field Engine：让src_ip从16.0.0.1循环变化到16.0.0.20
vm = STLScVmRaw([
STLVmFlowVar(
name="src_ip",
min_value="16.0.0.1",
max_value="16.0.0.20",
size=4,
op="inc"# inc = 递增
),
STLVmWrFlowVar(fv_name="src_ip", pkt_offset="IP.src"),
STLVmFixIpv4Checksum(), # IP地址变了，自动修正校验和
STLVmFixChecksumHw(l3_offset="IP", l4_offset="UDP"),
])
# 把数据包和Field Engine组合成Stream
stream = STLStream(
packet=STLPktBuilder(pkt=pkt, vm=vm),
mode=STLTXCont(pps=1000), # 持续发送，速率1000 pps
)
return stream
# ============================================================
# 第二步：连接测试仪，加载流量，发包
# ============================================================
defrun_test(server_ip="192.168.188.188", duration=10):
"""
连接SPX，在port0和port1上双向发流，持续duration秒后统计结果
"""
client = STLClient(server=server_ip)
try:
# 连接测试仪
client.connect()
print(f"[✓] 已连接到 {server_ip}")
# 占用端口0和端口1
client.acquire(ports=[0, 1], force=True)
client.reset(ports=[0, 1])
# 设置为L3模式，指定端口IP和网关
client.set_port_attr(ports=[0], promiscuous=False)
client.set_port_attr(ports=[1], promiscuous=False)
# 加载流量配置
stream_0 = create_stream()
# Port 1方向：源目互换
pkt_1 = (
Ether() /
IP(src="48.0.0.1", dst="16.0.0.1") /
UDP(sport=20000, dport=10000) /
Raw(load=b'\x00' * 64)
)
vm_1 = STLScVmRaw([
STLVmFlowVar(
name="src_ip",
min_value="48.0.0.1",
max_value="48.0.0.20",
size=4,
op="inc"
),
STLVmWrFlowVar(fv_name="src_ip", pkt_offset="IP.src"),
STLVmFixIpv4Checksum(),
])
stream_1 = STLStream(
packet=STLPktBuilder(pkt=pkt_1, vm=vm_1),
mode=STLTXCont(pps=1000),
)
client.add_streams(stream_0, ports=[0])
client.add_streams(stream_1, ports=[1])
# ============================================================
# 第三步：启动发包，等待，停止
# ============================================================
print(f"[→] 开始发包，持续 {duration} 秒...")
client.start(ports=[0, 1], mult="1", duration=duration)
client.wait_on_traffic(ports=[0, 1])
print("[■] 发包结束")
# ============================================================
# 第四步：读取统计结果
# ============================================================
stats = client.get_stats()
print_results(stats)
except STLError as e:
print(f"[✗] 测试失败: {e}")
finally:
client.disconnect()
print("[✓] 已断开连接")
# ============================================================
# 第五步：格式化输出结果
# ============================================================
defprint_results(stats):
"""解析并打印关键统计指标"""
print("\n" + "=" * 50)
print(" 测试结果汇总")
print("=" * 50)
for port_id in [0, 1]:
port_stats = stats[port_id]
tx_pkts = port_stats.get("opackets", 0)
rx_pkts = port_stats.get("ipackets", 0)
tx_bytes = port_stats.get("obytes", 0)
rx_bytes = port_stats.get("ibytes", 0)
loss = 0
if tx_pkts > 0:
# 简单用另一端的rx来计算（双端口对打）
other = stats[1 - port_id]
rx_from_other = other.get("ipackets", 0)
loss = max(0, tx_pkts - rx_from_other)
loss_pct = loss / tx_pkts * 100
print(f"\n Port {port_id}:")
print(f" 发包数：{tx_pkts:>10,} pkts {tx_bytes/1024/1024:.1f} MB")
print(f" 收包数：{rx_pkts:>10,} pkts {rx_bytes/1024/1024:.1f} MB")
print(f" 丢包率：{loss_pct:.2f}%")
print("=" * 50)
if __name__ == "__main__":
run_test(
server_ip="192.168.188.188",
duration=10, # 测试时长（秒）
)
3.2 运行脚本
python simple_test.py
预期输出：
[✓] 已连接到 192.168.188.188
[→] 开始发包，持续 10 秒...
[■] 发包结束
==================================================
测试结果汇总
==================================================
Port 0:
发包数： 10,000 pkts 1.2 MB
收包数： 10,000 pkts 1.2 MB
丢包率：0.00%
Port 1:
发包数： 10,000 pkts 1.2 MB
收包数： 10,000 pkts 1.2 MB
丢包率：0.00%
==================================================
[✓] 已断开连接
四、进阶：批量测试不同流量速率
掌握了基础脚本后，来看一个更实用的场景——自动遍历不同发包速率，找出设备的性能拐点：
defbatch_rate_test(server_ip="192.168.188.188"):
"""
自动测试不同pps速率下的丢包情况
用于初步摸清DUT的转发性能区间
"""
test_rates = [1000, 5000, 10000, 50000, 100000] # pps
results = []
for rate in test_rates:
print(f"\n[测试] 速率 {rate} pps ...")
loss_pct = run_single_rate(server_ip, rate, duration=5)
results.append({"rate_pps": rate, "loss_pct": loss_pct})
print(f" → 丢包率: {loss_pct:.2f}%")
# 打印汇总表格
print("\n\n批量测试汇总：")
print(f"{'速率 (pps)':<15}{'丢包率':<10}{'状态'}")
print("-" * 35)
for r in results:
status = "✓ 正常"if r["loss_pct"] == 0else"✗ 丢包"
print(f"{r['rate_pps']:<15}{r['loss_pct']:.2f}%{'':<5}{status}")
只需改一行 test_rates，就能自动完成所有测试，结果一目了然。
五、代码背后的设计思路
理解SPX Python API的两个核心概念，写代码会更顺手：
5.1 Stream（流）= 包模板 + 发送规则
Stream
├── STLPktBuilder → 数据包内容（Scapy构造）
│ └── STLScVmRaw → Field Engine（字段变化规则）
└── STLTXCont → 发送模式（持续/单次/突发）
一个Stream描述"发什么包、按什么规则变化、以什么节奏发"，是SPX的最小调度单元。
5.2 Field Engine = 硬件层面的包重写
每次发包时，DPDK在硬件层面修改字段值（如IP递增），不需要CPU重新构造每一个包，这正是SPX能达到高性能的关键。你在Python里写的Field Engine规则，最终会被翻译成DPDK的硬件指令。
六、常见报错与解决
| | |
Connection refused | | |
Port is owned by another | | |
Checksum error | | 在 vm 中加入 STLVmFixIpv4Checksum() |
STLError: no streams | | 确认 add_streams() 在 start() 前执行 |
ImportError: trex_stl_lib | | |
总结
通过本文，你学会了：
- • ✅ 用Field Engine配置IP递增变化
- • ✅ 用Python API完整控制发包、等待、读取统计
Python API让SPX从一台"手动仪表"变成了真正的自动化测试平台。一旦流程跑通，测试脚本就是你最好的测试文档——代码即规范，运行即验证。
下期预告：《有状态流量测试实战：用SPX模拟真实DNS/HTTP会话》——我们将用STF模式模拟客户端和服务器之间的完整会话，看看DUT在真实业务流量下的表现如何。
关于诺居科技
杭州诺居科技有限公司，专注于网络测试仪器的研发与创新，为网络工程师提供高性能、高可靠性的测试解决方案。
有问题？欢迎在评论区留言，或添加我们的技术交流群！
