Enforce best practices when using async in Python (project-chip#42095)

* Enforce best practices when using async in Python

* Do not use time.sleep in async functions

* Fix some more asyncio issues reported by ruff

* Fix issues reported by copilot

* Fix race between wait timeout and callback call

Author: arkq

pyproject.toml

@@ -28,6 +28,8 @@ exclude = [
 select = [
     # Standard errors and warnings
     "E", "W", "F",
+    # Ensure that async is used properly
+    "ASYNC",
     # Consistent commas
     "COM",
     # Simplify the code if possible
@@ -40,6 +42,8 @@ select = [
     "SLOT",
 ]
 ignore = [
+    "ASYNC109", # Sometimes we use "timeout" argument in async functions
+    "ASYNC230", # Allow open() calls in async functions
     "COM812", # Do not enforce trailing commas in multi-line collections for now
     "LOG015", # Do not enforce non-root loggers in the entire codebase for now
     "E501", # Do not report line length issues (formatter should handle this)

scripts/py_matter_yamltests/matter/yamltests/pseudo_clusters/clusters/delay_commands.py

@@ -13,8 +13,8 @@
 #    See the License for the specific language governing permissions and
 #    limitations under the License.
 
+import asyncio
 import sys
-import time
 
 from ..pseudo_cluster import PseudoCluster
 from .accessory_server_bridge import AccessoryServerBridge
@@ -59,7 +59,7 @@ async def WaitForMs(self, request):
                 duration_in_ms = argument['value']
 
         sys.stdout.flush()
-        time.sleep(int(duration_in_ms) / 1000)
+        await asyncio.sleep(int(duration_in_ms) / 1000)
 
     async def WaitForMessage(self, request):
         AccessoryServerBridge.waitForMessage(request)

scripts/py_matter_yamltests/matter/yamltests/websocket_runner.py

@@ -13,6 +13,7 @@
 #    See the License for the specific language governing permissions and
 #    limitations under the License.
 
+import asyncio
 import logging
 import re
 import select
@@ -92,7 +93,7 @@ async def _start_client(self, url, max_retries=_CONNECT_MAX_RETRIES_DEFAULT, int
                 duration = round((time.time() - start) * 1000, 0)
                 self._hooks.failure(duration)
                 self._hooks.retry(interval_between_retries)
-                time.sleep(interval_between_retries)
+                await asyncio.sleep(interval_between_retries)
                 return await self._start_client(url, max_retries - 1, interval_between_retries + 1)
 
         self._hooks.abort(url)
@@ -107,7 +108,7 @@ async def _start_server(self, command, url):
         if command:
             start_time = time.time()
 
-            instance = subprocess.Popen(
+            instance = subprocess.Popen(    # noqa: ASYNC220
                 command,
                 bufsize=0,                  # unbuffered
                 text=False,                 # keep output as bytes

src/controller/python/matter/ChipDeviceCtrl.py

@@ -1108,10 +1108,9 @@ async def DiscoverCommissionableNodes(self, filterType: discovery.FilterType = d
                     self.devCtrl, int(filterType), str(filter).encode("utf-8")))
 
             async def _wait_discovery():
-                while not await self._ChipStack.CallAsyncWithResult(
+                while not await self._ChipStack.CallAsyncWithResult(  # noqa: ASYNC110
                         lambda: self._dmLib.pychip_DeviceController_HasDiscoveredCommissionableNode(self.devCtrl)):
                     await asyncio.sleep(0.1)
-                return
 
             try:
                 if stopOnFirst:

src/controller/python/matter/ChipStack.py

@@ -120,6 +120,11 @@ def future(self):
         return self._future
 
     def _done(self):
+        if self._future.cancelled():
+            # If this helper is used with the asyncio.wait_for(), it might
+            # happen that the future will be cancelled before the callback
+            # is called. Do not raise an exception in this case.
+            return
         if self._exception:
             self._future.set_exception(self._exception)
         else:

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

@@ -1458,7 +1458,7 @@ async def TestSubscriptionResumptionCapacityStep2(self, nodeId: int, endpoint: i
             restartRemoteThread.start()
 
             # Wait for some time so that the device will be resolving the address of the first controller after restarting
-            time.sleep(8)
+            await asyncio.sleep(8)
             restartRemoteThread.join(10.0)
 
             self.logger.info("Send a new subscription request from the second controller")

src/python_testing/TC_ACL_2_10.py

@@ -32,10 +32,9 @@
 #       --endpoint 0
 # === END CI TEST ARGUMENTS ===
 
-
+import asyncio
 import logging
 import random
-import time
 
 from mobly import asserts
 
@@ -247,7 +246,7 @@ async def test_TC_ACL_2_10(self):
 
                 # The test runner will automatically wait for the app-ready-pattern before continuing
                 # Waiting 1 second after the app-ready-pattern is detected as we need to wait a tad longer for the app to be ready and stable, otherwise TH2 connection fails later on in test step 14.
-                time.sleep(1)
+                await asyncio.sleep(1)
 
                 # Expire sessions and re-establish connections
                 self.th1.ExpireSessions(self.dut_node_id)

src/python_testing/TC_CADMIN_1_11.py

@@ -31,10 +31,9 @@
 #     quiet: true
 # === END CI TEST ARGUMENTS ===
 
-
+import asyncio
 import logging
 import random
-from time import sleep
 
 from mobly import asserts
 from support_modules.cadmin_support import CADMINBaseTest
@@ -144,7 +143,7 @@ async def test_TC_CADMIN_1_11(self):
         revokeCmd = Clusters.AdministratorCommissioning.Commands.RevokeCommissioning()
         await self.th1.SendCommand(nodeId=self.dut_node_id, endpoint=0, payload=revokeCmd, timedRequestTimeoutMs=6000)
         # The failsafe cleanup is scheduled after the command completes, so give it a bit of time to do that
-        sleep(1)
+        await asyncio.sleep(1)
 
         self.step(9)
 
@@ -178,7 +177,7 @@ async def test_TC_CADMIN_1_11(self):
             revokeCmd = Clusters.AdministratorCommissioning.Commands.RevokeCommissioning()
             await self.th1.SendCommand(nodeId=self.dut_node_id, endpoint=0, payload=revokeCmd, timedRequestTimeoutMs=6000)
             # The failsafe cleanup is scheduled after the command completes, so give it a bit of time to do that
-            sleep(1)
+            await asyncio.sleep(1)
 
         else:
             self.skip_step("9a")

src/python_testing/TC_CADMIN_1_3_4.py

@@ -44,9 +44,9 @@
 #     quiet: true
 # === END CI TEST ARGUMENTS ===
 
+import asyncio
 import logging
 import random
-from time import sleep
 
 from mobly import asserts
 from support_modules.cadmin_support import CADMINBaseTest
@@ -221,7 +221,7 @@ async def combined_commission_val_steps(self, commission_type: str):
             revokeCmd = Clusters.AdministratorCommissioning.Commands.RevokeCommissioning()
             await self.th1.SendCommand(nodeId=self.dut_node_id, endpoint=0, payload=revokeCmd, timedRequestTimeoutMs=6000)
             # The failsafe cleanup is scheduled after the command completes, so give it a bit of time to do that
-            sleep(1)
+            await asyncio.sleep(1)
 
         if commission_type == "ECM":
             self.step(13)

src/python_testing/TC_CADMIN_1_5.py

@@ -31,9 +31,8 @@
 #     quiet: true
 # === END CI TEST ARGUMENTS ===
 
-import asyncio.exceptions as ae
+import asyncio
 import logging
-from time import sleep
 
 from mobly import asserts
 from support_modules.cadmin_support import CADMINBaseTest
@@ -63,7 +62,7 @@ async def commission_on_network(self, setup_code: int, discriminator: int, expec
                     nodeId=self.dut_node_id, setupPinCode=setup_code,
                     filterType=ChipDeviceCtrl.DiscoveryFilterType.LONG_DISCRIMINATOR, filter=discriminator)
 
-            except ae.CancelledError:
+            except asyncio.CancelledError:
                 # This is expected to fail due to timeout, however there is no code to validate here, so just passing since the correct exception was raised to get to this point
                 pass
 
@@ -104,6 +103,10 @@ def steps_TC_CADMIN_1_5(self) -> list[TestStep]:
     def pics_TC_CADMIN_1_5(self) -> list[str]:
         return ["CADMIN.S"]
 
+    @property
+    def default_timeout(self) -> int:
+        return 5 * 60  # 5 minutes
+
     @async_test_body
     async def test_TC_CADMIN_1_5(self):
         self.step(1)
@@ -125,7 +128,7 @@ async def test_TC_CADMIN_1_5(self):
             expected_discriminator=params.randomDiscriminator
         )
         logging.info(f"Successfully found service with CM={service.txt.get('CM')}, D={service.txt.get('D')}")
-        sleep(190)
+        await asyncio.sleep(190)
 
         self.step(4)
         await self.commission_on_network(setup_code=params.commissioningParameters.setupPinCode, discriminator=params.randomDiscriminator)
@@ -136,7 +139,7 @@ async def test_TC_CADMIN_1_5(self):
         self.step(6)
         revokeCmd = Clusters.AdministratorCommissioning.Commands.RevokeCommissioning()
         await self.th1.SendCommand(nodeId=self.dut_node_id, endpoint=0, payload=revokeCmd, timedRequestTimeoutMs=6000)
-        sleep(1)
+        await asyncio.sleep(1)
 
         self.step(7)
         await self.commission_on_network(setup_code=params2.commissioningParameters.setupPinCode, discriminator=params2.randomDiscriminator, expected_error=0x00000032)