xs
43 lines
1.1 KiB
Python
43 lines
1.1 KiB
Python
"""py_srv 单元测试:同进程 spin + 验证 Service 通信。"""
|
|
|
|
import time
|
|
|
|
import rclpy
|
|
import pytest
|
|
|
|
from py_srv.add_two_ints_server import AddTwoIntsServer
|
|
from py_srv.add_two_ints_client import AddTwoIntsClient
|
|
from example_interfaces.srv import AddTwoInts
|
|
|
|
|
|
@pytest.fixture(scope='module')
|
|
def ros_context():
|
|
rclpy.init()
|
|
yield
|
|
rclpy.shutdown()
|
|
|
|
|
|
def test_service_inproc_roundtrip(ros_context):
|
|
"""同进程内 server + client 互调,验证 a + b == sum。"""
|
|
server = AddTwoIntsServer()
|
|
client = AddTwoIntsClient()
|
|
|
|
exec_ = rclpy.executors.SingleThreadedExecutor()
|
|
exec_.add_node(server)
|
|
exec_.add_node(client)
|
|
|
|
req = AddTwoInts.Request()
|
|
req.a = 7
|
|
req.b = 35
|
|
|
|
future = client._client.call_async(req)
|
|
end = time.time() + 3.0
|
|
while not future.done() and time.time() < end:
|
|
exec_.spin_once(timeout_sec=0.05)
|
|
|
|
assert future.done(), 'service call did not complete in 3s'
|
|
assert future.result() is not None
|
|
assert future.result().sum == 42
|
|
|
|
exec_.remove_node(server)
|
|
exec_.remove_node(client) |