Vpp Bond单元测试源码分析

原创

一觉醒来写程序

修改于 2025-09-08 14:24:45

3800

代码可运行

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

运行总次数：0

代码可运行

1. 前言

本文将以bond单元测试为例，演示单元测试的运行、问题排查以及源码分析。大部分源码分析直接写在代码注释里。

涉及文件：

test/test_bond.py
test/vpp_bond_interface.py
test/vpp_interface.py

该系列文章有4篇：

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

本文已同步至：

个人博客：itwakeup.com

微信公众号：vpp与dpdk研习社（vpp_dpdk_lab）

2. 运行单元测试

这里以debug版本运行单元测试

make test-debug TEST=bond

==============================================================================
Bond Test Case [main thread only]
==============================================================================
Bond add/delete interface test                                      2.98 OK
Bond add_member/detach member test                                  2.76 OK
Bond hw interface link state test                                   2.79 OK
Bond traffic test                                                   2.95 OK

==============================================================================
TEST RESULTS:
    Scheduled tests: 4
     Executed tests: 4
       Passed tests: 4
==============================================================================

注：运行单元测试的时候会新启一个vpp实例，与当前运行的vpp实例不冲突。

3. 源码分析

3.1. test_bond.py代码分析

3.1.1. 初始化

3.1.1.1. setUpClass

本单元测试的全局设置函数，在bond单元测试的生命周期里只运行一次。

# classmethod：python装饰器，用于修饰类中的方法，使其变成“类方法”，可以通过类名直接调用这个方法，而不需要先创建类的实例。
@classmethod
    def setUpClass(cls):
        # 调用父类的 setUpClass
        super(TestBondInterface, cls).setUpClass()
        # 定义每次发送的数据包数量（未用到）
        cls.pkts_per_burst = 257
        # 创建4个 packet generator（pg）接口
        cls.create_pg_interfaces(range(4))
        # 定义测试中会用到的不同数据包大小
        cls.pg_if_packet_sizes = [64, 512, 1518]  # , 9018]

        # 将所有pg接口up起来
        for i in cls.pg_interfaces:
            i.admin_up()

3.1.1.2. setUp

bond单元测试里，执行每个测试函数之前运行。

def setUp(self):
    # 调用父类的 setUp，无其他逻辑
    super(TestBondInterface, self).setUp()

3.1.2. 辅助函数

3.1.2.1. show_commands_at_teardown

在单元测试结束，清理函数（tearDown）执行之前调用该函数。

def show_commands_at_teardown(self):
    # 调用vpp cli命令行"show interface"，并输出到info日志里。
    self.logger.info(self.vapi.ppcli("show interface"))

3.1.3. 单元测试函数

涉及到四个测试函数：

test_bond_traffic：测试 bond 接口的数据包转发功能
test_bond_add_member：测试 bond 接口的成员添加/删除
test_bond：测试 bond 接口的创建和删除功能
test_bond_link：测试 bond 接口的链路状态（up/down）

这里对test_bond_add_member和test_bond_traffic进行代码分析，其它函数类似。

3.1.3.1. test_bond_add_member

1）解释@unittest.skipIf

@unittest.skipIf(
        # 如果lacp插件被排除，则跳过该测试用例
        "lacp" in config.excluded_plugins, "Exclude tests requiring LACP plugin"
    )
    def test_bond_add_member(self):
        ....

可以在可以在test_<name>测试函数前添加装饰器，用于有条件地执行测试函数，可以实现：

插件依赖检查: 跳过依赖特定插件的测试
平台兼容性: 跳过不支持的平台
功能开关: 根据编译选项跳过测试
环境检查: 跳过需要特定环境的测试

常用的装饰器有：

@unittest.skipIf(condition, reason)
- condition: 布尔表达式，如果为 True，则跳过测试
- reason: 字符串，说明跳过测试的原因
@unittest.skipUnless(condition, reason)
- condition: 布尔表达式，如果为 False，则跳过测试
- reason: 字符串，说明跳过测试的原因
@unittest.skip("无条件跳过")

2）测试用例分析

这个例子通过api创建bond接口，添加删除成员并验证结果。

