[matter_yamltests] Add supports for OTA_SuccessfulTransfer specific S…

…ystemCommands and DelayCommands (#24659)

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

@@ -0,0 +1,124 @@
+#
+#    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])
+
+    def createOtaImage(request):
+        otaImageFilePath = _get_option(request, 'otaImageFilePath')
+        rawImageFilePath = _get_option(request, 'rawImageFilePath')
+        rawImageContent = _get_option(request, 'rawImageContent')
+
+        with xmlrpc.client.ServerProxy(_make_url(), allow_none=True) as proxy:
+            proxy.createOtaImage(otaImageFilePath, rawImageFilePath, rawImageContent)
+
+    def compareFiles(request):
+        file1 = _get_option(request, 'file1')
+        file2 = _get_option(request, 'file2')
+
+        with xmlrpc.client.ServerProxy(_make_url(), allow_none=True) as proxy:
+            proxy.compareFiles(file1, file2)

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,94 +13,27 @@
 #    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
+from .accessory_server_bridge import AccessoryServerBridge
 
 
 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)
+        AccessoryServerBridge.factoryReset(request)
+
+    async def CreateOtaImage(self, request):
+        AccessoryServerBridge.createOtaImage(request)
 
-        with xmlrpc.client.ServerProxy(make_url(), allow_none=True) as proxy:
-            proxy.factoryReset(register_key)
+    async def CompareFiles(self, request):
+        AccessoryServerBridge.compareFiles(request)

scripts/tests/chiptest/accessories.py

@@ -13,11 +13,17 @@
 #    See the License for the specific language governing permissions and
 #    limitations under the License.
 
+import filecmp
 import logging
+import os
+import subprocess
 import sys
 import threading
 from xmlrpc.server import SimpleXMLRPCServer
 
+_DEFAULT_CHIP_ROOT = os.path.abspath(
+    os.path.join(os.path.dirname(__file__), '..', '..', '..'))
+
 IP = '127.0.0.1'
 PORT = 9000
 
@@ -100,6 +106,26 @@ def waitForMessage(self, name, message):
             return accessory.waitForMessage(' '.join(message))
         return False
 
+    def createOtaImage(self, otaImageFilePath, rawImageFilePath, 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')
+        return True
+
+    def compareFiles(self, file1, file2):
+        if filecmp.cmp(file1, file2, shallow=False) is False:
+            raise Exception('Files %s and %s do not match' % (file1, file2))
+        return True
+
     def __startXMLRPCServer(self):
         self.server = SimpleXMLRPCServer((IP, PORT))
 
@@ -108,6 +134,8 @@ def __startXMLRPCServer(self):
         self.server.register_function(self.reboot, 'reboot')
         self.server.register_function(self.factoryReset, 'factoryReset')
         self.server.register_function(self.waitForMessage, 'waitForMessage')
+        self.server.register_function(self.compareFiles, 'compareFiles')
+        self.server.register_function(self.createOtaImage, 'createOtaImage')
 
         self.server_thread = threading.Thread(target=self.server.serve_forever)
         self.server_thread.start()

src/app/tests/suites/commands/system/scripts/CompareFiles.py

@@ -14,17 +14,18 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import filecmp
 import sys
+import xmlrpc.client
 
-file1 = sys.argv[1]
-file2 = sys.argv[2]
+IP = '127.0.0.1'
+PORT = 9000
 
+if sys.platform == 'linux':
+    IP = '10.10.10.5'
 
-def main():
-    if filecmp.cmp(file1, file2, shallow=False) is False:
-        raise Exception('Files %s and %s do not match' % (file1, file2))
-
-
-if __name__ == "__main__":
-    main()
+# Passing in sys.argv[2:] gets rid of the script name and key to the apps register. The remaining
+# values in the list are key-value pairs, e.g. [option1, value1, option2, value2, ...]
+with xmlrpc.client.ServerProxy('http://' + IP + ':' + str(PORT) + '/', allow_none=True) as proxy:
+    file1 = sys.argv[1]
+    file2 = sys.argv[2]
+    proxy.compareFiles(file1, file2)

src/app/tests/suites/commands/system/scripts/CreateOtaImage.py

@@ -14,32 +14,20 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import os
-import subprocess
 import sys
+import xmlrpc.client
 
-DEFAULT_CHIP_ROOT = os.path.abspath(
-    os.path.join(os.path.dirname(__file__), '..', '..', '..', '..', '..', '..', '..'))
+IP = '127.0.0.1'
+PORT = 9000
 
-otaImageFilePath = sys.argv[1]
-rawImageFilePath = sys.argv[2]
-rawImageContent = ' '.join(sys.argv[3:])
+if sys.platform == 'linux':
+    IP = '10.10.10.5'
 
+# Passing in sys.argv[2:] gets rid of the script name and key to the apps register. The remaining
+# values in the list are key-value pairs, e.g. [option1, value1, option2, value2, ...]
+with xmlrpc.client.ServerProxy('http://' + IP + ':' + str(PORT) + '/', allow_none=True) as proxy:
+    otaImageFilePath = sys.argv[1]
+    rawImageFilePath = sys.argv[2]
+    rawImageContent = ' '.join(sys.argv[3:])
 
-def main():
-    # 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')
-
-
-if __name__ == "__main__":
-    main()
+    proxy.createOtaImage(otaImageFilePath, rawImageFilePath, rawImageContent)