[matter_yamltests] Add supports for OTA_SuccessfulTransfer specific S…

…ystemCommands and DelayCommands

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

@@ -0,0 +1,109 @@
+#
+#    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
+
+_DEFAULT_KEY = 'default'
+_IP = '127.0.0.1'
+_PORT = 9000
+
+if sys.platform == 'linux':
+    _IP = '10.10.10.5'
+
+
+def _make_url():
+    return f'http://{_IP}:{_PORT}/'
+
+
+def _get_option(request, item_name: str, default_value=None):
+    if request.arguments:
+        values = request.arguments['values']
+        for item in values:
+            name = item['name']
+            value = item['value']
+            if name == item_name:
+                return value
+
+    return default_value
+
+
+def _get_start_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 AccessoryServerBridge():
+    def start(request):
+        register_key = _get_option(request, 'registerKey', _DEFAULT_KEY)
+        options = _get_start_options(request)
+
+        with xmlrpc.client.ServerProxy(_make_url(), allow_none=True) as proxy:
+            proxy.start(register_key, options)
+
+    def stop(request):
+        register_key = _get_option(request, 'registerKey', _DEFAULT_KEY)
+
+        with xmlrpc.client.ServerProxy(_make_url(), allow_none=True) as proxy:
+            proxy.stop(register_key)
+
+    def reboot(request):
+        register_key = _get_option(request, 'registerKey', _DEFAULT_KEY)
+
+        with xmlrpc.client.ServerProxy(_make_url(), allow_none=True) as proxy:
+            proxy.reboot(register_key)
+
+    def factoryReset(request):
+        register_key = _get_option(request, 'registerKey', _DEFAULT_KEY)
+
+        with xmlrpc.client.ServerProxy(_make_url(), allow_none=True) as proxy:
+            proxy.factoryReset(register_key)
+
+    def waitForMessage(request):
+        register_key = _get_option(request, 'registerKey', _DEFAULT_KEY)
+        message = _get_option(request, 'message')
+
+        with xmlrpc.client.ServerProxy(_make_url(), allow_none=True) as proxy:
+            proxy.waitForMessage(register_key, [message])

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

@@ -17,6 +17,7 @@
 import time
 
 from ..pseudo_cluster import PseudoCluster
+from .accessory_server_bridge import AccessoryServerBridge
 
 
 class DelayCommands(PseudoCluster):
@@ -31,3 +32,6 @@ async def WaitForMs(self, request):
 
         sys.stdout.flush()
         time.sleep(duration_in_ms / 1000)
+
+    async def WaitForMessage(self, request):
+        AccessoryServerBridge.waitForMessage(request)

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

@@ -13,64 +13,29 @@
 #    See the License for the specific language governing permissions and
 #    limitations under the License.
 
+import filecmp
+import os
+import subprocess
 import sys
-import xmlrpc.client
 
 from ..pseudo_cluster import PseudoCluster
+from .accessory_server_bridge import AccessoryServerBridge
 
-DEFAULT_KEY = 'default'
-IP = '127.0.0.1'
-PORT = 9000
+DEFAULT_CHIP_ROOT = os.path.abspath(
+    os.path.join(os.path.dirname(__file__), '..', '..', '..', '..', '..', '..', '..', '..'))
 
-if sys.platform == 'linux':
-    IP = '10.10.10.5'
 
+def _get_options(request, allowed_names):
+    options = {}
 
-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
+            if name in allowed_names:
+                options[name] = value
             else:
                 raise KeyError(f'Unknown key: {name}')
 
@@ -81,26 +46,49 @@ 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)
+        AccessoryServerBridge.start(request)
 
     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)
+        AccessoryServerBridge.stop(request)
 
     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)
+        AccessoryServerBridge.reboot(request)
 
     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)
+        AccessoryServerBridge.factoryReset(request)
+
+    async def CreateOtaImage(self, request):
+        allowed_options = [
+            'otaImageFilePath',
+            'rawImageFilePath',
+            'rawImageContent'
+        ]
+        options = _get_options(request, allowed_options)
+        otaImageFilePath = options.get('otaImageFilePath')
+        rawImageFilePath = options.get('rawImageFilePath')
+        rawImageContent = options.get('rawImageContent')
+
+        # Write the raw image content
+        with open(rawImageFilePath, 'w') as rawFile:
+            rawFile.write(rawImageContent)
+
+        # Add an OTA header to the raw file
+        otaImageTool = DEFAULT_CHIP_ROOT + '/src/app/ota_image_tool.py'
+        cmd = [otaImageTool, 'create', '-v', '0xDEAD', '-p', '0xBEEF', '-vn', '2',
+               '-vs', "2.0", '-da', 'sha256', rawImageFilePath, otaImageFilePath]
+        s = subprocess.Popen(cmd)
+        s.wait()
+        if s.returncode != 0:
+            raise Exception('Cannot create OTA image file')
+
+    async def CompareFiles(self, request):
+        allowed_options = [
+            'file1',
+            'file2'
+        ]
+        options = _get_options(request, allowed_options)
+        file1 = options.get('file1')
+        file2 = options.get('file2')
+
+        if filecmp.cmp(file1, file2, shallow=False) is False:
+            raise Exception('Files %s and %s do not match' % (file1, file2))