集成测试参考
面向系统测试与集成开发的完整 grpcurl 用例库:每条用例含命令、预期响应与判定标准。
前置:完成 01 §1.1 环境变量导出。
export REPO=/path/to/autonomy
export BRIDGE=127.0.0.1:5005
export PROTO_OPTS="-import-path $REPO -proto autonomy/bridge/proto/external_command_service.proto"
export SVC=autonomy.bridge.proto.AutonomyService
new_cmd_id() { echo "test-$(date +%s)-$RANDOM"; }
# JSON 请求体:heredoc + -d @-(标准输入,无需转义)
grpc_call() {
grpcurl -plaintext $PROTO_OPTS -d @- "$BRIDGE" "$SVC/$1"
}
13.1 服务发现
服务发现与 schema 查看 → 01 §1.2(list / describe 命令与判定标准)。
13.2 响应判读约定
用例含 命令、预期响应 JSON 与 判定。Command Stream 以末帧 ack.final=true 结束。
类型 |
结束判据 |
成功判据 |
|---|---|---|
Query / System Unary |
单条 JSON |
字段符合下表;System 看 |
Command Stream |
|
末帧 |
Push Stream |
无末帧,Ctrl+C 结束 |
周期输出且字段非空 |
TaskType 整型:导航=1 · 跟随=2 · 遥操=3 · 探索=4 · 对接=5 · 地图=6
TaskStatus 整型:IDLE=1 · RUNNING=2 · PAUSED=3 · SUCCEEDED=4 · FAILED=5 · CANCELED=6
每条 TC-* 用例采用与 Command 页相同的卡片:上方 发送指令(蓝)、下方 收到结果(绿);纯步骤/判定类用例仅展示绿色结果区。
13.3 Query 测试用例
TC-Q-001 GetCapabilities — 能力探测
TC-Q-002 GetRobotSnapshot — 空闲快照
TC-Q-003 GetActiveTask — 无活跃任务
TC-Q-004 GetActiveTask — 导航进行中
先执行 TC-NAV-001 且在 Stream 未结束前,另开终端:
13.4 Stream 测试用例
TC-S-001 ReceiveBotStates — 周期状态
TC-S-002 ReceiveBotEvents — 任务生命周期事件
TC-S-003 Snapshot 与 Stream 一致性
终端 A 订阅 ReceiveBotStates,终端 B 调用 GetRobotSnapshot:
13.5 System 测试用例
TC-SYS-001 EmergencyStop — 空闲时急停
TC-SYS-002 EmergencyStop — 导航中抢占
终端 A 发起 TC-NAV-001(保持 Stream 打开)
终端 B 执行 TC-SYS-001
TC-SYS-003 CancelAllTasks — 取消全部
TC-SYS-004 CancelAllTasks — 按类型取消
13.7 SendExplorationCommand 测试用例
TC-EXP-001 START 探索
TC-EXP-002 PAUSE / RESUME
TC-EXP-003 SAVE_MAP
TC-EXP-004 SET_AREA 限定区域
13.8 SendFollowCommand 测试用例
TC-FOL-001 跟随行人模式 START
TC-FOL-002 跟随目标位姿
TC-FOL-003 UPDATE_TARGET 动态更新
TC-FOL-004 目标丢失
目标离开视野且 reacquire_on_lost=false:
13.9 SendTeleopCommand 测试用例
grpcurl 对 Bidi Stream 能力有限;完整测试请用 Python(见 §13.14)。
TC-TEL-001 START 探测
TC-TEL-002 VELOCITY 连续发送(Python)
TC-TEL-003 STOP
13.10 SendDockCommand 测试用例
TC-DCK-001 按站点 ID 对接
TC-DCK-002 按位姿对接
TC-DCK-003 UNDOCK
13.11 SendMapCommand 测试用例
TC-MAP-001 LOAD 地图
TC-MAP-002 SWITCH 地图
TC-MAP-003 SET_INITIAL_POSE 重定位
TC-MAP-004 CLEAR_COSTMAP
TC-MAP-005 ADD / REMOVE 禁行区
13.12 端到端集成场景
E2E-001 标准导航闭环
步骤 |
RPC |
预期 |
|---|---|---|
1 |
|
|
2 |
|
|
3 |
|
Stream 进度帧 |
4 |
|
|
5 |
|
|
6 |
Stream 末帧 |
|
7 |
|
|
E2E-002 急停打断导航
步骤 |
RPC |
预期 |
|---|---|---|
1 |
START 导航 |
|
2 |
|
|
3 |
导航 Stream |
末帧 |
4 |
|
|
5 |
|
|
E2E-003 探索 → 保存地图 → 加载导航
步骤 |
RPC |
预期 |
|---|---|---|
1 |
|
|
2 |
|
|
3 |
|
|
4 |
|
在新地图上 |
E2E-004 重定位后导航
步骤 |
RPC |
预期 |
|---|---|---|
1 |
|
|
2 |
|
|
3 |
|
正常规划 |
13.13 一键冒烟序列(bash)
复制整段执行;每步打印 [OK] / [FAIL](需 Bridge 运行中):
set -e
REPO=${REPO:-/path/to/autonomy}
BRIDGE=${BRIDGE:-127.0.0.1:5005}
PROTO_OPTS="-import-path $REPO -proto autonomy/bridge/proto/external_command_service.proto"
SVC=autonomy.bridge.proto.AutonomyService
run() { echo ">>> $1"; shift; "$@" && echo "[OK] $1" || echo "[FAIL] $1"; }
run "list services" grpcurl -plaintext $PROTO_OPTS $BRIDGE list
run "GetCapabilities" grpcurl -plaintext $PROTO_OPTS -d '{}' $BRIDGE $SVC/GetCapabilities
run "GetRobotSnapshot" grpcurl -plaintext $PROTO_OPTS -d '{}' $BRIDGE $SVC/GetRobotSnapshot
run "GetActiveTask" grpcurl -plaintext $PROTO_OPTS -d '{}' $BRIDGE $SVC/GetActiveTask
CID="smoke-$(date +%s)"
run "Nav STOP" grpcurl -plaintext $PROTO_OPTS -d @- $BRIDGE $SVC/SendNavigationCommand <<EOF
{
"header": { "cmd_id": "${CID}-stop" },
"command": 2
}
EOF
run "EmergencyStop" grpcurl -plaintext $PROTO_OPTS -d @- $BRIDGE $SVC/EmergencyStop <<EOF
{
"header": { "cmd_id": "${CID}-estop" },
"reason": "smoke"
}
EOF
run "CancelAllTasks" grpcurl -plaintext $PROTO_OPTS -d @- $BRIDGE $SVC/CancelAllTasks <<EOF
{
"header": { "cmd_id": "${CID}-cancel" }
}
EOF
echo "Smoke done. For START nav / Stream subscribe, see sections 13.6 / 13.4."
13.14 Python 集成测试脚本
依赖:pip install grpcio grpcio-tools
#!/usr/bin/env python3
"""AutonomyService 集成测试示例 — 覆盖 Query / Nav STOP / Stream 采样。"""
import sys
import time
import grpc
from google.protobuf import empty_pb2
# 假设已 protoc 生成到 PYTHONPATH
from autonomy.bridge.proto import external_command_service_pb2 as pb
from autonomy.bridge.proto import external_command_service_pb2_grpc as stubs
BRIDGE = "127.0.0.1:5005"
CLIENT_ID = "integration-test"
def main() -> int:
ch = grpc.insecure_channel(BRIDGE)
stub = stubs.AutonomyServiceStub(ch)
# TC-Q-001
caps = stub.GetCapabilities(empty_pb2.Empty())
print("Capabilities:", caps.bridge_version, "nav=", caps.supports_navigation)
assert caps.bridge_version, "bridge_version empty"
# TC-Q-003
task = stub.GetActiveTask(empty_pb2.Empty())
print("ActiveTask:", task.type, task.status)
# TC-NAV-004 STOP
cmd_id = f"py-{int(time.time())}"
final = None
for resp in stub.SendNavigationCommand(pb.NavigationCommandRequest(
header=pb.RequestHeader(cmd_id=cmd_id, client_id=CLIENT_ID),
command=pb.NAV_CMD_STOP,
)):
print("Nav STOP ack:", resp.ack.success, resp.ack.final, resp.status)
if resp.ack.final:
final = resp
break
assert final and final.ack.final, "missing final frame"
# TC-S-001 采样 3 帧
stream = stub.ReceiveBotStates(empty_pb2.Empty())
for i in range(3):
state = next(stream)
print(f"State[{i}]: bat={state.battery_percent:.0f}% task={state.active_task_type}")
# TC-SYS-001
ack = stub.EmergencyStop(pb.EmergencyStopRequest(
header=pb.RequestHeader(cmd_id=f"estop-{cmd_id}", client_id=CLIENT_ID),
reason="python integration test",
))
print("Estop:", ack.success, ack.message)
assert ack.success
print("ALL PASSED")
return 0
if __name__ == "__main__":
sys.exit(main())
Teleop Bidi 示例(TC-TEL-002):
from autonomy.commsgs.proto import geometry_msgs_pb2
def teleop_velocity_test(stub, duration_sec=3.0):
cmd_id = f"teleop-{int(time.time())}"
def requests():
yield pb.TeleopCommandRequest(
header=pb.RequestHeader(cmd_id=cmd_id, client_id=CLIENT_ID),
command=pb.TELEOP_CMD_START,
watchdog_timeout_sec=1.0,
max_linear_speed=0.3,
)
t0 = time.time()
while time.time() - t0 < duration_sec:
yield pb.TeleopCommandRequest(
header=pb.RequestHeader(cmd_id=cmd_id, client_id=CLIENT_ID),
command=pb.TELEOP_CMD_VELOCITY,
velocity=geometry_msgs_pb2.TwistStamped(
twist=geometry_msgs_pb2.Twist(
linear=geometry_msgs_pb2.Vector3(x=0.1),
)
),
)
time.sleep(0.1)
yield pb.TeleopCommandRequest(
header=pb.RequestHeader(cmd_id=cmd_id, client_id=CLIENT_ID),
command=pb.TELEOP_CMD_STOP,
)
for resp in stub.SendTeleopCommand(requests()):
if resp.ack.final:
assert resp.status == pb.TELEOP_STATUS_IDLE or resp.ack.success
break
13.15 用例索引
分类 |
用例 ID |
章节 |
|---|---|---|
Query |
TC-Q-001 … 004 |
§13.3 |
Stream |
TC-S-001 … 003 |
§13.4 |
System |
TC-SYS-001 … 004 |
§13.5 |
Navigation |
TC-NAV-001 … 006, TC-NAV-NEG-001 … 003 |
§13.6 |
Exploration |
TC-EXP-001 … 004 |
§13.7 |
Follow |
TC-FOL-001 … 004 |
§13.8 |
Teleop |
TC-TEL-001 … 003 |
§13.9 |
Dock |
TC-DCK-001 … 003 |
§13.10 |
Map |
TC-MAP-001 … 005 |
§13.11 |
E2E |
E2E-001 … 004 |
§13.12 |