返回教程正文

配套源码

runtime_edge_cases.py

app/tests/runtime_edge_cases.py
ros_gateway Brick：在 App Lab 中建立可靠的 WebSocket 通道app/tests/runtime_edge_cases.py
Python182 行
  1. # SPDX-License-Identifier: MIT
  3. import argparse
  4. from contextlib import contextmanager
  5. import json
  6. import time
  8. from websockets.exceptions import ConnectionClosed
  9. from websockets.sync.client import connect
  12. def send_json(websocket, message):
  13. """
  14. @description : 向服务端发送紧凑 JSON 文本
  15. @param websocket : 已建立的 WebSocket 客户端连接
  16. @param message : 待发送消息字典
  17. @return : 无返回值
  18. """
  19. websocket.send(json.dumps(message, separators=(",", ":")))
  22. def receive_json(websocket, timeout=3.0):
  23. """
  24. @description : 接收并解析一条服务端 JSON 消息
  25. @param websocket : 已建立的 WebSocket 客户端连接
  26. @param timeout : 接收超时秒数
  27. @return : 解析后的消息字典
  28. """
  29. return json.loads(websocket.recv(timeout=timeout))
  32. @contextmanager
  33. def open_owner(url, node):
  34. """
  35. @description : 建立连接并完成 ROS 2 客户端握手
  36. @param url : WebSocket 服务地址
  37. @param node : 测试客户端节点名称
  38. @return : 产生已完成握手连接的上下文管理器
  39. """
  40. with connect(url, open_timeout=3.0, close_timeout=2.0) as websocket:
  41. send_json(
  42. websocket,
  43. {
  44. "version": 1,
  45. "type": "hello",
  46. "role": "ros2",
  47. "node": node,
  48. },
  49. )
  50. hello = receive_json(websocket)
  51. if hello.get("type") != "hello" or hello.get("role") != "app":
  52. raise RuntimeError("server hello was not received")
  53. yield websocket
  56. def test_wrong_path(base_url):
  57. """
  58. @description : 验证非 /ros 路径会被策略关闭
  59. @param base_url : 正确 WebSocket 服务地址
  60. @return : 关闭码为 1008 时返回 True
  61. """
  62. wrong_url = base_url.rsplit("/", 1)[0] + "/wrong"
  63. try:
  64. with connect(wrong_url, open_timeout=3.0, close_timeout=2.0) as websocket:
  65. websocket.recv(timeout=2.0)
  66. except ConnectionClosed as exc:
  67. return exc.code == 1008
  68. return False
  71. def test_protocol_errors(url):
  72. """
  73. @description : 验证非法 JSON 与非递增序号返回结构化错误
  74. @param url : WebSocket 服务地址
  75. @return : 两个错误路径均正确时返回 True
  76. """
  77. with open_owner(url, "edge_case_protocol") as websocket:
  78. websocket.send("{")
  79. invalid_json = receive_json(websocket)
  81. send_json(
  82. websocket,
  83. {
  84. "version": 1,
  85. "type": "heartbeat",
  86. "seq": 10,
  87. "timestamp_ms": time.time_ns() // 1_000_000,
  88. },
  89. )
  90. send_json(
  91. websocket,
  92. {
  93. "version": 1,
  94. "type": "heartbeat",
  95. "seq": 10,
  96. "timestamp_ms": time.time_ns() // 1_000_000,
  97. },
  98. )
  100. deadline = time.monotonic() + 3.0
  101. sequence_error = None
  102. while time.monotonic() < deadline:
  103. message = receive_json(websocket, max(0.01, deadline - time.monotonic()))
  104. if message.get("type") == "error" and message.get("code") == "non_monotonic_seq":
  105. sequence_error = message
  106. break
  108. return (
  109. invalid_json.get("type") == "error"
  110. and invalid_json.get("code") == "invalid_json"
  111. and sequence_error is not None
  112. )
  115. def test_single_owner(url):
  116. """
  117. @description : 验证第二个客户端不能抢占已握手客户端
  118. @param url : WebSocket 服务地址
  119. @return : 第二连接收到 1013 关闭码时返回 True
  120. """
  121. with open_owner(url, "edge_case_owner"):
  122. try:
  123. with connect(url, open_timeout=3.0, close_timeout=2.0) as contender:
  124. contender.recv(timeout=2.0)
  125. except ConnectionClosed as exc:
  126. return exc.code == 1013
  127. return False
  130. def test_heartbeat_timeout_and_reconnect(url):
  131. """
  132. @description : 验证无应用层心跳会断开，且随后能够重新握手
  133. @param url : WebSocket 服务地址
  134. @return : 超时关闭和再次握手均成功时返回 True
  135. """
  136. timeout_closed = False
  137. with open_owner(url, "edge_case_timeout") as websocket:
  138. deadline = time.monotonic() + 5.0
  139. while time.monotonic() < deadline:
  140. try:
  141. websocket.recv(timeout=max(0.01, deadline - time.monotonic()))
  142. except ConnectionClosed as exc:
  143. timeout_closed = exc.code == 1008
  144. break
  146. with open_owner(url, "edge_case_reconnect"):
  147. reconnected = True
  148. return timeout_closed and reconnected
  151. def run(url):
  152. """
  153. @description : 顺序执行 WebSocket 运行时边界测试
  154. @param url : WebSocket 服务地址
  155. @return : 测试名称到布尔结果的字典
  156. """
  157. return {
  158. "wrong_path_rejected": test_wrong_path(url),
  159. "protocol_errors": test_protocol_errors(url),
  160. "single_owner": test_single_owner(url),
  161. "heartbeat_timeout_and_reconnect": test_heartbeat_timeout_and_reconnect(url),
  162. }
  165. def main():
  166. """
  167. @description : 解析命令行参数并根据边界测试结果设置退出码
  168. @param : 无参数
  169. @return : 无返回值
  170. """
  171. parser = argparse.ArgumentParser(description="Test ROS Gateway runtime edge cases")
  172. parser.add_argument("--url", default="ws://127.0.0.1:8765/ros")
  173. arguments = parser.parse_args()
  174. results = run(arguments.url)
  175. print(json.dumps(results, ensure_ascii=False, indent=2))
  176. if not all(results.values()):
  177. raise SystemExit(1)
  180. if __name__ == "__main__":
  181. main()