VPP 测试框架之官方文档解读

原创

一觉醒来写程序

修改于 2025-08-31 19:57:14

12100

代码可运行

文章被收录于专栏：vpp开发与应用vpp开发与应用

运行总次数：0

代码可运行

通过阅读本文你将对vpp测试架有一个基本的了解。该系列文章有4篇：

VPP 测试框架之官方文档解读（本文）
VPP 测试框架之使用示例及参数说明
Vpp 测试框架之infra基础库测试
Vpp Bond单元测试源码分析

本文已同步至：

个人博客：itwakeup.com
微信公众号：vpp与dpdk研习社（vpp_dpdk_lab）

作为测试框架系列的第一篇文章，本文翻译自官方文档

1. 概述

VPP 测试框架的目标是简化 VPP 单元测试的编写、运行和调试。为此vpp选择了 Python 作为高级语言，以实现快速开发，并使用 Scapy 提供创建和解析数据包的必要工具。

2. 测试用例剖析

Python 的单元测试作为 VPP 构建测试的基础框架。VPP 测试框架中的测试套件由多个派生自VppTestCase 的类组成，而 VppTestCase 本身又派生自 TestCase 。测试类定义一个或多个测试函数，这些函数用于执行测试用例。

测试用例运行流程：

setUpClass：此函数会为每个测试类调用一次（一个测试类可以包含多个测试函数），用于执行一次性测试设置。如果此函数抛出异常，则所有测试函数均不会执行。
setUp：此函数在每个测试函数之前运行。如果此函数抛出除 AssertionError 或 SkipTest 之外的异常，则视为错误，而不是测试失败。
test_<name>：这是测试用例的核心函数，在各种测试场景中被执行，并使用 unittest 框架中的各种断言函数来检查执行结果。一个测试类中可以存在多个test_<name>。
tearDown：tearDown 函数在每个测试函数之后调用，目的是进行部分清理。
tearDownClass：在运行完所有测试函数后调用，以执行最后的清理</font>。

3. 日志记录

每个测试用例都会自动创建一个logger，并根据 logging 属性存储在 'logger' 属性中。使用debug()、info()、error() 等logger发送日志消息。所有日志消息将放在临时目录中的日志文件里（见下文）。

要控制打印到控制台的消息，请指定 V= 参数。

make test         # 最少信息输出
make test V=1     # 中等信息输出
make test V=2     # 最大信息输出

4. 并行测试执行

VPP测试框架的测试套件支持并行执行。每个测试套件都在由Python multiprocessing进程生成的独立进程中运行。

子测试套件的结果通过管道发送到父级，并在运行结束时进行汇总和总结。

子进程中记录的标准输出 (stdout)、标准错误输出 (stderr) 和日志会被重定向到父进程管理的独立队列。这些队列中的数据会按照测试套件完成的顺序发送到父进程的标准输出 (stdout)。如果没有已完成的测试套件（例如测试刚开始时），则会实时发送上次启动的测试套件的数据。

要启用并行测试运行，请指定并行进程的数量：

make test TEST_JOBS=n       # 最多将生成 n 个进程
make test TEST_JOBS=auto    # 根据核心数量和共享内存大小进行自动选择

5. 测试临时目录和 VPP 生命周期

测测试分离是通过分离测试文件和 VPP 实例来实现的：

每个测试都会在/tmp/下创建一个临时目录，如/tmp/vpp-unittest-TestBondInterface/
使用测试名称作为各种配置的前缀，用于运行 VPP 实例，如运行bond测试用例时，部分配置如下：

{
  ...
  api-segment { prefix vpp-unittest-TestBondInterface }
  ...
}

这样测试环境中运行的任何其他 VPP 实例与测试 VPP 之间就不会发生冲突。测试用例创建的所有临时文件都存储在此临时测试目录中。测试临时目录包含以下的文件：