@unittest.skipIf(
        "lacp" in config.excluded_plugins, "Exclude tests requiring LACP plugin"
    )
    def test_bond_add_member(self):
        # 设置单元测试名称，输出测试结果时会用到
        """Bond add_member/detach member test"""

        self.logger.info("create bond")
        # 通过api创建bond接口，设置为lacp模式
        bond0 = VppBondInterface(
            self, mode=VppEnum.vl_api_bond_mode_t.BOND_API_MODE_LACP, enable_gso=0
        )
        bond0.add_vpp_config()
        # bond接口up起来
        bond0.admin_up()

        # 验证接口添加添加、删除成员2次
        for i in range(2):
            # 验证pg0和pg1不是BondEthernet0的成员，pg接口在setUpClass时创建
            if_dump = self.vapi.sw_member_interface_dump(bond0.sw_if_index)
            self.assertFalse(self.pg0.is_interface_config_in_dump(if_dump))
            self.assertFalse(self.pg1.is_interface_config_in_dump(if_dump))

            # 添加成员pg0和pg1添加到BondEthernet0
            self.logger.info("bond add_member interface pg0 to BondEthernet0")
            bond0.add_member_vpp_bond_interface(
                sw_if_index=self.pg0.sw_if_index, is_passive=0, is_long_timeout=0
            )

            self.logger.info("bond add_member interface pg1 to BondEthernet0")
            bond0.add_member_vpp_bond_interface(
                sw_if_index=self.pg1.sw_if_index, is_passive=0, is_long_timeout=0
            )
            
            # 验证成员已被添加到BondEthernet0
            if_dump = self.vapi.sw_member_interface_dump(bond0.sw_if_index)
            self.assertTrue(self.pg0.is_interface_config_in_dump(if_dump))
            self.assertTrue(self.pg1.is_interface_config_in_dump(if_dump))

            # 删除成员pg0
            self.logger.info("detach interface pg0")
            bond0.detach_vpp_bond_interface(sw_if_index=self.pg0.sw_if_index)

            # 验证成员pg0已经从BondEthernet0删除, 但pg1还在
            if_dump = self.vapi.sw_member_interface_dump(bond0.sw_if_index)
            self.assertFalse(self.pg0.is_interface_config_in_dump(if_dump))
            self.assertTrue(self.pg1.is_interface_config_in_dump(if_dump))

            # 删除成员pg1
            self.logger.info("detach interface pg1")
            bond0.detach_vpp_bond_interface(sw_if_index=self.pg1.sw_if_index)

            # 验证pg0和pg1都已经从BondEthernet0删除
            if_dump = self.vapi.sw_member_interface_dump(bond0.sw_if_index)
            self.assertFalse(self.pg0.is_interface_config_in_dump(if_dump))
            self.assertFalse(self.pg1.is_interface_config_in_dump(if_dump))
        # 删除bond接口
        bond0.remove_vpp_config()

3.1.3.2. test_bond_traffic

该测试用例测试 bond 接口的数据包转发功能，拓扑如下，其中pg接口在setUpClass时创建。

# topology
#
# RX->              TX->
#
# pg2 ------+        +------pg0 (member)
#           |        |
#          BondEthernet0 (10.10.10.1/24)
#           |        |
# pg3 ------+        +------pg1 (memberx)

bond配置：使用XOR模式、L34算法、自定义MAC地址、IP配置10.10.10.1/24

生成测试流量：

从pg2发送数据包到10.10.10.12（通过bond接口转发到pg1）
从pg3发送数据包到10.10.10.11（通过bond接口转发到pg0）

验证结果：

检查bond接口的发送字节数（284字节）
检查pg2和pg3的接收字节数（各142字节）

接着看源码，几个要点：

使用scapy构造数据包
通过pg接口收发包
使用api接口创建bond、添加成功
使用cli查看接口统计信息并做验证

