[Python] Make it easier to run Python tests in CI

.github/workflows/build.yaml

@@ -260,6 +260,21 @@ jobs:
               run: |
                   scripts/run_in_build_env.sh 'pip3 install ./out/controller/python/chip-0.0-cp37-abi3-linux_x86_64.whl'
                   scripts/run_in_build_env.sh '(cd src/controller/python/test/unit_tests/ && python3 -m unittest -v)'
+            - name: Build all clusters app for test
+              timeout-minutes: 50
+              run: |
+                  scripts/examples/gn_build_example.sh examples/all-clusters-app/linux out/all_clusters_debug chip_enable_wifi=false
+            - name: Run Python REPL API Tests
+              timeout-minutes: 5
+              run: |
+                  scripts/run_in_build_env.sh '
+                      rm -rf /tmp/chip*
+                      ./out/all_clusters_debug/chip-all-clusters-app &
+                      ./src/controller/python/test/test_scripts/mobile-device-test.py -t 90 --disable-test datamodel.timedrequest.timeout
+                      res=$?
+                      killall -s SIGINT chip-all-clusters-app || true
+                      exit $res
+                      '
     build_darwin:
         name: Build on Darwin (clang, python_lib, simulated)
         timeout-minutes: 200

src/controller/python/test/test_scripts/base.py

@@ -54,6 +54,35 @@ def FailIfNot(cond, message):
         TestFail(message)
 
 
+_enabled_tests = []
+_disabled_tests = []
+
+
+def SetTestSet(enabled_tests, disabled_tests):
+    global _enabled_tests, _disabled_tests
+    _enabled_tests = enabled_tests[:]
+    _disabled_tests = disabled_tests[:]
+
+
+def TestIsEnabled(test_name: str):
+    enabled_len = -1
+    disabled_len = -1
+    if 'all' in _enabled_tests:
+        enabled_len = 0
+    if 'all' in _disabled_tests:
+        disabled_len = 0
+
+    for test_item in _enabled_tests:
+        if test_name.startswith(test_item) and (len(test_item) > enabled_len):
+            enabled_len = len(test_item)
+
+    for test_item in _disabled_tests:
+        if test_name.startswith(test_item) and (len(test_item) > disabled_len):
+            disabled_len = len(test_item)
+
+    return enabled_len > disabled_len
+
+
 class TestTimeout(threading.Thread):
     def __init__(self, timeout: int):
         threading.Thread.__init__(self)
@@ -135,6 +164,16 @@ def TestDiscovery(self, discriminator: int):
         self.logger.info(f"Found device at {res}")
         return res
 
+    def TestKeyExchangeBLE(self, discriminator: int, setuppin: int, nodeid: int):
+        self.logger.info(
+            "Conducting key exchange with device {}".format(discriminator))
+        if not self.devCtrl.ConnectBLE(discriminator, setuppin, nodeid):
+            self.logger.info(
+                "Failed to finish key exchange with device {}".format(discriminator))
+            return False
+        self.logger.info("Device finished key exchange.")
+        return True
+
     def TestKeyExchange(self, ip: str, setuppin: int, nodeid: int):
         self.logger.info("Conducting key exchange with device {}".format(ip))
         if not self.devCtrl.CommissionIP(ip.encode("utf-8"), setuppin, nodeid):

src/controller/python/test/test_scripts/cluster_objects.py

@@ -24,6 +24,8 @@
 import asyncio
 import time
 
+from base import TestIsEnabled
+
 logger = logging.getLogger('PythonMatterControllerTEST')
 logger.setLevel(logging.INFO)
 
@@ -319,6 +321,9 @@ async def TestReadEventRequests(cls, devCtrl, expectEventsNum):
 
     @classmethod
     async def TestTimedRequest(cls, devCtrl):
+        if not TestIsEnabled("datamodel.timedrequest"):
+            logger.info("*: TimedRequest is skipped")
+
         logger.info("1: Send Timed Command Request")
         req = Clusters.TestCluster.Commands.TimedInvokeRequest()
         await devCtrl.SendCommand(nodeid=NODE_ID, endpoint=1, payload=req, timedRequestTimeoutMs=1000)
@@ -331,26 +336,29 @@ async def TestTimedRequest(cls, devCtrl):
                                      ],
                                      timedRequestTimeoutMs=1000)
 