log.txt：包含最详细程度的日志输出
pg*_in.pcap：最后注入 VPP 的数据包流，以接口命名，因此对于 pg0，该文件将被命名为 pg0_in.pcap
pg*_out.pcap：VPP 为接口创建的最后一个捕获文件，同样以接口命名，例如对于 pg1，该文件将被命名为 pg1_out.pcap
历史文件：每当重新启动捕获或添加新流时，现有文件都会被滚动存档和重命名，因此所有 pcap 文件始终会保存以供以后需要时进行调试
core：如果 vpp 输出core文件，它将存储在临时目录中
vpp_stdout.txt：包含 vpp 打印到 stdout 的输出的文件
vpp_stderr.txt：包含 vpp 打印到 stderr 的输出的文件

注意：调用make test*或make retest*时，名为vpp-unittest-*的现有临时目录会被自动删除，以保持临时目录清洁。

6. 虚拟环境

Virtualenv 是一个 Python 模块，它提供了一种创建包含 VPP 测试框架所需依赖项的环境的方法，从而允许与任何现有的系统级软件包分离。VPP 测试框架的 Make 文件会自动在 build-root 中创建一个虚拟环境，并在该环境中安装所需的软件包。每当通过 make test 目标之一执行测试时，都会进入该环境。

7. 命名约定

大多数单元测试都会涉及某种形式的数据包操作——在VPP（Vector Packet Processing）与连接到VPP的虚拟主机之间发送和接收数据包。在描述方向、地址等时，始终以VPP的视角为参照，因此：

local_前缀用于 VPP 端。例如，local_ip4 <VppInterface.local_ip4> 地址是分配给 VPP 接口的 IPv4 地址。
remote_前缀用于虚拟主机端。例如， remote_mac <VppInterface.remote_mac> 地址是分配给连接到 VPP 的虚拟主机的 MAC 地址。

8. 自动生成的地址

要发送数据包，通常需要提供某些地址，否则数据包会被丢弃。VPP测试框架中的接口对象通常会根据其索引自动分配地址，这既能避免地址冲突，又能通过统一的编址方案简化调试。

def set_sw_if_index(self, sw_if_index):
    # ...
    self._local_ip4 = "172.16.%u.1" % self.sw_if_index
    self._local_ip4_len = 24
    # ...

测试用例的开发者通常无需直接处理实际数值，而是通过对象属性来操作。地址通常分为两种形式：

<address> ：Python字符串格式的地址
<address>n（注意带'n'后缀）：通过socket.inet_pton转换为网络字节序的原始格式——这种格式适合直接作为参数传递给VPP API。

例如，分配给 VPP 接口的 IPv4 地址：

local_ip4：VPP 接口上的本地 IPv4 地址（字符串）
local_ip4n：网络字节序的本地 IPv4 地址，适合作为 API 参数。

这些地址需要在 VPP 中进行配置才能使用，例如 VppInterface.config_ip4 API。请参阅文档以 VppInterface 了解更多详细信息。

默认情况下，为 L3 创建的每种类型的远程地址都有一个： remote_ip4 和 remote_ip6。如果测试需要模拟更多远程主机，因而需要创建更多地址，可以使用 generate_remote_hosts API 并使用 configure_ipv4_neighbors API 将它们的条目插入到 ARP 表中。

9. VPP测试框架中的数据包流

9.1. 测试框架 -> VPP

VPP测试框架并不会直接向VPP发送数据包，而是通过包生成器接口（由VppPGInterface类实现）注入流量。具体工作流程如下：

数据包写入临时文件undefined测试框架会先将数据包写入一个临时.pcap文件
VPP读取并注入流量undefinedVPP随后读取该文件，并将数据包注入到其内部处理流程中

要将数据包列表添加到接口，请调用VppPGInterface.add_stream方法。一切准备就绪后，调用 pg_start 方法启动 VPP 端的数据包生成器。

9.2. VPP-> 测试框架

同样，VPP 不会直接向 VPP 测试框架发送任何数据包。相反，它会使用数据包捕获功能捕获流量并将其写入临时的 .pcap 文件，然后由 VPP 测试框架读取和分析该文件。

以下 API 可用于测试用例读取 pcap 文件。

