Fail python tests run on unhandled asyncio exceptions (project-chip#41976)

* Improve typing in the matter/clusters/Command.py

* Replace get_event_loop with get_running_loop to ensure loop was created elsewhere

* Fail the test run on unhandled exceptions

* Refactor _handleError() logic to be more readable

* Update according to LLM suggestions

---------

Co-authored-by: Andrei Litvin <[email protected]>

Author: arkq

examples/fabric-admin/scripts/fabric-sync-app.py

@@ -42,7 +42,7 @@ async def forward_f(prefix: bytes, f_in: asyncio.StreamReader,
 
 async def forward_stdin(f_out: asyncio.StreamWriter):
     """Forward stdin to f_out."""
-    loop = asyncio.get_event_loop()
+    loop = asyncio.get_running_loop()
     reader = asyncio.StreamReader()
     protocol = asyncio.StreamReaderProtocol(reader)
     await loop.connect_read_pipe(lambda: protocol, sys.stdin)
@@ -175,7 +175,7 @@ async def main(args):
             passcode=args.passcode,
         ))
 
-    loop = asyncio.get_event_loop()
+    loop = asyncio.get_running_loop()
 
     def terminate():
         with contextlib.suppress(ProcessLookupError):

scripts/tests/chipyaml/runner.py

@@ -280,8 +280,7 @@ def parse(parser_group: ParserGroup):
     runner_config = None
 
     runner = TestRunner()
-    loop = asyncio.get_event_loop()
-    return loop.run_until_complete(runner.run(parser_group.builder_config, runner_config))
+    return asyncio.run(runner.run(parser_group.builder_config, runner_config))
 
 
 @runner_base.command()
@@ -291,8 +290,7 @@ def dry_run(parser_group: ParserGroup):
     runner_config = TestRunnerConfig(hooks=TestRunnerLogger())
 
     runner = TestRunner()
-    loop = asyncio.get_event_loop()
-    return loop.run_until_complete(runner.run(parser_group.builder_config, runner_config))
+    return asyncio.run(runner.run(parser_group.builder_config, runner_config))
 
 
 @runner_base.command()