def test_bond_traffic(self):
    """Bond traffic test"""
    bond0_mac = "02:fe:38:30:59:3c"
    # 将MAC地址字符串转换为二进制格式
    mac = MACAddress(bond0_mac).packed
    bond0 = VppBondInterface(
        self,
        mode=VppEnum.vl_api_bond_mode_t.BOND_API_MODE_XOR,
        lb=VppEnum.vl_api_bond_lb_algo_t.BOND_API_LB_ALGO_L34,
        numa_only=0,
        use_custom_mac=1,
        mac_address=mac,
    )
    # 创建bond接口，并up
    bond0.add_vpp_config()
    bond0.admin_up()
    # bond接口配置IP地址
    self.vapi.sw_interface_add_del_address(
        sw_if_index=bond0.sw_if_index, prefix="10.10.10.1/24"
    )

    # 根据ifindex值给pg2和pg3配置IP地址（172.16.xx.1），同时配置ARP表
    # 假设ifindex=3，IP地址为172.16.3.1，确保各接口不会IP冲突
    self.pg2.config_ip4()
    self.pg2.resolve_arp()
    self.pg3.config_ip4()
    self.pg3.resolve_arp()

    self.logger.info(self.vapi.cli("show interface"))
    self.logger.info(self.vapi.cli("show interface address"))
    self.logger.info(self.vapi.cli("show ip neighbors"))

    # 将pg0和pg1添加到bond接口作为成员
    self.logger.info("bond add member interface pg0 to BondEthernet0")
    bond0.add_member_vpp_bond_interface(sw_if_index=self.pg0.sw_if_index)
    self.logger.info("bond add_member interface pg1 to BondEthernet0")
    bond0.add_member_vpp_bond_interface(sw_if_index=self.pg1.sw_if_index)

    # 确认pg0和pg1已添加成员到BondEthernet0
    if_dump = self.vapi.sw_member_interface_dump(bond0.sw_if_index)
    self.assertTrue(self.pg0.is_interface_config_in_dump(if_dump))
    self.assertTrue(self.pg1.is_interface_config_in_dump(if_dump))

    # 通过scapy库，生成测试数据包
    # 路径：pg2 -> BondEthernet0 -> pg1
    # BondEthernet0 TX hashes 计算出该包会到达 pg1
    # 数据包长度：142字节
    p2 = (
        Ether(src=bond0_mac, dst=self.pg2.local_mac)
        / IP(src=self.pg2.local_ip4, dst="10.10.10.12")
        / UDP(sport=1235, dport=1235)
        / Raw(b"\xa5" * 100)
    )
    # 将构建好的数据包添加到pg2接口的发送队列，等待发送
    self.pg2.add_stream(p2)

    # 生成测试数据包 from pg3 -> BondEthernet0 -> pg0
    # 修改数据包的五元组信息，BondEthernet0 TX hashes 计算出该包会到达 pg0
    p3 = (
        Ether(src=bond0_mac, dst=self.pg3.local_mac)
        / IP(src=self.pg3.local_ip4, dst="10.10.10.11")
        / UDP(sport=1234, dport=1234)
        / Raw(b"\xa5" * 100)
    )
    self.pg3.add_stream(p3)

    # 在所有pg接口上启用数据包捕获功能，用于验证数据包的流向
    self.pg_enable_capture(self.pg_interfaces)

    # 在bond接口上配置静态ARP，避免bond接口尝试ARP解析目标IP地址
    self.logger.info(
        self.vapi.cli(
            "set ip neighbor static BondEthernet0 10.10.10.12 abcd.abcd.0002"
        )
    )
    self.logger.info(
        self.vapi.cli(
            "set ip neighbor static BondEthernet0 10.10.10.11 abcd.abcd.0004"
        )
    )

    # 通过cli清空接口统计信息
    self.logger.info(self.vapi.cli("clear interfaces"))
    # 开始发送数据包（pg2和pg3）
    self.pg_start()

    self.logger.info("check the interface counters")

    # 验证统计信息
    # BondEthernet0 发送了284字节（142 * 2）
    intfs = self.vapi.cli("show interface BondEthernet0").split("\n")
    found = 0
    for intf in intfs:
        if "tx bytes" in intf and "284" in intf:
            found = 1
    self.assertEqual(found, 1)

    # 再次验证BondEthernet0 发送了284字节，这里重复验证有两种可能：
    # 1）想确认BondEthernet0没有再发包了
    # 2）历史问题遗留的重复代码
    intfs = self.vapi.cli("show interface BondEthernet0").split("\n")
    found = 0
    for intf in intfs:
        if "tx bytes" in intf and "284" in intf:
            found = 1
    self.assertEqual(found, 1)

    # 验证pg2收到了142字节
    intfs = self.vapi.cli("show interface pg2").split("\n")
    found = 0
    for intf in intfs:
        if "rx bytes" in intf and "142" in intf:
            found = 1
    self.assertEqual(found, 1)

    # 验证pg3收到了142字节
    intfs = self.vapi.cli("show interface pg3").split("\n")
    found = 0
    for intf in intfs:
        if "rx bytes" in intf and "142" in intf:
            found = 1
    self.assertEqual(found, 1)

    #删除bond接口
    bond0.remove_vpp_config()