VppPGInterface.get_capture：此API适用于Bulk（大量数据包）或Batch（分批处理的数据包）测试，这会准备并发送一组数据包，然后读取并验证接收到的数据包。该API需要预期捕获的数据包数量（忽略被过滤的数据包——详见下文）以确定VPP何时完全写完pcap文件。若使用packet infos进行数据包验证，则packet infos的计数可被VppPGInterface.get_capture自动用于获取正确计数（此时可为expected_count参数提供默认值None或直接省略该参数）。
VppPGInterface.wait_for_packet：此API适用于交互式测试场景，例如执行会话管理、三次握手等操作。该API会等待并返回单个数据包，同时保留捕获文件并维持上下文状态。重复调用时，将从同一捕获文件（即同一接口上到达的数据包）中返回后续数据包（若超时则抛出异常）。

注意：除非理解这些API的内部工作原理，否则不建议混用它们。这些API都不会轮换（重新创建或更新捕获文件）pcap捕获文件，因此如果在调用VppPGInterface.wait_for_packet后再调用VppPGInterface.get_capture，将返回已读取过的数据包。只有在调用VppPGInterface.enable_capture后切换API才是安全的，因为该API会轮换捕获文件。

9.3. 自动过滤数据包

默认情况下，这两个API（VppPGInterface.get_capture 和 VppPGInterface.wait_for_packet）会对数据包捕获进行过滤，移除已知的无用数据包——包括IPv6路由器通告（Router Advertisements）和IPv6路由器警报（Router Alerts）。这些数据包是未经请求的，从VPP测试框架的角度来看是随机的。

如果测试需要接收这些数据包，则应在 filter_out_fn 参数中指定 None 或自定义的过滤函数。

9.4. 发送/接收数据包的通用 API 流程

我们将描述一个简单的场景：假设已通过 create_pg_interfaces API 创建了接口，数据包会从 pg0 接口发送至 pg1 接口。

1）为 pg0 创建数据包列表：

packet_count = 10
packets = create_packets(src=self.pg0, dst=self.pg1,
                         count=packet_count)

2）将该数据包列表添加到源接口：

self.pg0.add_stream(packets)

3）在目标接口上启用捕获：

self.pg1.enable_capture()

4）启动数据包生成器：

self.pg_start()

5）等待捕获文件出现并读取它：

capture = self.pg1.get_capture(expected_count=packet_count)

6）验证数据包是否与发送的数据包匹配：

self.verify_capture(send=packets, captured=capture)

10. 测试框架对象

以下对象提供VPP抽象层，使测试用例能够便捷执行常规操作：

VppInterface：抽象基类，代表通用VPP接口，包含供派生类使用的公共功能
VppPGInterface：实现VPP报文生成器接口的类。该接口会随对象创建/销毁而自动建立/删除
VppSubInterface：VPP子接口抽象类，包含如VppDot1QSubint和VppDot1ADSubint等子类的通用功能

11. VPP API/CLI 的调用方式

VPP通过名为vpp-papi的Python模块提供接口绑定，该模块由测试框架安装在虚拟环境中。基于vpp-papi构建的VppPapiProvider抽象层主要实现以下功能：

返回值自动校验
每次API调用后自动校验返回值（默认期望值为0，可自定义）
若校验失败则抛出异常
钩子函数自动调用
before_cli <Hook.before_cli> 和 before_api <Hook.before_api> 钩子用于：
- 调试日志记录
- 测试单步执行
after_cli <Hook.after_cli> 和 after_api <Hook.after_api> 钩子用于：
- 监控VPP进程崩溃情况
API调用简化
针对多数需要大量参数的VPP接口（如ip_add_del_route需约25个参数）：
- 提供合理的默认参数配置
- 常规场景仅需指定3个关键参数即可
- 显著提升代码可读性和易用性

12. 实用方法

一些有用的工具方法包括：

ppp：'Pretty Print Packet' - 返回包含与Scapy的packet.show()相同输出的字符串
ppc：'Pretty Print Capture' - 使用ppp返回包含抓包输出的字符串（可配置打印的数据包数量上限）

注意：不要在测试中使用Scapy的packet.show()，因为它会将输出打印到stdout。所有输出都应发送到与测试用例关联的logger。

13. 示例：如何添加新测试

