more things

Author: cecille

src/python_testing/matter_testing_support.py

@@ -1892,22 +1892,52 @@ def has_feature(cluster: ClusterObjects.ClusterObjectDescriptor, feature: IntFla
     return partial(_has_feature, cluster=cluster, feature=feature)
 
 
-async def should_run_test_on_endpoint(self: MatterBaseTest, accept_function: EndpointCheckFunction) -> list[uint]:
+async def _get_all_matching_endpoints(self: MatterBaseTest, accept_function: EndpointCheckFunction) -> list[uint]:
+    """ Returns a list of endpoints matching the accept condition. """
+    wildcard = await self.default_controller.Read(self.dut_node_id, [(Clusters.Descriptor), Attribute.AttributePath(None, None, GlobalAttributeIds.ATTRIBUTE_LIST_ID), Attribute.AttributePath(None, None, GlobalAttributeIds.FEATURE_MAP_ID), Attribute.AttributePath(None, None, GlobalAttributeIds.ACCEPTED_COMMAND_LIST_ID)])
+    matching = [e for e in wildcard.attributes.keys() if accept_function(wildcard, e)]
+    return matching
+
+
+async def should_run_test_on_endpoint(self: MatterBaseTest, accept_function: EndpointCheckFunction) -> bool:
     """ Helper function for the run_if_endpoint_matches decorator.
 
-        Returns a list of endpoints on which the test should be run given the accept_function for the test.
+        Returns True if self.matter_test_config.endpoint matches the accept function.
     """
-
     if self.matter_test_config.endpoint is None:
         msg = """
               The --endpoint flag is required for this test.
               """
         asserts.fail(msg)
+    matching = await (_get_all_matching_endpoints(self, accept_function))
+    return self.matter_test_config.endpoint in matching
 
-    wildcard = await self.default_controller.Read(self.dut_node_id, [(Clusters.Descriptor), Attribute.AttributePath(None, None, GlobalAttributeIds.ATTRIBUTE_LIST_ID), Attribute.AttributePath(None, None, GlobalAttributeIds.FEATURE_MAP_ID), Attribute.AttributePath(None, None, GlobalAttributeIds.ACCEPTED_COMMAND_LIST_ID)])
-    matching = [e for e in wildcard.attributes.keys() if accept_function(wildcard, e)]
 
-    return self.matter_test_config.endpoint in matching
+def run_on_singleton_matching_endpoint(accept_function: EndpointCheckFunction):
+    """ Test decorator for a test that needs to be run on the endpoint that matches the given accept function.
+
+        This decorator should be used for tests where the endpoint is not known a-priori (dynamic endpoints).
+        Note that currently this test is limited to devices with a SINGLE matching endpoint.
+    """
+    def run_on_singleton_matching_endpoint_internal(body):
+        def matching_runner(self: MatterBaseTest, *args, **kwargs):
+            runner_with_timeout = asyncio.wait_for(_get_all_matching_endpoints(self, accept_function), timeout=30)
+            matching = asyncio.run(runner_with_timeout)
+            asserts.assert_less_equal(len(matching), 1, "More than one matching endpoint found for singleton test.")
+            if not matching:
+                logging.info("Test is not applicable to any endpoint - skipping test")
+                asserts.skip('No endpoint matches test requirements')
+                return
+            # Exceptions should flow through, hence no except block
+            try:
+                old_endpoint = self.matter_test_config.endpoint
+                self.matter_test_config.endpoint = matching[0]
+                logging.info(f'Running test on endpoint {self.matter_test_config.endpoint}')
+                _async_runner(body, self, *args, **kwargs)
+            finally:
+                self.matter_test_config.endpoint = old_endpoint
+        return matching_runner
+    return run_on_singleton_matching_endpoint_internal
 
 
 def run_if_endpoint_matches(accept_function: EndpointCheckFunction):

src/python_testing/test_testing/TestDecorators.py

@@ -33,11 +33,13 @@
 
 try:
     from matter_testing_support import (MatterBaseTest, MatterTestConfig, async_test_body, has_attribute,
-                                        has_cluster, has_feature, run_if_endpoint_matches, should_run_test_on_endpoint)
+                                        has_cluster, has_feature, run_if_endpoint_matches, run_on_singleton_matching_endpoint,
+                                        should_run_test_on_endpoint)
 except ImportError:
     sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
     from matter_testing_support import (MatterBaseTest, MatterTestConfig, async_test_body, has_attribute,
-                                        has_cluster, has_feature, run_if_endpoint_matches, should_run_test_on_endpoint)
+                                        has_cluster, has_feature, run_if_endpoint_matches, run_on_singleton_matching_endpoint,
+                                        should_run_test_on_endpoint)
 
 from typing import Optional
 