3.1.4. 清理函数

3.1.4.1. tearDown

在每个测试函数之后调用，进行部分清理。

def tearDown(self):
    # 调用父类的 tearDown，无其他逻辑
    super(TestBondInterface, self).tearDown()

3.1.4.2. tearDownClass

在运行完所有测试函数后调用，以执行最后的清理

def tearDownClass(cls):
    # 调用父类的 tearDownClass，无其他逻辑
    super(TestBondInterface, cls).tearDownClass()

3.2. 断言方法

这里列举出常用的断言方法

3.2.1. 布尔断言

self.assertTrue(expr, msg=None)      # 断言表达式为真
self.assertFalse(expr, msg=None)     # 断言表达式为假

3.2.2. 相等性断言

self.assertEqual(first, second, msg=None)           # 断言两个值相等
self.assertNotEqual(first, second, msg=None)        # 断言两个值不相等
self.assertAlmostEqual(first, second, places=7, msg=None)  # 断言浮点数近似相等
self.assertNotAlmostEqual(first, second, places=7, msg=None)  # 断言浮点数不近似相等

3.2.3. 比较断言

self.assertGreater(first, second, msg=None)         # 断言 first > second
self.assertGreaterEqual(first, second, msg=None)    # 断言 first >= second
self.assertLess(first, second, msg=None)            # 断言 first < second
self.assertLessEqual(first, second, msg=None)       # 断言 first <= second

3.2.4. 成员关系断言

self.assertIn(member, container, msg=None)          # 断言成员在容器中
self.assertNotIn(member, container, msg=None)       # 断言成员不在容器中

3.2.5. 类型断言

self.assertIs(expr1, expr2, msg=None)               # 断言两个对象是同一个
self.assertIsNot(expr1, expr2, msg=None)            # 断言两个对象不是同一个
self.assertIsInstance(obj, cls, msg=None)           # 断言对象是指定类的实例
self.assertNotIsInstance(obj, cls, msg=None)        # 断言对象不是指定类的实例

3.2.6. 异常断言

self.assertRaises(expected_exception, callable, *args, **kwargs)  # 断言调用会抛出指定异常
self.assertRaisesRegex(expected_exception, expected_regex, callable, *args, **kwargs)  # 断言异常消息匹配正则表达式
self.assertWarns(expected_warning, callable, *args, **kwargs)     # 断言调用会产生警告

3.2.7. 序列断言

self.assertSequenceEqual(seq1, seq2, msg=None)      # 断言两个序列相等
self.assertListEqual(list1, list2, msg=None)        # 断言两个列表相等
self.assertTupleEqual(tuple1, tuple2, msg=None)     # 断言两个元组相等
self.assertSetEqual(set1, set2, msg=None)           # 断言两个集合相等
self.assertDictEqual(dict1, dict2, msg=None)        # 断言两个字典相等

3.2.8. 正则表达式断言

self.assertRegex(text, expected_regex, msg=None)    # 断言文本匹配正则表达式
self.assertNotRegex(text, unexpected_regex, msg=None)  # 断言文本不匹配正则表达式

4. 其它说明

ppcli/cli

vapi.cli：只返回命令输出，适合代码内部逻辑处理。

vapi.ppcli：返回“命令+输出”，适合日志、调试、信息展示。

举例：

self.vapi.cli("show interface handoff pg0")只返回命令结果。
self.vapi.ppcli("show interface handoff pg0")返回show interface handoff pg0的命令输出，便于日志查看。

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

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

软件开发

智能高性能网络

网络安全

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

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

登录后参与评论

0 条评论

热度