📚 课程目录第4章核心测试场景实现

4、核心测试场景实现：服务发现与通信质量验证

这一章，我们直接切入SOME/IP自动化测试中最核心的四个场景。说白了，就是验证你的ECU能不能“找到人”、“说上话”、“不丢消息”、“数据正确”。

我刚开始搭建测试平台时，总想着把用例写得越全越好。后来发现，把这四个场景吃透，80%的回归问题都能提前拦住。咱们一个一个来。

4.1 服务发现测试：Offer与Subscribe的博弈

服务发现是SOME/IP的“敲门砖”。客户端得先知道谁提供了什么服务，才能发起后续通信。测试的核心，就是验证Offer/Subscribe的时序和内容是否正确。

核心验证点：

服务端是否在正确端口发送OfferService报文
客户端收到Offer后，是否发送SubscribeEventgroup
服务端对Subscribe的Ack/Nack响应是否合规
多客户端场景下，服务发现是否互不干扰

我个人习惯用scapy构造报文，然后结合python-someip库做状态机校验。下面是一个简化的测试用例框架：

import someip
import scapy.all as scapy

def test_service_discovery_offer_subscribe():
    # 1. 监听SOME/IP SD报文
    sd_sniffer = scapy.AsyncSniffer(iface="eth0",
                                    filter="udp port 30490",
                                    prn=handle_sd_packet)
    sd_sniffer.start()
    
    # 2. 等待服务端发送OfferService
    offer = wait_for_offer(service_id=0x1234, timeout=5.0)
    assert offer is not None, "服务端未发送OfferService"
    
    # 3. 构造并发送SubscribeEventgroup
    subscribe = build_subscribe(service_id=0x1234,
                                instance_id=0x0001,
                                eventgroup_id=0x0001)
    send_sd_packet(subscribe)
    
    # 4. 验证服务端回复SubscribeAck
    ack = wait_for_subscribe_ack(service_id=0x1234, timeout=3.0)
    assert ack is not None, "未收到SubscribeAck"
    
    # 5. 清理
    sd_sniffer.stop()
    print("服务发现测试通过")

避坑指南：我曾经遇到过一个问题——测试环境里有两个ECU同时发送OfferService，我们的客户端只处理了第一个，导致第二个服务的订阅全部失败。后来我在测试用例里加了“多Offer并发处理”的校验，才把这个bug抓出来。

4.2 Method调用超时测试：别让客户端死等

Method调用超时，是分布式系统里最常见的故障之一。服务端挂了、网络延迟了、或者处理线程卡住了，客户端不能无限等下去。

测试思路其实很简单：构造一个“慢响应”或“无响应”的服务端，然后看客户端是否在预期时间内返回超时错误。

测试场景	模拟方式	预期结果
服务端无响应	不回复Method响应报文	客户端抛出TimeoutException
服务端延迟响应	延迟3秒后回复（超时阈值设为2秒）	客户端超时，丢弃延迟响应
服务端返回错误码	回复E_NOT_OK	客户端正确处理错误码
网络丢包	iptables模拟丢包率10%	客户端重试后超时

def test_method_call_timeout():
    # 配置客户端超时时间为2秒
    client = SomeIpClient(timeout=2.0)
    
    # 构造一个Method请求
    request = build_method_request(service_id=0x1234,
                                   method_id=0x0001,
                                   payload=b'\x01\x02\x03')
    
    # 发送请求，但不模拟服务端回复
    start_time = time.time()
    try:
        response = client.call_method(request)
        # 如果走到这里，说明测试失败
        assert False, "应该超时但收到了响应"
    except TimeoutError:
        elapsed = time.time() - start_time
        # 验证超时时间在合理范围内（1.8s ~ 2.5s）
        assert 1.8 <= elapsed <= 2.5, f"超时时间异常: {elapsed}s"
        print(f"Method调用超时测试通过，耗时{elapsed:.2f}s")

注意：超时测试一定要考虑系统时钟精度和调度延迟。我一般允许±20%的误差范围，否则测试在CI环境里会频繁误报。

4.3 Event通知频率测试：别让总线爆炸