@@ -222,6 +224,18 @@ async def test_fail_on_ep1(self):
         if self.matter_test_config.endpoint == 1:
             asserts.fail("Expected failure")
 
+    @run_on_singleton_matching_endpoint(has_cluster(Clusters.OnOff))
+    async def test_run_on_singleton_matching_endpoint(self):
+        pass
+
+    @run_on_singleton_matching_endpoint(has_cluster(Clusters.OnOff))
+    async def test_run_on_singleton_matching_endpoint_failure(self):
+        asserts.fail("Expected failure")
+
+    @run_on_singleton_matching_endpoint(has_attribute(Clusters.OnOff.Attributes.OffWaitTime))
+    async def test_no_run_on_singleton_matching_endpoint(self):
+        pass
+
 
 def main():
     failures = []
@@ -308,6 +322,64 @@ def check_all_skipped(test_name: str):
     if ok:
         failures.append(f"Did not get expected test assertion on {test_name}")
 
+    def run_singleton_dynamic(test_name: str, cluster_list: list[int]) -> tuple[bool, DecoratorTestRunnerHooks]:
+        nonlocal failures
+        read_resp = get_clusters(cluster_list)
+        test_runner.set_test('TestDecorators.py', 'TestDecorators', test_name)
+        test_runner.set_test_config(MatterTestConfig(endpoint=2))
+        hooks = DecoratorTestRunnerHooks()
+        ok = test_runner.run_test_with_mock_read(read_resp, hooks)
+        # for all tests, we need to ensure the endpoint was set back to the prior values
+        if test_runner.config.endpoint != 2:
+            failures.append(f"Dynamic tests {test_name} with clusters {cluster_list} did not set endpoint back to prior")
+        # All tests should have a start and a stop
+        started_ok = len(hooks.started) == 1
+        stopped_ok = hooks.stopped == 1
+        if not started_ok or not stopped_ok:
+            failures.append(
+                f'Hooks failure on {test_name}, Runs: {hooks.started}, skips: {hooks.skipped} stops: {hooks.stopped}')
+        return ok, hooks
+
+    def expect_success_dynamic(test_name: str, cluster_list: list[int]):
+        ok, hooks = run_singleton_dynamic(test_name, [0])
+        if not ok:
+            failures.append(f"Unexpected failure on {test_name} with cluster list {cluster_list}")
+        if hooks.skipped:
+            failures.append(f'Unexpected skip call on {test_name} with cluster list {cluster_list}')
+
+    def expect_failure_dynamic(test_name: str, cluster_list: list[int]):
+        ok = run_singleton_dynamic(test_name, [0])
+        if ok:
+            failures.append(f"Unexpected success on {test_name} with cluster list {cluster_list}")
+        if hooks.skipped:
+            # We don't expect a skip call because the test actually failed.
+            failures.append(f'Skip called for {test_name} with cluster list {cluster_list}')
+
+    def expect_skip_dynamic(test_name: str, cluster_list: list[int]):
+        ok = run_singleton_dynamic(test_name, [0])
+        if not ok:
+            failures.append(f"Unexpected failure on {test_name} with cluster list {cluster_list}")
+        if not hooks.skipped:
+            # We don't expect a skip call because the test actually failed.
+            failures.append(f'Skip not called for {test_name} with cluster list {cluster_list}')
+
+    test_name = 'test_run_on_singleton_matching_endpoint'
+    expect_success_dynamic(test_name, [0])
+    expect_success_dynamic(test_name, [1])
+    # expect failure because there is more than 1 endpoint
+    expect_failure_dynamic(test_name, [0, 1])
+
+    test_name = 'test_run_on_singleton_matching_endpoint_failure'
+    expect_failure_dynamic(test_name, [0])
+    expect_failure_dynamic(test_name, [1])
+    expect_failure_dynamic(test_name, [0, 1])
+
+    test_name = 'test_no_run_on_singleton_matching_endpoint'
+    # no failure, expect skips on single endpoints, expect asserts on multiple matching
+    expect_skip_dynamic(test_name, [0])
+    expect_skip_dynamic(test_name, [1])
+    expect_failure_dynamic(test_name, [0, 1])
+
     test_runner.Shutdown()
     print(
         f"Test of Decorators: test response incorrect: {len(failures)}")