Add pseudo clusters with a set of default clusters for running some o…

…f the yaml tests (#24620)

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()