基于Python的eBPF程序单元测试与Mock实现咨询

问题背景 #

需要对基于BCC的Python-eBPF代码做拆分测试：单独验证Python逻辑和eBPF C代码，核心要解决三个测试场景：Mock内核交互、模拟内核消息、测试异常消息长度。示例代码如下：

from bcc import BPF
# define BPF program
prog = """
int hello(void *ctx) {
    bpf_trace_printk("Hello, World!\n");
    return 0;
}
"""
# load BPF program
b = BPF(text=prog)
b.attach_kprobe(event=b.get_syscall_fnname("clone"), fn_name="hello")
# header
print("%-18s %-16s %-6s %s" % ("TIME(s)", "COMM", "PID", "MESSAGE"))
# format output
while 1:
    try:
        (task, pid, cpu, flags, ts, msg) = b.trace_fields()
    except ValueError:
        continue
    print("%-18.9f %-16s %-6d %s" % (ts, task, pid, msg))

核心测试方案 #

1. Mock内核交互（Python单元测试） #

直接用Python标准库的unittest.mockPatch BCC的BPF类及其核心方法，完全绕开真实内核交互：

• Patch BPF构造函数，避免真的加载eBPF程序到内核
• Mock get_syscall_fnname返回固定字符串（比如__x64_sys_clone）
• Mock attach_kprobe做空实现，无需真挂载kprobe
• 重点Mock trace_fields方法，通过控制返回值模拟内核行为

示例测试代码片段：

import unittest
from unittest.mock import patch, MagicMock
from your_script import BPF  # 替换为你的脚本模块名

class TestBPFScript(unittest.TestCase):
    @patch('your_script.BPF')
    def test_mock_kernel_interaction(self, mock_bpf):
        # 配置Mock对象
        mock_bpf_instance = MagicMock()
        mock_bpf.return_value = mock_bpf_instance
        mock_bpf_instance.get_syscall_fnname.return_value = "__x64_sys_clone"
        
        # 执行脚本逻辑（建议把脚本核心逻辑封装成函数）
        run_your_bpf_script()
        
        # 验证方法调用正确性
        mock_bpf_instance.attach_kprobe.assert_called_once_with(
            event="__x64_sys_clone", fn_name="hello"
        )

2. 模拟内核向用户态发送消息 #

通过Mock trace_fields方法返回预设元组，模拟内核输出的trace数据：

• 让trace_fields返回符合格式的(task, pid, cpu, flags, ts, msg)元组
• 用side_effect多次返回不同值，模拟多条消息
• 若要终止循环，可让它在某次调用时抛出KeyboardInterrupt异常

示例：

@patch('your_script.BPF')
def test_simulate_kernel_messages(self, mock_bpf):
    mock_bpf_instance = MagicMock()
    mock_bpf.return_value = mock_bpf_instance
    
    # 模拟两条内核消息+终止信号
    mock_bpf_instance.trace_fields.side_effect = [
        ("bash", 1234, 0, 0, 162000000.123456789, "Hello, World!\n"),
        ("python", 5678, 1, 0, 162000001.987654321, "Another message\n"),
        KeyboardInterrupt  # 触发循环退出
    ]
    
    # 执行脚本输出逻辑
    with self.assertLogs(level="INFO") as log:
        run_output_loop()
    
    # 验证输出内容
    self.assertIn("162000000.123456789      bash             1234   Hello, World!", log.output[0])

3. 测试异常长度的消息 #

通过Mock trace_fields返回不符合预期格式的数据，测试脚本异常处理逻辑：

• 返回长度不足的元组，触发ValueError，验证脚本是否跳过该消息
• 返回长度过长的元组，测试解包时的异常处理
• 模拟msg字段长度超出/小于预期，测试格式化输出的兼容性

示例：

@patch('your_script.BPF')
def test_abnormal_message_length(self, mock_bpf):
    mock_bpf_instance = MagicMock()
    mock_bpf.return_value = mock_bpf_instance
    
    # 模拟长度不足的元组+有效消息+终止信号
    mock_bpf_instance.trace_fields.side_effect = [
        ("bash", 1234, 0, 0, 162000000.123),  # 缺少msg字段，触发ValueError
        ("python", 5678, 1, 0, 162000001.987, "Valid message\n"),
        KeyboardInterrupt
    ]
    
    # 执行输出循环
    with self.assertLogs(level="INFO") as log:
        run_output_loop()
    
    # 验证仅有效消息被输出
    self.assertEqual(len(log.output), 1)
    self.assertIn("Valid message", log.output[0])

eBPF C代码的独立测试 #

无需加载到内核，直接验证C代码合法性：

• 语法检查：用clang加eBPF目标选项编译，检查语法错误：

  clang -target bpf -c your_bpf_code.c -o /dev/null
  

• BCC编译验证：在单元测试中调用BPF编译方法，不挂载到内核：

  def test_bpf_code_compile(self):
      prog = """
      int hello(void *ctx) {
          bpf_trace_printk("Hello, World!\n");
          return 0;
      }
      """
      # 仅编译，不加载到内核
      b = BPF(text=prog, debug=0)
      # 验证编译成功无异常
      self.assertIsNotNone(b)
  

• 逻辑验证：可使用eBPF模拟器（如bpf_sim）或在虚拟机中加载代码，用trace工具验证输出，这部分属于集成测试范畴。

内容的提问来源于stack exchange，提问作者Kian Reddy