核心内容摘要
通义千问3-Reranker-0.6B保姆级教程：从环境配置到Web服务一键启动
第 20 篇搭建集成测试基础框架 模拟器闭环测试下面我们将一步一步详细完成集成测试基础框架搭建并实现模拟器闭环测试串口收到数据 → 协议解析 → VirtualDevice 状态更新 → 可选返回响应。
目标不依赖真实串口硬件使用 mock 完全控制输入/输出验证 Pelco-D/P 命令到达 VirtualDevice 后状态是否正确变化为后续宏 → 串口 → 模拟器链路打基础
2
1 项目结构调整添加 integration 目录在项目根目录下创建以下结构如果还没有KBD300A/ ├── core/ │ ├── protocol/ │ ├── macro/ │ ├── serial/ │ ├── simulator/ │ └── ... ├── ui/ ├── config/ ├── resources/ ├── tests/ ← 新建 │ ├── unit/ ← 已有的单元测试 │ ├── integration/ ← 集成测试放这里 │ │ ├── test_simulator_loop.py │ │ ├── test_macro_to_serial.py │ │ └── ... │ ├── conftest.py ← 全局 fixture │ └── test_data/ ← 测试用 Pelco 命令样本 └── requirements-dev.txt ← 测试依赖requirements-dev.txt安装这些包Python
7 兼容pytest
6.
5 pytest-mock
3.
1 pytest-qt
4.
2 # 如果涉及 Qt 信号等待 coverage
5 # 可选覆盖率统计安装pipinstall-r requirements-dev.txt
2
2 创建 conftest.py全局 fixturetests/conftest.py# tests/conftest.pyimportpytestfromunittest.mockimportMagicMockfromcore.simulator.virtual_deviceimportVirtualDevicefromcore.protocolimportget_protocolfromcore.serial.managerimportSerialManagerpytest.fixturedefvirtual_device():每个测试一个干净的虚拟设备returnVirtualDevice(cam_id
pytest.fixturedefmock_serial():模拟 serial.Serial 对象mock_serMagicMock()mock_ser.is_openTruemock_ser.in_waiting0mock_ser.read.side_effectlambdasize:b# 默认返回空mock_ser.writeMagicMock()returnmock_serpytest.fixturedefmock_serial_manager(mock_serial):模拟 SerialManager注入 mock 串口managerSerialManager(portCOMfake,baud9600,protocolD)manager._sermock_serial# 直接替换底层串口manager.is_openTruereturnmanagerpytest.fixturedefpelco_d_protocol(mock_serial_manager):Pelco-D 协议实例returnget_protocol(mock_serial_manager,D)pytest.fixturedefpelco_p_protocol(mock_serial_manager):Pelco-P 协议实例returnget_protocol(mock_serial_manager,P)
2
3 准备测试数据Pelco 命令样本在tests/test_data/pelco_commands.json中保存一些典型命令方便复用{pelco_d:{ptz_right_fast:{bytes:FF0100001020004F,description:地址1右转速度32,expected_pan:
3
0},call_preset_5:{bytes:FF010000071F0067,description:调用预置位5,preset_id:5},aux_on_2:{bytes:FF0100000B02000E,description:AUX2 开}},pelco_p:{ptz_up_medium:{bytes:A0010008100800AF,description:地址1上仰速度8}}}加载方式示例importjsonfrompathlibimportPath TEST_DATAjson.loads(Path(tests/test_data/pelco_commands.json).read_text())
2
4 实现模拟器闭环测试核心文件tests/integration/test_simulator_loop.py# tests/integration/test_simulator_loop.pyimportpytestimporttimefromunittest.mockimportcallfromcore.serial.workerimportSerialWorkerfromcore.protocol.pelco_dimportPelcoDProtocolfromcore.simulator.virtual_deviceimportVirtualDevicefromtests.conftestimportmock_serial_manager# 如果需要defload_command_hex(hex_str):将 hex 字符串转为 bytesreturnbytes.fromhex(hex_str)pytest.mark.integrationdeftest_pelco_d_ptz_command_updates_virtual_device(virtual_device,mock_serial,pelco_d_protocol): 测试收到 Pelco-D 右转命令 → VirtualDevice pan 角度增加 # 准备输入右转速度 32 的命令cmd_hexFF 01 00 00 20 00 21# 校验和 0x21 100320cmd_bytesload_command_hex(cmd_hex.replace( ,))# 模拟 SerialWorker 收到数据直接注入 bufferworkerSerialWorker(portCOMfake,baud9600,protocolD)worker._sermock_serial worker._bufferbytearray(cmd_bytes)# 关键注入 VirtualDevice 到协议层需要小改 protocol 代码支持# 临时方案在测试中 monkey-patch 或直接调用 processpelco_d_protocol.virtual_devicevirtual_device# 假设已添加属性# 模拟一次数据处理实际项目中可调用 worker._process_buffer()parsedpelco_d_protocol.parse_response(cmd_bytes)# 或者更真实模拟 worker 处理流程worker._buffer.extend(cmd_bytes)worker._process_buffer()# 如果 _process_buffer 是 public 或可访问# 检查 VirtualDevice 状态statusvirtual_device.get_status_dict()assertpaninstatus pan_valuefloat(status[pan].rstrip(°))assertpan_value0,右转命令应使 pan 角度增加# 可选验证具体角度根据 VirtualDevice 内部实现比例# 假设 VirtualDevice 将 speed 32 映射为 32 度assertabs(pan_value-
32.
0)
0,f预期 pan ≈32实际{pan_value}pytest.mark.integrationdeftest_pelco_d_call_preset_triggers_action(virtual_device,mock_serial,pelco_d_protocol):测试调用预置位命令是否触发 VirtualDevice 相应逻辑cmd_hexFF 01 00 00 07 05 0D# 调用预置位 5校验和 10075130x0Dcmd_bytesload_command_hex(cmd_hex.replace( ,))# 注入pelco_d_protocol.virtual_devicevirtual_device# 处理命令workerSerialWorker(portCOMfake,baud9600,protocolD)worker._sermock_serial worker._bufferbytearray(cmd_bytes)worker._process_buffer()# 或直接调用协议 parse# 假设 VirtualDevice 在 process_command 中记录 preset 调用# 你可以添加断言例如检查日志或状态# 或者临时添加一个调用计数器属性到 VirtualDeviceasserthasattr(virtual_device,last_preset_called)assertvirtual_device.last_preset_called5pytest.mark.integrationdeftest_pelco_p_aux_on_updates_aux_state(virtual_device,mock_serial):Pelco-P 打开 AUX2 测试cmd_hexA0 01 00 0B 02 00 0E AF# AUX2 ONcmd_bytesload_command_hex(cmd_hex.replace( ,))protocolPelcoDProtocol(mock_serial)# 改用 P 协议protocol.virtual_devicevirtual_device# 处理parsedprotocol._parse_response(cmd_bytes)assertparsed.get(type)auxassertparsed.get(aux_id)2assertparsed.get(state)on# 验证虚拟设备状态assertvirtual_device.aux_states[2]isTrue
2
5 小幅修改生产代码以便测试在core/simulator/virtual_device.py中添加classVirtualDevice:def__init__(self,cam_id:int
:...self.last_preset_calledNone# 测试用defprocess_command(self,data:bytes)-Optional[bytes]:parsedparse_pelco_packet(data)ifparsed.get(type)presetandparsed.get(operation)call:self.last_preset_calledparsed.get(preset_id)# ... 原有逻辑在core/protocol/base.py或具体协议类中添加classProtocolBase(ABC):def__init__(self,serial_mgr):self.serial_mgrserial_mgr self.virtual_deviceNone# 测试时注入
2