-        logger.info("3: Send Timed Command Request -- Timeout")
-        try:
-            req = Clusters.TestCluster.Commands.TimedInvokeRequest()
-            # 10ms is a pretty short timeout, RTT is 400ms in simulated network on CI, so this test should fail.
-            await devCtrl.SendCommand(nodeid=NODE_ID, endpoint=1, payload=req, timedRequestTimeoutMs=10)
-            raise AssertionError("Timeout expected!")
-        except chip.exceptions.ChipStackException:
-            pass
-
-        logger.info("4: Send Timed Write Request -- Timeout")
-        try:
-            await devCtrl.WriteAttribute(nodeid=NODE_ID,
-                                         attributes=[
-                                             (1, Clusters.TestCluster.Attributes.TimedWriteBoolean(
-                                                 True)),
-                                         ],
-                                         timedRequestTimeoutMs=10)
-            raise AssertionError("Timeout expected!")
-        except chip.exceptions.ChipStackException:
-            pass
+        if not TestIsEnabled("datamodel.timedrequest.timeout"):
+            logger.info("*: Timeout in timed request is skipped")
+        else:
+            logger.info("3: Send Timed Command Request -- Timeout")
+            try:
+                req = Clusters.TestCluster.Commands.TimedInvokeRequest()
+                # 10ms is a pretty short timeout, RTT is 400ms in simulated network on CI, so this test should fail.
+                await devCtrl.SendCommand(nodeid=NODE_ID, endpoint=1, payload=req, timedRequestTimeoutMs=10)
+                raise AssertionError("Timeout expected!")
+            except chip.exceptions.ChipStackException:
+                pass
+
+            logger.info("4: Send Timed Write Request -- Timeout")
+            try:
+                await devCtrl.WriteAttribute(nodeid=NODE_ID,
+                                             attributes=[
+                                                 (1, Clusters.TestCluster.Attributes.TimedWriteBoolean(
+                                                     True)),
+                                             ],
+                                             timedRequestTimeoutMs=10)
+                raise AssertionError("Timeout expected!")
+            except chip.exceptions.ChipStackException:
+                pass
 
         logger.info(
             "5: Sending TestCluster-TimedInvokeRequest without timedRequestTimeoutMs should be rejected")

src/controller/python/test/test_scripts/mobile-device-test.py

@@ -18,10 +18,14 @@
 #
 
 # Commissioning test.
+from logging import disable
 import os
 import sys
-from optparse import OptionParser
-from base import TestFail, TestTimeout, BaseTestHelper, FailIfNot, logger
+import click
+import coloredlogs
+import chip.logging
+import logging
+from base import TestFail, TestTimeout, BaseTestHelper, FailIfNot, logger, TestIsEnabled, SetTestSet
 from cluster_objects import NODE_ID, ClusterObjectTests
 from network_commissioning import NetworkCommissioningTests
 import asyncio
@@ -40,53 +44,30 @@
 # Network id, for the thread network, current a const value, will be changed to XPANID of the thread network.
 TEST_THREAD_NETWORK_ID = "fedcba9876543210"
 TEST_DISCRIMINATOR = 3840
+TEST_SETUPPIN = 20202021
 
 ENDPOINT_ID = 0
 LIGHTING_ENDPOINT_ID = 1
 GROUP_ID = 0
 
+TEST_CONTROLLER_NODE_ID = 112233
+TEST_DEVICE_NODE_ID = 1
 
-def main():
-    optParser = OptionParser()
-    optParser.add_option(
-        "-t",
-        "--timeout",
-        action="store",
-        dest="testTimeout",
-        default=75,
-        type='int',
-        help="The program will return with timeout after specified seconds.",
-        metavar="<timeout-second>",
-    )
-    optParser.add_option(
-        "-a",
-        "--address",
-        action="store",
-        dest="deviceAddress",
-        default='',
-        type='str',
-        help="Address of the device",
-        metavar="<device-addr>",
-    )
-
-    (options, remainingArgs) = optParser.parse_args(sys.argv[1:])
-
-    timeoutTicker = TestTimeout(options.testTimeout)
-    timeoutTicker.start()
+ALL_TESTS = ['nwprov', 'datamodel']
 
-    test = BaseTestHelper(nodeid=112233)
 
+def ethernet_commissioning(test: BaseTestHelper, discriminator, setup_pin, device_nodeid):
     logger.info("Testing discovery")
-    FailIfNot(test.TestDiscovery(discriminator=TEST_DISCRIMINATOR),
-              "Failed to discover any devices.")
+    address = test.TestDiscovery(discriminator=discriminator)
+    FailIfNot(address, "Failed to discover any devices.")
 
     # FailIfNot(test.SetNetworkCommissioningParameters(dataset=TEST_THREAD_NETWORK_DATASET_TLV),
     #           "Failed to finish network commissioning")
 
     logger.info("Testing key exchange")
