[chiptool.py] Add some mechanism to add additional custom pseudo clus…

…ters by using a custom folder (#26694)

scripts/py_matter_yamltests/matter_yamltests/pseudo_clusters/pseudo_clusters.py

@@ -28,6 +28,9 @@ def __init__(self, clusters: list[PseudoCluster]):
     def supports(self, request) -> bool:
         return False if self.__get_command(request) is None else True
 
+    def add(self, cluster: PseudoCluster):
+        self.clusters.append(cluster)
+
     async def execute(self, request):
         status = {'error': 'FAILURE'}
 

scripts/tests/chiptest/__init__.py

@@ -136,6 +136,7 @@ def _GetInDevelopmentTests() -> Set[str]:
     return {
         "Test_AddNewFabricFromExistingFabric.yaml",     # chip-repl does not support GetCommissionerRootCertificate and IssueNocChain command
         "TestEqualities.yaml",              # chip-repl does not support pseudo-cluster commands that return a value
+        "TestExampleCluster.yaml",          # chip-repl does not load custom pseudo clusters
         "TestClientMonitoringCluster.yaml"  # Client Monitoring Tests need a rework after the XML update
     }
 

scripts/tests/yaml/chiptool.py

@@ -27,10 +27,12 @@
 from runner import CONTEXT_SETTINGS, chiptool, runner_base
 from tests_logger import TestColoredLogPrinter, WebSocketRunnerLogger
 
+_DEFAULT_EXTENSIONS_DIR = 'scripts/tests/yaml/extensions'
+
 
 @click.pass_context
-def send_yaml_command(ctx, test_name: str, server_path: str, server_arguments: str, pics: str, commands: list[str]):
-    kwargs = {'test_name': test_name, 'pics': pics}
+def send_yaml_command(ctx, test_name: str, server_path: str, server_arguments: str, pics: str, additional_pseudo_clusters_directory: str, commands: list[str]):
+    kwargs = {'test_name': test_name, 'pics': pics, 'additional_pseudo_clusters_directory': additional_pseudo_clusters_directory}
 
     index = 0
     while len(commands) - index > 1:
@@ -40,6 +42,7 @@ def send_yaml_command(ctx, test_name: str, server_path: str, server_arguments: s
 
     del ctx.params['commands']
     del ctx.params['pics']
+    del ctx.params['additional_pseudo_clusters_directory']
 
     return ctx.forward(chiptool)
 
@@ -89,6 +92,8 @@ def chiptool_runner_options(f):
                      help='Do not stop running the test suite on first error.')(f)
     f = click.option('--PICS', type=click.Path(exists=True), show_default=True, default=_DEFAULT_PICS_FILE,
                      help='Path to the PICS file to use.')(f)
+    f = click.option('--additional_pseudo_clusters_directory', type=click.Path(), show_default=True, default=_DEFAULT_EXTENSIONS_DIR,
+                     help='Path to a directory containing additional pseudo clusters.')(f)
     return f
 
 
@@ -120,14 +125,15 @@ def maybe_update_stop_on_error(ctx):
 @click.argument('commands', nargs=-1)
 @chiptool_runner_options
 @click.pass_context
-def chiptool_py(ctx, commands: list[str], server_path: str, server_name: str, server_arguments: str, trace_file: str, trace_decode: bool, delay_in_ms: int, continueonfailure: bool, pics: str):
+def chiptool_py(ctx, commands: list[str], server_path: str, server_name: str, server_arguments: str, trace_file: str, trace_decode: bool, delay_in_ms: int, continueonfailure: bool, pics: str, additional_pseudo_clusters_directory: str):
     success = False
 
     server_arguments = maybe_update_server_arguments(ctx)
     maybe_update_stop_on_error(ctx)
 
     if len(commands) > 1 and commands[0] == 'tests':
-        success = send_yaml_command(commands[1], server_path, server_arguments, pics, commands[2:])
+        success = send_yaml_command(commands[1], server_path, server_arguments, pics,
+                                    additional_pseudo_clusters_directory, commands[2:])
     else:
         if server_path is None and server_name:
             paths_finder = PathsFinder()

scripts/tests/yaml/extensions/example_cluster.py

@@ -0,0 +1,72 @@
+#
+#    Copyright (c) 2023 Project CHIP Authors
+#
+#    Licensed under the Apache License, Version 2.0 (the 'License');
+#    you may not use this file except in compliance with the License.
+#    You may obtain a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an 'AS IS' BASIS,
+#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#    See the License for the specific language governing permissions and
+#    limitations under the License.
+
+
+from matter_yamltests.pseudo_clusters.pseudo_cluster import PseudoCluster
+
+
+class example_cluster(PseudoCluster):
+    name = 'ExampleCluster'
+
+    async def CommandSuccess(self, request):
+        return {}
+
+    async def CommandError(self, request):
+        return {'error': 'UNSUPPORTED_COMMAND'}
+
+    async def CommandWithReturnValues(self, request):
+        rv = {
+            'BooleanTrue': True,
+            'BooleanFalse': False,
+            'PositiveNumber': 123456789,
+            'NegativeNumber': -123456789,
+            'ListOfBoolean': [True, False, True],
+            'Struct': {
+                'boolean': True,
+                'number': 1,
+                'struct': {
+                    'boolean': False,
+                    'number': 2,
+                },
+                'list': [
+                    {
+                        'boolean': True,
+                        'number': 1,
+                    },
+                    {
+                        'boolean': False,
+                        'number': 3,
+                    },
+                ]
+            }
+        }
+        return {'value': rv}
+
+    async def CommandWithInputValues(self, request):
+        try:
+            arg1 = int(self._get_argument_value(request, 'Argument1'))
+            arg2 = int(self._get_argument_value(request, 'Argument2'))
+        except NameError:
+            return {'error': 'INVALID_COMMAND'}
+
+        rv = {'ReturnValue': arg1 + arg2}
+        return {'value': rv}
+
+    def _get_argument_value(self, request, name):
+        for argument in request.arguments['values']:
+            if argument['name'] == name:
+                return argument['value']
+
+        raise NameError

scripts/tests/yaml/runner.py

@@ -16,6 +16,8 @@
 
 import relative_importer  # isort: split # noqa: F401
 
+import importlib
+import os
 import sys
 import traceback
 from dataclasses import dataclass
@@ -42,6 +44,21 @@
 _DEFAULT_PICS_FILE = 'src/app/tests/suites/certification/ci-pics-values'
 
 
+def get_custom_pseudo_clusters(additional_pseudo_clusters_directory: str):
+    clusters = get_default_pseudo_clusters()
+
+    if additional_pseudo_clusters_directory:
+        sys.path.insert(0, additional_pseudo_clusters_directory)
+        for filepath in os.listdir(additional_pseudo_clusters_directory):
+            if filepath != '__init__.py' and filepath[-3:] == '.py':
+                module = importlib.import_module(f'{filepath[:-3]}')
+                constructor = getattr(module, module.__name__)
+                if constructor:
+                    clusters.add(constructor())
+
+    return clusters
+
+
 def test_parser_options(f):
     f = click.option('--configuration_name', type=str, show_default=True, default=_DEFAULT_CONFIG_NAME,
                      help='Name of the collection configuration json file to use.')(f)
@@ -55,6 +72,8 @@ def test_parser_options(f):
                      help='Stop parsing on first error.')(f)
     f = click.option('--use_default_pseudo_clusters', type=bool, show_default=True, default=True,
                      help='If enable this option use the set of default clusters provided by the matter_yamltests package.')(f)
