Convert TestOnOffCluster to use asyncio/SendCommand

Remove ZCLSend API usage and call SendCommand directly. Also convert the test to a test using asyncio.
project-chip · May 13, 2024 · 5d94829 · 5d94829
1 parent ae1476e
commit 5d94829
Show file tree

Hide file tree

Showing 5 changed files with 30 additions and 30 deletions.
diff --git a/src/controller/python/test/test_scripts/base.py b/src/controller/python/test/test_scripts/base.py
@@ -1097,20 +1097,24 @@ def SetNetworkCommissioningParameters(self, dataset: str):
         self.devCtrl.SetThreadOperationalDataset(bytes.fromhex(dataset))
         return True
 
-    def TestOnOffCluster(self, nodeid: int, endpoint: int, group: int):
+    async def TestOnOffCluster(self, nodeid: int, endpoint: int):
         self.logger.info(
             "Sending On/Off commands to device {} endpoint {}".format(nodeid, endpoint))
-        err, resp = self.devCtrl.ZCLSend("OnOff", "On", nodeid,
-                                         endpoint, group, {}, blocking=True)
-        if err != 0:
+
+        try:
+            await self.devCtrl.SendCommand(nodeid, endpoint,
+                                           Clusters.OnOff.Commands.On())
+        except IM.InteractionModelError as ex:
             self.logger.error(
-                "failed to send OnOff.On: error is {} with im response{}".format(err, resp))
+                "failed to send OnOff.On: error is {}".format(ex.status))
             return False
-        err, resp = self.devCtrl.ZCLSend("OnOff", "Off", nodeid,
-                                         endpoint, group, {}, blocking=True)
-        if err != 0:
+
+        try:
+            await self.devCtrl.SendCommand(nodeid, endpoint,
+                                           Clusters.OnOff.Commands.Off())
+        except IM.InteractionModelError as ex:
             self.logger.error(
-                "failed to send OnOff.Off: error is {} with im response {}".format(err, resp))
+                "failed to send OnOff.Off: error is {}".format(ex.status))
             return False
         return True
 

diff --git a/src/controller/python/test/test_scripts/commissioning_failure_test.py b/src/controller/python/test/test_scripts/commissioning_failure_test.py
@@ -19,6 +19,7 @@
 
 # Commissioning test.
 
+import asyncio
 import os
 import sys
 from optparse import OptionParser
@@ -121,9 +122,8 @@ def main():
     FailIfNot(test.TestCommissionFailure(1, 0), "Failed to commission device")
 
     logger.info("Testing on off cluster")
-    FailIfNot(test.TestOnOffCluster(nodeid=1,
-                                    endpoint=LIGHTING_ENDPOINT_ID,
-                                    group=GROUP_ID), "Failed to test on off cluster")
+    FailIfNot(asyncio.run(test.TestOnOffCluster(nodeid=1,
+                                                endpoint=LIGHTING_ENDPOINT_ID)), "Failed to test on off cluster")
 
     timeoutTicker.stop()
 

diff --git a/src/controller/python/test/test_scripts/commissioning_test.py b/src/controller/python/test/test_scripts/commissioning_test.py
@@ -19,6 +19,7 @@
 
 # Commissioning test.
 
+import asyncio
 import os
 import sys
 from optparse import OptionParser
@@ -146,9 +147,8 @@ def main():
         TestFail("Must provide device address or setup payload to commissioning the device")
 
     logger.info("Testing on off cluster")
-    FailIfNot(test.TestOnOffCluster(nodeid=options.nodeid,
-                                    endpoint=LIGHTING_ENDPOINT_ID,
-                                    group=GROUP_ID), "Failed to test on off cluster")
+    FailIfNot(asyncio.run(test.TestOnOffCluster(nodeid=options.nodeid,
+                                                endpoint=LIGHTING_ENDPOINT_ID)), "Failed to test on off cluster")
 
     FailIfNot(test.TestUsedTestCommissioner(),
               "Test commissioner check failed")

diff --git a/src/controller/python/test/test_scripts/mobile-device-test.py b/src/controller/python/test/test_scripts/mobile-device-test.py
@@ -102,19 +102,17 @@ def TestDatamodel(test: BaseTestHelper, device_nodeid: int):
     logger.info("Testing datamodel functions")
 
     logger.info("Testing on off cluster")
-    FailIfNot(test.TestOnOffCluster(nodeid=device_nodeid,
-                                    endpoint=LIGHTING_ENDPOINT_ID,
-                                    group=GROUP_ID), "Failed to test on off cluster")
+    FailIfNot(asyncio.run(test.TestOnOffCluster(nodeid=device_nodeid,
+                                                endpoint=LIGHTING_ENDPOINT_ID)), "Failed to test on off cluster")
 
     logger.info("Testing level control cluster")
     FailIfNot(asyncio.run(test.TestLevelControlCluster(nodeid=device_nodeid,
                                                        endpoint=LIGHTING_ENDPOINT_ID)),
               "Failed to test level control cluster")
 
     logger.info("Testing sending commands to non exist endpoint")
-    FailIfNot(not test.TestOnOffCluster(nodeid=device_nodeid,
-                                        endpoint=233,
-                                        group=GROUP_ID), "Failed to test on off cluster on non-exist endpoint")
+    FailIfNot(not asyncio.run(test.TestOnOffCluster(nodeid=device_nodeid,
+                                                    endpoint=233)), "Failed to test on off cluster on non-exist endpoint")
 
     # Test experimental Python cluster objects API
     logger.info("Testing cluster objects API")
@@ -149,9 +147,8 @@ def TestDatamodel(test: BaseTestHelper, device_nodeid: int):
               "Failed to validated re-subscription")
 
     logger.info("Testing on off cluster over resolved connection")
-    FailIfNot(test.TestOnOffCluster(nodeid=device_nodeid,
-                                    endpoint=LIGHTING_ENDPOINT_ID,
-                                    group=GROUP_ID), "Failed to test on off cluster")
+    FailIfNot(asyncio.run(test.TestOnOffCluster(nodeid=device_nodeid,
+                                                endpoint=LIGHTING_ENDPOINT_ID)), "Failed to test on off cluster")
 
     logger.info("Testing writing/reading fabric sensitive data")
     asyncio.run(test.TestFabricSensitive(nodeid=device_nodeid))

diff --git a/src/controller/python/test/test_scripts/split_commissioning_test.py b/src/controller/python/test/test_scripts/split_commissioning_test.py
@@ -19,6 +19,7 @@
 
 # Commissioning test.
 
+import asyncio
 import os
 import sys
 from optparse import OptionParser
@@ -118,14 +119,12 @@ def main():
               "Failed to commission device 2")
 
     logger.info("Testing on off cluster on device 1")
-    FailIfNot(test.TestOnOffCluster(nodeid=1,
-                                    endpoint=LIGHTING_ENDPOINT_ID,
-                                    group=GROUP_ID), "Failed to test on off cluster on device 1")
+    FailIfNot(asyncio.run(test.TestOnOffCluster(nodeid=1,
+                                                endpoint=LIGHTING_ENDPOINT_ID)), "Failed to test on off cluster on device 1")
 
     logger.info("Testing on off cluster on device 2")
-    FailIfNot(test.TestOnOffCluster(nodeid=2,
-                                    endpoint=LIGHTING_ENDPOINT_ID,
-                                    group=GROUP_ID), "Failed to test on off cluster on device 2")
+    FailIfNot(asyncio.run(test.TestOnOffCluster(nodeid=2,
+                                                endpoint=LIGHTING_ENDPOINT_ID)), "Failed to test on off cluster on device 2")
 
     timeoutTicker.stop()