[matter_yamltests] Add an abstract interface for pseudo clusters and a base class to register/use them

scripts/py_matter_yamltests/BUILD.gn

@@ -32,13 +32,20 @@ pw_python_package("matter_yamltests") {
     "matter_yamltests/fixes.py",
     "matter_yamltests/parser.py",
     "matter_yamltests/pics_checker.py",
+    "matter_yamltests/pseudo_clusters/__init__.py",
+    "matter_yamltests/pseudo_clusters/clusters/delay_commands.py",
+    "matter_yamltests/pseudo_clusters/clusters/log_commands.py",
+    "matter_yamltests/pseudo_clusters/clusters/system_commands.py",
+    "matter_yamltests/pseudo_clusters/pseudo_cluster.py",
+    "matter_yamltests/pseudo_clusters/pseudo_clusters.py",
   ]
 
   python_deps = [ "${chip_root}/scripts/py_matter_idl:matter_idl" ]
 
   tests = [
     "test_spec_definitions.py",
     "test_pics_checker.py",
+    "test_pseudo_clusters.py",
   ]
 
   # TODO: at a future time consider enabling all (* or missing) here to get

scripts/py_matter_yamltests/matter_yamltests/pseudo_clusters/clusters/delay_commands.py

@@ -0,0 +1,33 @@
+#
+#    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.
+
+import sys
+import time
+
+from ..pseudo_cluster import PseudoCluster
+
+
+class DelayCommands(PseudoCluster):
+    name = 'DelayCommands'
+
+    async def WaitForMs(self, request):
+        duration_in_ms = 0
+
+        for argument in request.arguments['values']:
+            if argument['name'] == 'ms':
+                duration_in_ms = argument['value']
+
+        sys.stdout.flush()
+        time.sleep(duration_in_ms / 1000)

scripts/py_matter_yamltests/matter_yamltests/pseudo_clusters/clusters/log_commands.py

@@ -0,0 +1,26 @@
+#
+#    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 ..pseudo_cluster import PseudoCluster
+
+
+class LogCommands(PseudoCluster):
+    name = 'LogCommands'
+
+    async def UserPrompt(self, request):
+        pass
+
+    async def Log(self, request):
+        pass

scripts/py_matter_yamltests/matter_yamltests/pseudo_clusters/clusters/system_commands.py

@@ -0,0 +1,106 @@
+#
+#    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.
+
+import sys
+import xmlrpc.client
+
+from ..pseudo_cluster import PseudoCluster
+
+DEFAULT_KEY = 'default'
+IP = '127.0.0.1'
+PORT = 9000
+
+if sys.platform == 'linux':
+    IP = '10.10.10.5'
+
+
+def make_url():
+    return 'http://' + IP + ':' + str(PORT) + '/'
+
+
+def get_register_key(request):
+    if request.arguments:
+        values = request.arguments['values']
+        for item in values:
+            name = item['name']
+            value = item['value']
+            if name == 'registerKey':
+                return value
+
+    return DEFAULT_KEY
+
+
+def get_options(request):
+    options = []
+
+    if request.arguments:
+        values = request.arguments['values']
+        for item in values:
+            name = item['name']
+            value = item['value']
+
+            if name == 'discriminator':
+                options.append('--discriminator')
+                options.append(str(value))
+            elif name == 'port':
+                options.append('--secured-device-port')
+                options.append(str(value))
+            elif name == 'kvs':
+                options.append('--KVS')
+                options.append(str(value))
+            elif name == 'minCommissioningTimeout':
+                options.append('--min_commissioning_timeout')
+                options.append(str(value))
+            elif name == 'filepath':
+                options.append('--filepath')
+                options.append(str(value))
+            elif name == 'otaDownloadPath':
+                options.append('--otaDownloadPath')
+                options.append(str(value))
+            elif name == 'registerKey':
+                pass
+            else:
+                raise KeyError(f'Unknown key: {name}')
+
+    return options
+
+
+class SystemCommands(PseudoCluster):
+    name = 'SystemCommands'
+
+    async def Start(self, request):
+        register_key = get_register_key(request)
+        options = get_options(request)
+
+        with xmlrpc.client.ServerProxy(make_url(), allow_none=True) as proxy:
+            proxy.start(register_key, options)
+
+    async def Stop(self, request):
+        register_key = get_register_key(request)
+
+        with xmlrpc.client.ServerProxy(make_url(), allow_none=True) as proxy:
+            proxy.stop(register_key)
+
+    async def Reboot(self, request):
+        register_key = get_register_key(request)
+
+        with xmlrpc.client.ServerProxy(make_url(), allow_none=True) as proxy:
+            proxy.reboot(register_key)
+
+    async def FactoryReset(self, request):
+        register_key = get_register_key(request)
+
+        with xmlrpc.client.ServerProxy(make_url(), allow_none=True) as proxy:
+            proxy.factoryReset(register_key)