+    f = click.option('--additional_pseudo_clusters_directory', type=click.Path(), show_default=True, default=None,
+                     help='Path to a directory containing additional pseudo clusters.')(f)
     return f
 
 
@@ -229,8 +248,9 @@ def __add_custom_params(self, ctx):
 @click.argument('test_name')
 @test_parser_options
 @click.pass_context
-def runner_base(ctx, configuration_directory: str, test_name: str, configuration_name: str, pics: str, specifications_paths: str, stop_on_error: bool, use_default_pseudo_clusters: bool, **kwargs):
-    pseudo_clusters = get_default_pseudo_clusters() if use_default_pseudo_clusters else PseudoClusters([])
+def runner_base(ctx, configuration_directory: str, test_name: str, configuration_name: str, pics: str, specifications_paths: str, stop_on_error: bool, use_default_pseudo_clusters: bool, additional_pseudo_clusters_directory: str, **kwargs):
+    pseudo_clusters = get_custom_pseudo_clusters(
+        additional_pseudo_clusters_directory) if use_default_pseudo_clusters else PseudoClusters([])
     specifications = SpecDefinitionsFromPaths(specifications_paths.split(','), pseudo_clusters)
     tests_finder = TestsFinder(configuration_directory, configuration_name)
 

src/app/tests/suites/TestExampleCluster.yaml

@@ -0,0 +1,78 @@
+# Copyright (c) 2023 Project CHIP Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+name: Example Cluster Tests
+
+config:
+    cluster: "ExampleCluster"
+
+tests:
+    - label: "Call a command that returns a success"
+      command: "CommandSuccess"
+
+    - label: "Call a command that returns an error"
+      command: "CommandError"
+      response:
+          error: UNSUPPORTED_COMMAND
+
+    - label: "Call a command with some return values"
+      command: "CommandWithReturnValues"
+      response:
+          values:
+              - name: "BooleanTrue"
+                value: true
+              - name: "BooleanFalse"
+                value: false
+              - name: "PositiveNumber"
+                value: 123456789
+              - name: "NegativeNumber"
+                value: -123456789
+              - name: "ListOfBoolean"
+                value: [true, false, true]
+              - name: "Struct"
+                value:
+                    {
+                        "boolean": true,
+                        "number": 1,
+                        "struct": { "boolean": false, "number": 2 },
+                        "list":
+                            [
+                                { "boolean": true, "number": 1 },
+                                { "boolean": false, "number": 3 },
+                            ],
+                    }
+
+    - label: "Call a command with some input values"
+      command: "CommandWithInputValues"
+      arguments:
+          values:
+              - name: "Argument1"
+                value: 123
+              - name: "Argument2"
+                value: 456
+      response:
+          - values:
+                - name: "ReturnValue"
+                  value: 579
+
+    - label:
+          "Call a command with some input values but expect a failure this time
+          because some argument is missing"
+      command: "CommandWithInputValues"
+      arguments:
+          values:
+              - name: "Argument1"
+                value: 123
+      response:
+          error: INVALID_COMMAND