返回教程正文

配套源码

test_websocket_client.py

ros2/test/test_websocket_client.py
ros_gateway Brick：在 App Lab 中建立可靠的 WebSocket 通道ros2/test/test_websocket_client.py
Python113 行
  1. # SPDX-License-Identifier: MIT
  3. import json
  4. import unittest
  5. from unittest.mock import Mock
  7. from ventuno_app_bridge.websocket_client import WebSocketBridgeClient
  10. class TestWebSocketBridgeClient(unittest.TestCase):
  11. """验证 ROS 2 WebSocket 客户端的本地队列与协议编码。"""
  13. def setUp(self):
  14. """
  15. @description : 为每个测试创建相互隔离的客户端和回调记录器
  16. @param : 无参数
  17. @return : 无返回值
  18. """
  19. self.messages = []
  20. self.connection_changes = []
  21. self.logs = []
  22. self.client = WebSocketBridgeClient(
  23. websocket_url="ws://127.0.0.1:8765/ros",
  24. reconnect_interval=0.1,
  25. heartbeat_interval=1.0,
  26. command_timeout=0.3,
  27. message_callback=self.messages.append,
  28. connection_callback=self.connection_changes.append,
  29. log_callback=lambda level, message: self.logs.append((level, message)),
  30. )
  32. def test_latest_velocity_replaces_older_command(self):
  33. """
  34. @description : 验证速度队列仅保留最新命令并编码协议公共字段
  35. @param : 无参数
  36. @return : 无返回值
  37. """
  38. websocket = Mock()
  39. self.client.send_cmd_vel(0.1, 0.2, 0.3)
  40. self.client.send_cmd_vel(-0.4, 0.5, -0.6)
  42. self.client._send_latest_command(websocket)
  44. websocket.send.assert_called_once()
  45. message = json.loads(websocket.send.call_args.args[0])
  46. self.assertEqual(message["version"], 1)
  47. self.assertEqual(message["type"], "cmd_vel")
  48. self.assertEqual(message["seq"], 1)
  49. self.assertEqual(message["vx"], -0.4)
  50. self.assertEqual(message["vy"], 0.5)
  51. self.assertEqual(message["wz"], -0.6)
  53. def test_stale_velocity_is_not_sent(self):
  54. """
  55. @description : 验证超过本地时限的速度命令会被丢弃
  56. @param : 无参数
  57. @return : 无返回值
  58. """
  59. websocket = Mock()
  60. self.client._latest_command.put_nowait(
  61. {
  62. "timestamp_ms": self.client._now_ms() - 1000,
  63. "vx": 0.1,
  64. "vy": 0.0,
  65. "wz": 0.0,
  66. }
  67. )
  69. self.client._send_latest_command(websocket)
  71. websocket.send.assert_not_called()
  72. self.assertTrue(any("stale local cmd_vel" in item[1] for item in self.logs))
  74. def test_connection_callback_only_reports_changes(self):
  75. """
  76. @description : 验证连接状态回调不会重复发布相同状态
  77. @param : 无参数
  78. @return : 无返回值
  79. """
  80. self.client._set_connected(True)
  81. self.client._set_connected(True)
  82. self.client._set_connected(False)
  83. self.client._set_connected(False)
  85. self.assertEqual(self.connection_changes, [True, False])
  87. def test_mode_queue_is_bounded(self):
  88. """
  89. @description : 验证控制请求队列达到上限后拒绝继续增长
  90. @param : 无参数
  91. @return : 无返回值
  92. """
  93. results = [self.client.request_mode("ROS_TELEOP") for _ in range(9)]
  95. self.assertEqual(results, ([True] * 8) + [False])
  96. self.assertEqual(self.client._control_queue.qsize(), 8)
  98. def test_receive_requires_json_object(self):
  99. """
  100. @description : 验证服务端消息根节点必须是 JSON 对象
  101. @param : 无参数
  102. @return : 无返回值
  103. """
  104. websocket = Mock()
  105. websocket.recv.return_value = "[]"
  107. with self.assertRaisesRegex(RuntimeError, "JSON root"):
  108. self.client._receive_json(websocket, timeout=0.1)
  111. if __name__ == "__main__":
  112. unittest.main()