在此示例中，我们将描述如何添加一个测试基本 IPv4 转发的新测试用例。

1）在测试目录中添加一个名为 test _ip4_fwd.py 的新文件，并导入依赖包：

from framework import VppTestCase
from scapy.layers.l2 import Ether
from scapy.packet import Raw
from scapy.layers.inet import IP, UDP
from random import randint

2）创建一个从 VppTestCase继承的类：

class IP4FwdTestCase(VppTestCase):
    """ IPv4 simple forwarding test case """

3）添加一个 setUpClass 函数，其中包含运行测试所需的设置：

@classmethod
def setUpClass(self):
    super(IP4FwdTestCase, self).setUpClass()
    self.create_pg_interfaces(range(2))  #  create pg0 and pg1
    for i in self.pg_interfaces:
        i.admin_up()  # put the interface up
        i.config_ip4()  # configure IPv4 address on the interface
        i.resolve_arp()  # resolve ARP, so that we know VPP MAC

4）创建一个辅助方法来创建要发送的数据包：

def create_stream(self, src_if, dst_if, count):
    packets = []
    for i in range(count):
        # create packet info stored in the test case instance
        info = self.create_packet_info(src_if, dst_if)
        # convert the info into packet payload
        payload = self.info_to_payload(info)
        # create the packet itself
        p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
             IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4) /
             UDP(sport=randint(1000, 2000), dport=5678) /
             Raw(payload))
        # store a copy of the packet in the packet info
        info.data = p.copy()
        # append the packet to the list
        packets.append(p)

    # return the created packet list
    return packets

5）创建一个辅助方法来验证捕获数据：

def verify_capture(self, src_if, dst_if, capture):
    packet_info = None
    for packet in capture:
        try:
            ip = packet[IP]
            udp = packet[UDP]
            # convert the payload to packet info object
            payload_info = self.payload_to_info(packet[Raw])
            # make sure the indexes match
            self.assert_equal(payload_info.src, src_if.sw_if_index,
                              "source sw_if_index")
            self.assert_equal(payload_info.dst, dst_if.sw_if_index,
                              "destination sw_if_index")
            packet_info = self.get_next_packet_info_for_interface2(
                src_if.sw_if_index,
                dst_if.sw_if_index,
                packet_info)
            # make sure we didn't run out of saved packets
            self.assertIsNotNone(packet_info)
            self.assert_equal(payload_info.index, packet_info.index,
                              "packet info index")
            saved_packet = packet_info.data  # fetch the saved packet
            # assert the values match
            self.assert_equal(ip.src, saved_packet[IP].src,
                              "IP source address")
            # ... more assertions here
            self.assert_equal(udp.sport, saved_packet[UDP].sport,
                              "UDP source port")
        except:
            self.logger.error(ppp("Unexpected or invalid packet:",
                                  packet))
            raise
    remaining_packet = self.get_next_packet_info_for_interface2(
        src_if.sw_if_index,
        dst_if.sw_if_index,
        packet_info)
    self.assertIsNone(remaining_packet,
                      "Interface %s: Packet expected from interface "
                      "%s didn't arrive" % (dst_if.name, src_if.name))

6）在 test_basic 函数中添加测试代码：

def test_basic(self):
    count = 10
    # create the packet stream
    packets = self.create_stream(self.pg0, self.pg1, count)
    # add the stream to the source interface
    self.pg0.add_stream(packets)
    # enable capture on both interfaces
    self.pg0.enable_capture()
    self.pg1.enable_capture()
    # start the packet generator
    self.pg_start()
    # get capture - the proper count of packets was saved by
    # create_packet_info() based on dst_if parameter
    capture = self.pg1.get_capture()
    # assert nothing captured on pg0 (always do this last, so that
    # some time has already passed since pg_start())
    self.pg0.assert_nothing_captured()
    # verify capture
    self.verify_capture(self.pg0, self.pg1, capture)

7）执行 make test命令来运行测试，如果想仅运行此特定测试，则执行make test TEST=test_ip4_fwd。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

智能高性能网络

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

智能高性能网络

#vpp

#测试框架

登录后参与评论

0 条评论

热度