[Jazzy] Fix mypy errors (backport #1084)

rosapi/src/rosapi/params.py

@@ -141,7 +141,7 @@ async def _set_param(
 
     assert _node is not None
     client = _node.create_client(
-        SetParameters,  # type: ignore[arg-type]
+        SetParameters,
         f"{node_name}/set_parameters",
         callback_group=MutuallyExclusiveCallbackGroup(),
     )
@@ -202,7 +202,7 @@ async def _get_param(node_name: str, name: str) -> ParameterValue:
     """
     assert _node is not None
     client = _node.create_client(
-        GetParameters,  # type: ignore[arg-type]
+        GetParameters,
         f"{node_name}/get_parameters",
         callback_group=MutuallyExclusiveCallbackGroup(),
     )
@@ -278,7 +278,7 @@ async def get_param_names(params_glob: str | None) -> list[str]:
             continue
 
         client = _node.create_client(
-            ListParameters,  # type: ignore[arg-type]
+            ListParameters,
             f"{node_name}/list_parameters",
             callback_group=MutuallyExclusiveCallbackGroup(),
         )

rosbridge_library/test/capabilities/test_call_service.py

@@ -5,7 +5,7 @@
 import unittest
 from json import dumps, loads
 from threading import Thread
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 import rclpy
 from rclpy.callback_groups import ReentrantCallbackGroup
@@ -20,24 +20,27 @@
 )
 from rosbridge_library.protocol import Protocol
 
+if TYPE_CHECKING:
+    from std_srvs.srv import SetBool_Request, SetBool_Response, Trigger_Request, Trigger_Response
+
 
 class TestCallService(unittest.TestCase):
-    def trigger_cb(self, _request: Trigger.Request, response: Trigger.Response) -> Trigger.Response:
+    def trigger_cb(self, _request: Trigger_Request, response: Trigger_Response) -> Trigger_Response:
         """Handle request for a test service with no arguments."""
         response.success = True
         response.message = "called trigger service successfully"
         return response
 
     def trigger_long_cb(
-        self, _request: Trigger.Request, response: Trigger.Response
-    ) -> Trigger.Response:
+        self, _request: Trigger_Request, response: Trigger_Response
+    ) -> Trigger_Response:
         """Handle request for a long running test service with no arguments."""
         time.sleep(0.5)
         response.success = True
         response.message = "called trigger service successfully"
         return response
 
-    def set_bool_cb(self, request: SetBool.Request, response: SetBool.Response) -> SetBool.Response:
+    def set_bool_cb(self, request: SetBool_Request, response: SetBool_Response) -> SetBool_Response:
         """Handle request for a test service with arguments."""
         response.success = request.data
         if request.data:
@@ -61,19 +64,19 @@ def setUp(self) -> None:
         # Create service servers with a separate callback group
         self.cb_group = ReentrantCallbackGroup()
         self.trigger_srv = self.node.create_service(
-            Trigger,  # type: ignore[arg-type]
+            Trigger,
             self.node.get_name() + "/trigger",
             self.trigger_cb,
             callback_group=self.cb_group,
         )
         self.trigger_long_srv = self.node.create_service(
-            Trigger,  # type: ignore[arg-type]
+            Trigger,
             self.node.get_name() + "/trigger_long",
             self.trigger_long_cb,
             callback_group=self.cb_group,
         )
         self.set_bool_srv = self.node.create_service(
-            SetBool,  # type: ignore[arg-type]
+            SetBool,
             self.node.get_name() + "/set_bool",
             self.set_bool_cb,
             callback_group=self.cb_group,
@@ -104,7 +107,7 @@ def test_invalid_arguments(self) -> None:
 
     def test_call_service_works(self) -> None:
         client = self.node.create_client(
-            Trigger,  # type: ignore[arg-type]
+            Trigger,
             self.trigger_srv.srv_name,
         )
         assert client.wait_for_service(1.0)
@@ -135,7 +138,7 @@ def cb(
 
     def test_call_service_args(self) -> None:
         client = self.node.create_client(
-            SetBool,  # type: ignore[arg-type]
+            SetBool,
             self.set_bool_srv.srv_name,
         )
         assert client.wait_for_service(1.0)
@@ -173,7 +176,7 @@ def cb(
 
     def test_call_service_fails(self) -> None:
         client = self.node.create_client(
-            Trigger,  # type: ignore[arg-type]
+            Trigger,
             self.trigger_srv.srv_name,
         )
         assert client.wait_for_service(1.0)
@@ -210,7 +213,7 @@ def cb(
 
     def test_call_service_timeout(self) -> None:
         client = self.node.create_client(
-            Trigger,  # type: ignore[arg-type]
+            Trigger,
             self.trigger_long_srv.srv_name,
         )
         assert client.wait_for_service(1.0)

rosbridge_library/test/internal/actions/test_actions.py

@@ -29,7 +29,7 @@ def __init__(self, executor: Executor) -> None:
         self.executor.add_node(self.node)
         self.action_server = ActionServer(
             self.node,
-            Fibonacci,  # type: ignore[arg-type]
+            Fibonacci,
             "get_fibonacci_sequence",
             self.execute_callback,
         )
@@ -156,7 +156,7 @@ def get_result_callback(future: Future) -> None:
         # First, call the action the 'proper' way
         client = ActionClient(
             self.node,
-            Fibonacci,  # type: ignore[arg-type]
+            Fibonacci,
             "get_fibonacci_sequence",
         )
         client.wait_for_server()

rosbridge_library/test/internal/services/test_services.py

@@ -182,7 +182,7 @@ def test_service_call(self) -> None:
 
         # First, call the service the 'proper' way
         p = self.node.create_client(
-            ListParameters,  # type: ignore[arg-type]
+            ListParameters,
             self.node.get_name() + "/list_parameters",
         )
         p.wait_for_service(0.5)
@@ -212,7 +212,7 @@ def test_service_caller(self) -> None:
 
         # First, call the service the 'proper' way
         p = self.node.create_client(
-            ListParameters,  # type: ignore[arg-type]
+            ListParameters,
             self.node.get_name() + "/list_parameters",
         )
         p.wait_for_service(0.5)

rosbridge_library/test/internal/test_cbor_conversion.py

@@ -64,23 +64,23 @@ def test_numbers(self) -> None:
         # msg: ROSMessage
         # msg_type: type[ROSMessage]
         for msg_type in [Int8, Int16, Int32, Int64]:
-            msg = msg_type(data=-5)
+            msg = msg_type(data=-5)  # type: ignore[abstract]
             assert isinstance(msg, Int8 | Int16 | Int32 | Int64)
             extracted = extract_cbor_values(msg)
 
             self.assertEqual(extracted["data"], msg.data, f"type={msg_type}")
             self.assertEqual(type(extracted["data"]), int, f"type={msg_type}")
 
         for msg_type in [UInt8, UInt16, UInt32, UInt64]:
-            msg = msg_type(data=5)
+            msg = msg_type(data=5)  # type: ignore[abstract]
             assert isinstance(msg, UInt8 | UInt16 | UInt32 | UInt64)
             extracted = extract_cbor_values(msg)
 
             self.assertEqual(extracted["data"], msg.data, f"type={msg_type}")
             self.assertEqual(type(extracted["data"]), int, f"type={msg_type}")
 
         for msg_type in [Float32, Float64]:
-            msg = msg_type(data=2.3)
+            msg = msg_type(data=2.3)  # type: ignore[abstract]
             assert isinstance(msg, Float32 | Float64)
             extracted = extract_cbor_values(msg)
 
@@ -117,7 +117,7 @@ def test_integer_array(self) -> None:
             UInt32MultiArray,
             UInt64MultiArray,
         ]:
-            msg = msg_type(data=[0, 1, 2])
+            msg = msg_type(data=[0, 1, 2])  # type: ignore[abstract]
             assert isinstance(
                 msg,
                 Int8MultiArray

rosbridge_server/test/websocket/advertise_action.test.py

@@ -59,7 +59,7 @@ async def test_two_concurrent_calls(
         )
         client: ActionClient = ActionClient(
             node,
-            Fibonacci,  # type: ignore[arg-type]
+            Fibonacci,
             "/test_fibonacci_action",
         )
         client.wait_for_server()

rosbridge_server/test/websocket/advertise_action_feedback.test.py

@@ -57,7 +57,7 @@ async def test_feedback(
         )
         client: ActionClient = ActionClient(
             node,
-            Fibonacci,  # type: ignore[arg-type]
+            Fibonacci,
             "/test_fibonacci_action",
         )
         client.wait_for_server()

rosbridge_server/test/websocket/advertise_service.test.py

@@ -40,7 +40,7 @@ async def test_two_concurrent_calls(
             }
         )
         client: Client = node.create_client(
-            SetBool,  # type: ignore[arg-type]
+            SetBool,
             "/test_service",
         )
         client.wait_for_service()

rosbridge_server/test/websocket/advertise_service_duplicate.test.py

@@ -40,7 +40,7 @@ async def test_double_advertise(
             }
         )
         client: Client = node.create_client(
-            SetBool,  # type: ignore[arg-type]
+            SetBool,
             "/test_service",
         )
         client.wait_for_service()

rosbridge_server/test/websocket/call_service.test.py

@@ -20,7 +20,7 @@
 
     from common import TestClientProtocol
     from rclpy.node import Node
-    # from rclpy.task import Future
+    from std_srvs.srv import SetBool_Request, SetBool_Response
 
 log.startLogging(sys.stderr)
 
@@ -32,14 +32,14 @@ class TestCallService(unittest.TestCase):
     async def test_one_call(
         self, node: Node, make_client: Callable[[], Awaitable[TestClientProtocol]]
     ) -> None:
-        def service_cb(req: SetBool.Request, res: SetBool.Response) -> SetBool.Response:
+        def service_cb(req: SetBool_Request, res: SetBool_Response) -> SetBool_Response:
             self.assertTrue(req.data)
             res.success = True
             res.message = "Hello, world!"
             return res
 
         service = node.create_service(
-            SetBool,  # type: ignore[arg-type]
+            SetBool,
             "/test_service",
             service_cb,
             callback_group=ReentrantCallbackGroup(),
@@ -72,15 +72,15 @@ def service_cb(req: SetBool.Request, res: SetBool.Response) -> SetBool.Response:
 
         node.destroy_service(service)
 
-        def service_long_cb(req: SetBool.Request, res: SetBool.Response) -> SetBool.Response:
+        def service_long_cb(req: SetBool_Request, res: SetBool_Response) -> SetBool_Response:
             time.sleep(0.2)
             self.assertTrue(req.data)
             res.success = True
             res.message = "Hello, world!"
             return res
 
         service = node.create_service(
-            SetBool,  # type: ignore[arg-type]
+            SetBool,
             "/test_service_long",
             service_long_cb,
             callback_group=ReentrantCallbackGroup(),

rosbridge_server/test/websocket/common.py

@@ -66,7 +66,7 @@ def generate_test_description() -> LaunchDescription:
 
 async def get_server_port(node: Node) -> int:
     """Return the port which the WebSocket server is running on."""
-    client: Client = node.create_client(GetParameters, "/rosbridge_websocket/get_parameters")  # type: ignore[arg-type]
+    client: Client = node.create_client(GetParameters, "/rosbridge_websocket/get_parameters")
     try:
         if not client.wait_for_service(5):
             msg = "GetParameters service not available"

rosbridge_server/test/websocket/send_action_goal.test.py

@@ -21,6 +21,7 @@
     from collections.abc import Awaitable, Callable
 
     from common import TestClientProtocol
+    from example_interfaces.action._fibonacci import Fibonacci_Result
     from rclpy.action.server import ServerGoalHandle
     from rclpy.node import Node
 
@@ -33,7 +34,7 @@ class TestSendActionGoal(unittest.TestCase):
     def cancel_callback(self, _: ServerGoalHandle) -> CancelResponse:
         return CancelResponse.ACCEPT
 
-    def execute_callback(self, goal: ServerGoalHandle) -> Fibonacci.Result:
+    def execute_callback(self, goal: ServerGoalHandle) -> Fibonacci_Result:
         feedback_msg = Fibonacci.Feedback()
         feedback_msg.sequence = [0, 1]
 
@@ -55,7 +56,7 @@ async def test_one_call(
     ) -> None:
         action_server = ActionServer(
             node,
-            Fibonacci,  # type: ignore[arg-type]
+            Fibonacci,
             "/test_fibonacci_action",
             execute_callback=self.execute_callback,
             cancel_callback=self.cancel_callback,  # type: ignore[arg-type]