@@ -306,8 +304,7 @@ def run(parser_group: ParserGroup, adapter: str, stop_on_error: bool, stop_on_wa
     runner_config = TestRunnerConfig(adapter, parser_group.pseudo_clusters, runner_options, runner_hooks)
 
     runner = TestRunner()
-    loop = asyncio.get_event_loop()
-    return loop.run_until_complete(runner.run(parser_group.builder_config, runner_config))
+    return asyncio.run(runner.run(parser_group.builder_config, runner_config))
 
 
 @runner_base.command()
@@ -330,8 +327,7 @@ def websocket(parser_group: ParserGroup, adapter: str, stop_on_error: bool, stop
         server_address, server_port, server_path, server_arguments, websocket_runner_hooks)
 
     runner = WebSocketRunner(websocket_runner_config)
-    loop = asyncio.get_event_loop()
-    return loop.run_until_complete(runner.run(parser_group.builder_config, runner_config))
+    return asyncio.run(runner.run(parser_group.builder_config, runner_config))
 
 
 @runner_base.command()
@@ -349,8 +345,7 @@ def matter_repl(parser_group: ParserGroup, adapter: str, stop_on_error: bool, st
     if commission_on_network_dut:
         node_id_to_commission = parser_group.builder_config.parser_config.config_override['nodeId']
     runner = __import__(runner, fromlist=[None]).Runner(repl_storage_path, node_id_to_commission=node_id_to_commission)
-    loop = asyncio.get_event_loop()
-    return loop.run_until_complete(runner.run(parser_group.builder_config, runner_config))
+    return asyncio.run(runner.run(parser_group.builder_config, runner_config))
 
 
 @runner_base.command()

src/controller/python/matter/ChipStack.py

@@ -110,7 +110,7 @@ class AsyncioCallableHandle:
 
     def __init__(self, callback):
         self._callback = callback
-        self._loop = asyncio.get_event_loop()
+        self._loop = asyncio.get_running_loop()
         self._future = self._loop.create_future()
         self._result = None
         self._exception = None

src/controller/python/matter/clusters/Command.py

@@ -79,26 +79,24 @@ def FindCommandClusterObject(isClientSideCommand: bool, path: CommandPath):
 
 
 class AsyncCommandTransaction:
-    def __init__(self, future: Future, eventLoop, expectType: Type):
+    def __init__(self, future: Future, eventLoop, expectType: type[ClusterCommand] | None):
         self._event_loop = eventLoop
         self._future = future
         self._expect_type = expectType
 
     def _handleResponse(self, path: CommandPath, status: Status, response: bytes):
-        if (len(response) == 0):
+        if len(response) == 0:
             self._future.set_result(None)
         else:
             # If a type hasn't been assigned, let's auto-deduce it.
-            if (self._expect_type is None):
+            if self._expect_type is None:
                 self._expect_type = FindCommandClusterObject(False, path)
 
             if self._expect_type:
                 try:
-                    self._future.set_result(
-                        self._expect_type.FromTLV(response))
+                    self._future.set_result(self._expect_type.FromTLV(response))
                 except Exception as ex:
-                    self._handleError(
-                        status, 0, ex)
+                    self._handleError(status, PyChipError.from_code(0), ex)
             else:
                 self._future.set_result(None)
 
@@ -107,101 +105,89 @@ def handleResponse(self, path: CommandPath, index: int, status: Status, response
         # checking `index`. We just share a callback API with batch commands. If we ever get a
         # second call to `handleResponse` we will see a different error on trying to set future
         # that has already been set.
-        self._event_loop.call_soon_threadsafe(
-            self._handleResponse, path, status, response)
-
-    def _handleError(self, imError: Status, chipError: PyChipError, exception: Exception):
-        if exception:
-            self._future.set_exception(exception)
-        elif chipError != 0:
-            self._future.set_exception(chipError.to_exception())
-        else:
+        self._event_loop.call_soon_threadsafe(self._handleResponse, path, status, response)
+
+    def _handleError(self, imError: Status, chipError: PyChipError, exception: Exception | None):
+        if chipError != 0:
+            exception = chipError.to_exception()
+        if exception is None:
             try:
                 # If you got an exception from this call other than AttributeError please
                 # add it to the except block below. We changed Exception->AttributeError as
                 # that is what we thought we are trying to catch here.
-                self._future.set_exception(
-                    InteractionModelError(InteractionModelStatus(imError.IMStatus), imError.ClusterStatus))
+                exception = InteractionModelError(InteractionModelStatus(imError.IMStatus), imError.ClusterStatus)
             except AttributeError:
-                logger.exception("Failed to map interaction model status received: %s. Remapping to Failure." % imError)
-                self._future.set_exception(InteractionModelError(
-                    InteractionModelStatus.Failure, imError.ClusterStatus))
+                logger.exception("Failed to map interaction model status received: %s. Remapping to Failure.", imError)
+                exception = InteractionModelError(InteractionModelStatus.Failure, imError.ClusterStatus)
+        self._future.set_exception(exception)
 
     def handleError(self, status: Status, chipError: PyChipError):
-        self._event_loop.call_soon_threadsafe(
-            self._handleError, status, chipError, None
-        )
+        self._event_loop.call_soon_threadsafe(self._handleError, status, chipError, None)
 
     def handleDone(self):
         ctypes.pythonapi.Py_DecRef(ctypes.py_object(self))
 
 
 class AsyncBatchCommandsTransaction:
-    def __init__(self, future: Future, eventLoop, expectTypes: List[Type]):
+    def __init__(self, future: Future, eventLoop, expectTypes: list[type[ClusterCommand] | None]):
         self._event_loop = eventLoop
         self._future = future
         self._expect_types = expectTypes
-        default_im_failure = InteractionModelError(
-            InteractionModelStatus.NoCommandResponse)
-        self._responses = [default_im_failure] * len(expectTypes)
+        default_im_failure = InteractionModelError(InteractionModelStatus.NoCommandResponse)
+        self._responses: list[ClusterCommand | InteractionModelError | None] = [default_im_failure] * len(expectTypes)
 
     def _handleResponse(self, path: CommandPath, index: int, status: Status, response: bytes):
         if index > len(self._responses):
-            self._handleError(status, 0, IndexError(f"CommandSenderCallback has given us an unexpected index value {index}"))
+            self._handleError(status, PyChipError.from_code(0),
+                              IndexError(f"CommandSenderCallback has given us an unexpected index value {index}"))
             return
 
         if status.IMStatus != InteractionModelStatus.Success:
             try:
                 self._responses[index] = InteractionModelError(
                     InteractionModelStatus(status.IMStatus), status.ClusterStatus)
             except AttributeError as ex:
-                self._handleError(status, 0, ex)
-        elif (len(response) == 0):
+                self._handleError(status, PyChipError.from_code(0), ex)
+        elif len(response) == 0:
             self._responses[index] = None
         else:
             # If a type hasn't been assigned, let's auto-deduce it.
-            if (self._expect_types[index] is None):
+            if self._expect_types[index] is None:
                 self._expect_types[index] = FindCommandClusterObject(False, path)
 
-            if self._expect_types[index]:
+            if expectType := self._expect_types[index]:
                 try:
                     # If you got an exception from this call other than AttributeError please
                     # add it to the except block below. We changed Exception->AttributeError as
                     # that is what we thought we are trying to catch here.
-                    self._responses[index] = self._expect_types[index].FromTLV(response)
+                    self._responses[index] = expectType.FromTLV(response)
                 except AttributeError as ex:
-                    self._handleError(status, 0, ex)
+                    self._handleError(status, PyChipError.from_code(0), ex)
             else:
                 self._responses[index] = None
 
     def handleResponse(self, path: CommandPath, index: int, status: Status, response: bytes):
-        self._event_loop.call_soon_threadsafe(
-            self._handleResponse, path, index, status, response)
+        self._event_loop.call_soon_threadsafe(self._handleResponse, path, index, status, response)
 
-    def _handleError(self, imError: Status, chipError: PyChipError, exception: Exception):
+    def _handleError(self, imError: Status, chipError: PyChipError, exception: Exception | None):
         if self._future.done():
-            logger.exception(f"Recieved another error. Only expecting one error. imError:{imError}, chipError {chipError}")
+            logger.exception(f"Received another error. Only expecting one error. imError:{imError}, chipError {chipError}")
             return
-        if exception:
-            self._future.set_exception(exception)
-        elif chipError != 0:
-            self._future.set_exception(chipError.to_exception())
-        else:
+        if chipError != 0:
+            exception = chipError.to_exception()
+        if exception is None:
             try:
                 # If you got an exception from this call other than AttributeError please
                 # add it to the except block below. We changed Exception->AttributeError as
                 # that is what we thought we are trying to catch here.
-                self._future.set_exception(
-                    InteractionModelError(InteractionModelStatus(imError.IMStatus), imError.ClusterStatus))
+                exception = InteractionModelError(InteractionModelStatus(imError.IMStatus), imError.ClusterStatus)
             except AttributeError:
-                logger.exception("Failed to map interaction model status received: %s. Remapping to Failure." % imError)
-                self._future.set_exception(InteractionModelError(
-                    InteractionModelStatus.Failure, imError.ClusterStatus))
+                logger.exception("Failed to map interaction model status received: %s. Remapping to Failure.", imError)
+                exception = InteractionModelError(InteractionModelStatus.Failure, imError.ClusterStatus)
+        self._future.set_exception(exception)
 
     def handleError(self, status: Status, chipError: PyChipError):
-        self._event_loop.call_soon_threadsafe(
-            self._handleError, status, chipError, None
-        )
+        self._event_loop.call_soon_threadsafe(self._handleError, status, chipError, None)
 
     def _handleDone(self):
         # Future might already be set with exception from `handleError`
@@ -210,13 +196,11 @@ def _handleDone(self):
         ctypes.pythonapi.Py_DecRef(ctypes.py_object(self))
 
     def handleDone(self):
-        self._event_loop.call_soon_threadsafe(
-            self._handleDone
-        )
+        self._event_loop.call_soon_threadsafe(self._handleDone)
 
 
 class TestOnlyAsyncBatchCommandsTransaction(AsyncBatchCommandsTransaction):
-    def __init__(self, future: Future, eventLoop, expectTypes: List[Type]):
+    def __init__(self, future: Future, eventLoop, expectTypes: list[type[ClusterCommand] | None]):
         self._responseMessageCount = 0
         super().__init__(future, eventLoop, expectTypes)
 
@@ -264,7 +248,8 @@ def _TestOnlyOnCommandSenderDoneCallback(closure, testOnlyDoneInfo: TestOnlyPyOn
     closure.handleDone()
 
 
-def TestOnlySendCommandTimedRequestFlagWithNoTimedInvoke(future: Future, eventLoop, responseType, device, commandPath, payload):
+def TestOnlySendCommandTimedRequestFlagWithNoTimedInvoke(future: Future, eventLoop,
+                                                         responseType: type[ClusterCommand] | None, device, commandPath, payload):
     ''' ONLY TO BE USED FOR TEST: Sends the payload with a TimedRequest flag but no TimedInvoke transaction
     '''
     if (responseType is not None) and (not issubclass(responseType, ClusterCommand)):
@@ -285,7 +270,8 @@ def TestOnlySendCommandTimedRequestFlagWithNoTimedInvoke(future: Future, eventLo
         ))
 
 
-async def SendCommand(future: Future, eventLoop, responseType: Type, device, commandPath: CommandPath, payload: ClusterCommand,
+async def SendCommand(future: Future, eventLoop, responseType: type[ClusterCommand] | None,
+                      device, commandPath: CommandPath, payload: ClusterCommand,
                       timedRequestTimeoutMs: Union[None, int] = None, interactionTimeoutMs: Union[None, int] = None,
                       busyWaitMs: Union[None, int] = None, suppressResponse: Union[None, bool] = None, allowLargePayload: Union[None, bool] = None) -> PyChipError:
     ''' Send a cluster-object encapsulated command to a device and does the following:
@@ -322,7 +308,8 @@ async def SendCommand(future: Future, eventLoop, responseType: Type, device, com
         ))
 
 
-def _BuildPyInvokeRequestData(commands: List[InvokeRequestInfo], timedRequestTimeoutMs: Optional[int], responseTypes, suppressTimedRequestMessage: bool = False) -> List[PyInvokeRequestData]:
+def _BuildPyInvokeRequestData(commands: List[InvokeRequestInfo], timedRequestTimeoutMs: Optional[int],
+                              responseTypes: list[type[ClusterCommand] | None], suppressTimedRequestMessage: bool = False) -> List[PyInvokeRequestData]:
     numberOfCommands = len(commands)
     pyBatchCommandsDataArrayType = PyInvokeRequestData * numberOfCommands
     pyBatchCommandsData = pyBatchCommandsDataArrayType()
@@ -377,7 +364,7 @@ async def SendBatchCommands(future: Future, eventLoop, device, commands: List[In
     '''
     handle = GetLibraryHandle()
 
-    responseTypes: List[Type] = []
+    responseTypes: list[type[ClusterCommand] | None] = []
     pyBatchCommandsData = _BuildPyInvokeRequestData(commands, timedRequestTimeoutMs, responseTypes)
 
     transaction = AsyncBatchCommandsTransaction(future, eventLoop, responseTypes)
@@ -412,7 +399,7 @@ def TestOnlySendBatchCommands(future: Future, eventLoop, device, commands: List[
 
     handle = GetLibraryHandle()
 
-    responseTypes: List[Type] = []
+    responseTypes: list[type[ClusterCommand] | None] = []
     pyBatchCommandsData = _BuildPyInvokeRequestData(commands, timedRequestTimeoutMs,
                                                     responseTypes, suppressTimedRequestMessage=suppressTimedRequestMessage)
 

src/controller/python/matter/webrtc/browser_webrtc_client.py

@@ -27,7 +27,7 @@ def __init__(self, ws_client: AsyncWebSocketClient, id: int):
         self.event_callbacks: dict[str, Callable] = {}
         self.ws_client = ws_client
         self.pending_cmd_responses: dict[str, asyncio.Future] = {}
-        self.event_loop = asyncio.get_event_loop()
+        self.event_loop = asyncio.get_running_loop()
         self._event_callbacks_without_parameter = ("GATHERING_STATE_COMPLETE",)
         self.id = id
 

src/controller/python/tests/scripts/commissioning_window_test.py

@@ -116,9 +116,7 @@ async def main():
 
 if __name__ == "__main__":
     try:
-        loop = asyncio.get_event_loop()
-        loop.run_until_complete(main())
-        loop.close()
+        asyncio.run(main())
     except Exception as ex:
         logger.exception(ex)
         TestFail("Exception occurred when running tests.")

src/controller/python/tests/scripts/icd_device_test.py

@@ -145,9 +145,7 @@ async def main():
 
 if __name__ == "__main__":
     try:
-        loop = asyncio.get_event_loop()
-        loop.run_until_complete(main())
-        loop.close()
+        asyncio.run(main())
     except Exception as ex:
         logger.exception(ex)
         TestFail("Exception occurred when running tests.")

src/python_testing/matter_testing_infrastructure/matter/testing/runner.py

@@ -405,8 +405,7 @@ def run_tests_no_exit(
 
     # NOTE: It's not possible to pass event loop via Mobly TestRunConfig user params, because the
     #       Mobly deep copies the user params before passing them to the test class and the event
-    # loop is not serializable. So, we are setting the event loop as a test
-    # class member.
+    #       loop is not serializable. So, we are setting the event loop as a test class member.
     CommissionDeviceTest.event_loop = event_loop
     test_class.event_loop = event_loop
 
@@ -455,6 +454,15 @@ def run_tests_no_exit(
         # Execute the test class with the config
         ok = True
 
+        def _handler(loop, context):
+            loop.default_exception_handler(context)
+            nonlocal ok
+            # Fail the test run on unhandled exceptions.
+            ok = False
+
+        # Set custom exception handler to catch unhandled exceptions.
+        event_loop.set_exception_handler(_handler)
+
         runner = TestRunner(log_dir=test_config.log_path,
                             testbed_name=test_config.testbed_name)