-    FailIfNot(test.TestKeyExchange(ip=options.deviceAddress,
-                                   setuppin=20202021,
-                                   nodeid=1),
+    FailIfNot(test.TestKeyExchange(ip=address.decode("utf-8"),
+                                   setuppin=setup_pin,
+                                   nodeid=device_nodeid),
               "Failed to finish key exchange")
 
     #
@@ -95,35 +76,53 @@ def main():
     #
     # Issue: #15688
     #
-    # asyncio.run(test.TestMultiFabric(ip=options.deviceAddress,
+    # asyncio.run(test.TestMultiFabric(ip=address.decode("utf-8"),
     #                                  setuppin=20202021,
     #                                  nodeid=1))
 
     logger.info("Testing closing sessions")
-    FailIfNot(test.TestCloseSession(nodeid=1), "Failed to close sessions")
-
-    logger.info("Testing resolve")
-    FailIfNot(test.TestResolve(nodeid=1),
-              "Failed to resolve nodeid")
-
-    # Still test network commissioning
-    logger.info("Testing network commissioning")
-    FailIfNot(asyncio.run(NetworkCommissioningTests(devCtrl=test.devCtrl, nodeid=1).run()),
-              "Failed to finish network commissioning")
-
+    FailIfNot(test.TestCloseSession(nodeid=device_nodeid),
+              "Failed to close sessions")
+
+
+@click.command()
+@click.option("--controller-nodeid", default=TEST_CONTROLLER_NODE_ID, type=int, help="NodeId of the controller.")
+@click.option("--device-nodeid", default=TEST_DEVICE_NODE_ID, type=int, help="NodeId of the device.")
+@click.option("--timeout", "-t", default=240, type=int, help="The program will return with timeout after specified seconds.")
+@click.option("--discriminator", default=TEST_DISCRIMINATOR, type=int, help="Discriminator of the device.")
+@click.option("--setup-pin", default=TEST_SETUPPIN, type=int, help="Setup pincode of the device.")
+@click.option("--commissioning", default='eth', type=click.Choice(['eth'], case_sensitive=False), help="The network type of the device, used during commissioning test.")
+@click.option('--enable-test', default=['all'], multiple=True, help='The tests to be executed.')
+@click.option('--disable-test', default=[], multiple=True, help='The tests to be excluded.')
+@click.option('--log-level', default='WARN', type=click.Choice(['ERROR', 'WARN', 'INFO', 'DEBUG']), help="The log level of the test.")
+def main(controller_nodeid, device_nodeid, timeout, discriminator, setup_pin, commissioning, enable_test, disable_test, log_level):
+    logger.info("Test Parameters:")
+    logger.info(f"\tController NodeId: {controller_nodeid}")
+    logger.info(f"\tDevice NodeId:     {device_nodeid}")
+    logger.info(f"\tTest Timeout:      {timeout}s")
+    logger.info(f"\tCommission Over:   {commissioning}")
+    logger.info(f"\tDiscriminator:     {discriminator}")
+    logger.info(f"\tEnabled Tests:     {enable_test}")
+    logger.info(f"\tDisabled Tests:    {disable_test}")
+    SetTestSet(enable_test, disable_test)
+    do_tests(controller_nodeid, device_nodeid, timeout,
+             discriminator, setup_pin, commissioning, log_level)
+
+
+def test_datamodel(test: BaseTestHelper, device_nodeid: int):
     logger.info("Testing on off cluster")
-    FailIfNot(test.TestOnOffCluster(nodeid=1,
+    FailIfNot(test.TestOnOffCluster(nodeid=device_nodeid,
                                     endpoint=LIGHTING_ENDPOINT_ID,
                                     group=GROUP_ID), "Failed to test on off cluster")
 
     logger.info("Testing level control cluster")
-    FailIfNot(test.TestLevelControlCluster(nodeid=1,
+    FailIfNot(test.TestLevelControlCluster(nodeid=device_nodeid,
                                            endpoint=LIGHTING_ENDPOINT_ID,
                                            group=GROUP_ID),
               "Failed to test level control cluster")
 
     logger.info("Testing sending commands to non exist endpoint")
-    FailIfNot(not test.TestOnOffCluster(nodeid=1,
+    FailIfNot(not test.TestOnOffCluster(nodeid=device_nodeid,
                                         endpoint=233,
                                         group=GROUP_ID), "Failed to test on off cluster on non-exist endpoint")
 
@@ -133,13 +132,13 @@ def main():
               "Failed when testing Python Cluster Object APIs")
 
     logger.info("Testing attribute reading")
-    FailIfNot(test.TestReadBasicAttributes(nodeid=1,
+    FailIfNot(test.TestReadBasicAttributes(nodeid=device_nodeid,
                                            endpoint=ENDPOINT_ID,
                                            group=GROUP_ID),
               "Failed to test Read Basic Attributes")
 
     logger.info("Testing attribute writing")
-    FailIfNot(test.TestWriteBasicAttributes(nodeid=1,
+    FailIfNot(test.TestWriteBasicAttributes(nodeid=device_nodeid,
                                             endpoint=ENDPOINT_ID,
                                             group=GROUP_ID),
               "Failed to test Write Basic Attributes")
@@ -151,18 +150,58 @@ def main():
               "Failed to test Read Basic Attributes")
 
     logger.info("Testing subscription")
-    FailIfNot(test.TestSubscription(nodeid=1, endpoint=LIGHTING_ENDPOINT_ID),
+    FailIfNot(test.TestSubscription(nodeid=device_nodeid, endpoint=LIGHTING_ENDPOINT_ID),
               "Failed to subscribe attributes.")
 
     logger.info("Testing another subscription that kills previous subscriptions")
-    FailIfNot(test.TestSubscription(nodeid=1, endpoint=LIGHTING_ENDPOINT_ID),
+    FailIfNot(test.TestSubscription(nodeid=device_nodeid, endpoint=LIGHTING_ENDPOINT_ID),
               "Failed to subscribe attributes.")
 
     logger.info("Testing on off cluster over resolved connection")
-    FailIfNot(test.TestOnOffCluster(nodeid=1,
+    FailIfNot(test.TestOnOffCluster(nodeid=device_nodeid,
                                     endpoint=LIGHTING_ENDPOINT_ID,
                                     group=GROUP_ID), "Failed to test on off cluster")
 
+
+def do_tests(controller_nodeid, device_nodeid, timeout, discriminator, setup_pin, commissioning, log_level):
+    timeoutTicker = TestTimeout(timeout)
+    timeoutTicker.start()
+
+    test = BaseTestHelper(nodeid=controller_nodeid)
+
+    coloredlogs.install(level='DEBUG')
+    chip.logging.RedirectToPythonLogging()
+
+    # logging.getLogger().setLevel(logging.DEBUG)
+    logging.getLogger().setLevel(log_level)
+
+    commissioning_method = None
+
+    if commissioning == 'eth':
+        commissioning_method = ethernet_commissioning
+    else:
+        TestFail(f"Unknown commissioning method: {commissioning}")
+
+    commissioning_method(test, discriminator, setup_pin,
+                         device_nodeid)
+
+    logger.info("Testing resolve")
+    FailIfNot(test.TestResolve(nodeid=device_nodeid),
+              "Failed to resolve nodeid")
+
+    # Still test network commissioning
+    if TestIsEnabled('nwprov'):
+        logger.info("Testing network commissioning")
+        FailIfNot(asyncio.run(NetworkCommissioningTests(devCtrl=test.devCtrl, nodeid=device_nodeid).run()),
+                  "Failed to finish network commissioning")
+
+    if TestIsEnabled('datamodel'):
+        logger.info("Testing datamodel functions")
+        test_datamodel(test, device_nodeid)
+
+    logger.info("Testing non-controller APIs")
+    FailIfNot(test.TestNonControllerAPIs(), "Non controller API test failed")
+
     timeoutTicker.stop()
 
     logger.info("Test finished")

src/controller/python/test/test_scripts/network_commissioning.py

@@ -295,6 +295,11 @@ async def test_thread(self, endpointId):
 
     async def run(self):
         try:
+            clusters = await self._devCtrl.ReadAttribute(nodeid=self._nodeid, attributes=[(Clusters.Descriptor.Attributes.ServerList)], returnClusterObject=True)
+            if Clusters.NetworkCommissioning.id not in clusters[0].serverList:
+                logger.info(
+                    f"Network commissioning cluster {endpoint} is not enabled on this device.")
+                return True
             endpoints = await self._devCtrl.ReadAttribute(nodeid=self._nodeid, attributes=[(Clusters.NetworkCommissioning.Attributes.FeatureMap)], returnClusterObject=True)
             logger.info(endpoints)
             endpoints = endpoints[0]

src/test_driver/linux-cirque/MobileDeviceTest.py

@@ -91,10 +91,9 @@ def run_controller_test(self):
         self.execute_device_cmd(req_device_id, "pip3 install {}".format(os.path.join(
             CHIP_REPO, "out/debug/linux_x64_gcc/controller/python/chip-0.0-cp37-abi3-linux_x86_64.whl")))
 
-        command = "gdb -return-child-result -q -ex run -ex bt --args python3 {} -t 150 -a {}".format(
+        command = "gdb -return-child-result -q -ex run -ex bt --args python3 {} -t 150".format(
             os.path.join(
-                CHIP_REPO, "src/controller/python/test/test_scripts/mobile-device-test.py"),
-            ethernet_ip)
+                CHIP_REPO, "src/controller/python/test/test_scripts/mobile-device-test.py"))
         ret = self.execute_device_cmd(req_device_id, command)
 
         self.assertEqual(ret['return_code'], '0',