scripts/py_matter_yamltests/matter_yamltests/pseudo_clusters/pseudo_cluster.py

@@ -0,0 +1,53 @@
+# 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 abc import ABC, abstractproperty
+
+
+class PseudoCluster(ABC):
+    """
+    PseudoCluster is an abstract interface that custom pseudo clusters
+    should inherit from.
+
+    The interface expose a name property that is used while looking
+    where to dispatch the test step to.
+
+    The implementation should then expose methods where the name match
+    the name used in the YAML test file.
+
+    For example, the 'CustomCommands' pseudo cluster can be implemented as:
+
+    class CustomCommand(PseudoCluster):
+        name = 'CustomCommands'
+
+        async def MyCustomMethod(self, request):
+            pass
+
+        async def MyCustomMethod(self, request):
+            pass
+
+    It can then be called from any test step as:
+
+    - label: "Call a custom method"
+      cluster: "CustomCommands"
+      command: "MyCustomMethod"
+      arguments:
+          values:
+              - name: "MyCustomParameter"
+                value: "this_is_a_custom_value"
+    """
+
+    @abstractproperty
+    def name(self):
+        pass

scripts/py_matter_yamltests/matter_yamltests/pseudo_clusters/pseudo_clusters.py

@@ -0,0 +1,53 @@
+#    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 .clusters.delay_commands import DelayCommands
+from .clusters.log_commands import LogCommands
+from .clusters.system_commands import SystemCommands
+from .pseudo_cluster import PseudoCluster
+
+
+class PseudoClusters:
+    def __init__(self, clusters: list[PseudoCluster]):
+        self.__clusters = clusters
+
+    def supports(self, request) -> bool:
+        return False if self.__get_command(request) is None else True
+
+    async def execute(self, request):
+        status = {'error': 'FAILURE'}
+
+        command = self.__get_command(request)
+        if command:
+            status = await command(request)
+            # If the command does not returns an error, it is considered a success.
+            if status == None:
+                status = {}
+
+        return status, []
+
+    def __get_command(self, request):
+        for cluster in self.__clusters:
+            if request.cluster == cluster.name and getattr(cluster, request.command, None):
+                return getattr(cluster, request.command)
+        return None
+
+
+def get_default_pseudo_clusters() -> PseudoClusters:
+    clusters = [
+        DelayCommands(),
+        LogCommands(),
+        SystemCommands()
+    ]
+    return PseudoClusters(clusters)

scripts/py_matter_yamltests/test_pseudo_clusters.py

@@ -0,0 +1,66 @@
+#!/usr/bin/env -S python3 -B
+#
+#    Copyright (c) 2022 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.
+
+import unittest
+
+from matter_yamltests.pseudo_clusters.pseudo_clusters import PseudoCluster, PseudoClusters
+
+
+class MockStep:
+    def __init__(self, cluster: str, command: str):
+        self.cluster = cluster
+        self.command = command
+
+
+class MyCluster(PseudoCluster):
+    name = 'MyCluster'
+
+    async def MyCommand(self, request):
+        pass
+
+    async def MyCommandWithCustomSuccess(self, request):
+        return 'CustomSuccess'
+
+
+unsupported_cluster_step = MockStep('UnsupportedCluster', 'MyCommand')
+unsupported_command_step = MockStep('MyCluster', 'UnsupportedCommand')
+supported_step = MockStep('MyCluster', 'MyCommand')
+supported_step_with_custom_success = MockStep(
+    'MyCluster', 'MyCommandWithCustomSuccess')
+
+default_failure = ({'error': 'FAILURE'}, [])
+default_success = ({}, [])
+custom_success = ('CustomSuccess', [])
+
+clusters = PseudoClusters([MyCluster()])
+
+
+class TestPseudoClusters(unittest.IsolatedAsyncioTestCase):
+    def test_supports(self):
+        self.assertFalse(clusters.supports(unsupported_cluster_step))
+        self.assertFalse(clusters.supports(unsupported_command_step))
+        self.assertTrue(clusters.supports(supported_step))
+        self.assertTrue(clusters.supports(supported_step_with_custom_success))
+
+    async def test_execute_return_value(self):
+        self.assertEqual(await clusters.execute(unsupported_cluster_step), default_failure)
+        self.assertEqual(await clusters.execute(unsupported_command_step), default_failure)
+        self.assertEqual(await clusters.execute(supported_step), default_success)
+        self.assertEqual(await clusters.execute(supported_step_with_custom_success), custom_success)
+
+
+if __name__ == '__main__':
+    unittest.main()