Event通知是SOME/IP里最“吵”的通信模式。如果服务端以100Hz的频率狂发Event，而客户端只能处理10Hz，那缓冲区很快就会溢出。

测试目标很明确：验证服务端是否按照配置的频率发送Event，以及客户端在超频情况下是否稳定。

测试步骤：

订阅一个Eventgroup，记录服务端配置的通知频率（如20Hz）
在5秒内捕获所有Event报文
计算实际接收频率 = 报文总数 / 5秒
验证实际频率在配置频率的±10%范围内
模拟服务端以2倍频率发送Event，验证客户端是否丢包或崩溃

def test_event_notification_frequency():
    # 订阅Event
    subscriber = subscribe_event(service_id=0x1234,
                                 eventgroup_id=0x0001)
    
    # 捕获5秒内的Event报文
    packets = []
    def capture(pkt):
        packets.append(pkt)
    
    sniffer = start_sniffer(callback=capture, timeout=5.0)
    sniffer.join()
    
    # 计算频率
    actual_freq = len(packets) / 5.0
    expected_freq = 20.0  # 配置频率
    
    # 允许±10%误差
    lower_bound = expected_freq * 0.9
    upper_bound = expected_freq * 1.1
    assert lower_bound <= actual_freq <= upper_bound, \
        f"Event频率异常: 期望{expected_freq}Hz, 实际{actual_freq:.2f}Hz"
    
    print(f"Event通知频率测试通过: {actual_freq:.2f}Hz")

个人经验：有一次我在测试中发现Event频率忽高忽低，排查了半天，最后发现是服务端的定时器用了系统默认的HZ=100，而我们的测试机HZ=250。两个系统的时间基准不一致，导致频率计算偏差。从那以后，我都在测试脚本里先校准时间基准。

4.4 Field属性读写测试：数据完整性的最后防线

Field是SOME/IP里比较特殊的概念——它既有Getter/Setter方法，又有Notifier事件。测试Field，本质上是在验证“读、写、通知”三条路径的数据一致性。

测试项	操作	验证点
Getter	发送Get请求	返回的Field值与服务端当前值一致
Setter	发送Set请求，写入新值	服务端更新成功，返回确认
Notifier	订阅Field变更通知	服务端值变化后，客户端收到通知
边界值	写入最大值/最小值/空值	服务端正确处理，不崩溃

def test_field_read_write():
    # 1. 读取Field初始值
    initial_value = read_field(service_id=0x1234,
                               instance_id=0x0001,
                               field_id=0x0001)
    print(f"初始值: {initial_value}")
    
    # 2. 写入新值
    new_value = b'\x00\x00\x00\x2A'  # 42
    write_response = write_field(service_id=0x1234,
                                 instance_id=0x0001,
                                 field_id=0x0001,
                                 value=new_value)
    assert write_response.return_code == 0x00, "写入失败"
    
    # 3. 再次读取，验证一致性
    read_back = read_field(service_id=0x1234,
                           instance_id=0x0001,
                           field_id=0x0001)
    assert read_back == new_value, \
        f"读写不一致: 写入{new_value}, 读回{read_back}"
    
    # 4. 验证Notifier通知
    notification = wait_for_notification(service_id=0x1234,
                                         field_id=0x0001,
                                         timeout=2.0)
    assert notification is not None, "未收到Field变更通知"
    assert notification.value == new_value, "通知内容与写入值不一致"
    
    print("Field属性读写测试通过")

特别提醒：Field的Setter操作可能会触发服务端的业务逻辑（比如校验、计算、联动其他Field）。我建议在测试用例里加入“写入非法值”的负面测试，看看服务端会不会因为异常输入而崩溃。我曾经就遇到过——写入一个负数，服务端直接段错误了。

4.5 本章知识体系总览

为了让你更直观地理解这四个测试场景的关系，我画了一张图：

这四个场景，其实覆盖了SOME/IP通信的“建立连接”、“请求响应”、“持续通知”、“数据操作”四个维度。你想想看，只要把这四个维度测透了，大部分通信层面的问题都能提前暴露。