Software Update base class (project-chip#41145)

* SUBase Test class start

* OTA Provider Wrapper changes for SU 2.7

* Fix Wrapper updated

* Added kwargs as the params as we are just forwarding params

* Updates for OTAProviderSubprocess

* Treat the logs as separate.

* Changes for SoftwareUpdate base class

* Removed static values. Updated methods with defaults. Added commissionprovider method.

* Mypy fixes

* Fix the log_path in base class. Added some method description

* Updated name for TC_SUTestBase.py added TC_SUTestBase.py into not_automated at test_metadata.yaml

* Close fd if is not None

* Ignore TC_SUTestBase.py

* Methods updated

* Removed hardcoded from launch_provider, accept kwargs and filter arguments. Added method to get ota_image_path

* Fix ruff check

* Update wrapper: removed acl. Updated SUBaseTest added create_acl_entry and removed commisionprovider from base. Added Class ACL Handler into SuBaseTest as it will be used by in other tests.

* Removed unused import

* Update TC_SUTestBase.py

Fixed method name typo

* Restyled by autopep8

* Added Requestor to wrapper

* Added changes for RequestorSubProcess

* Update TC_SUTestBase.py

Fix: invalid method

* Removed requestor wrapper. Removed requestor start. Removed hardcoded paths. ota-image path and chip-ota-provider path will be provider to the test as an argument, Removed static versions version should be provided in the test as argument.

* Mypy fixes. Review fixes.

* Update append and read logs

* Fix typo in variable

* ANSI clean chagnes

* Dont delete the file

* Remove read_from_logs

* Remove ACLHandler Class and remove unused import

---------

Co-authored-by: Restyled.io <[email protected]>

Author: jtrejoespinoza-grid

src/python_testing/TC_SUTestBase.py

@@ -0,0 +1,302 @@
+#
+#    Copyright (c) 2025 Project CHIP Authors
+#    All rights reserved.
+#
+#    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 logging
+import tempfile
+from os import path
+from typing import Optional
+
+from mobly import asserts
+
+import matter.clusters as Clusters
+from matter import ChipDeviceCtrl
+from matter.clusters.Types import NullValue
+from matter.interaction_model import Status
+from matter.testing.apps import OtaImagePath, OTAProviderSubprocess
+from matter.testing.matter_testing import MatterBaseTest
+
+logger = logging.getLogger(__name__)
+
+
+class SoftwareUpdateBaseTest(MatterBaseTest):
+    """This is the base test class for SoftwareUpdate Test Cases"""
+
+    current_provider_app_proc: Optional[OTAProviderSubprocess] = None
+    provider_app_path: Optional[str] = None
+
+    def start_provider(self,
+                       provider_app_path: str = "",
+                       ota_image_path: str = "",
+                       setup_pincode: int = 20202021,
+                       discriminator: int = 1234,
+                       port: int = 5541,
+                       storage_dir='/tmp',
+                       extra_args: list = [],
+                       kvs_path: Optional[str] = None,
+                       log_file: Optional[str] = None, expected_output: str = "Status: Satisfied",
+                       timeout: int = 10):
+        """Start the provider process using the provided configuration.
+
+        Args:
+            provider_app_path (str): Path of Requestor app to load.
+            ota_image_path (str): Ota image to load within the provider.
+            setup_pincode (int, optional): Setup pincode for the provider process. Defaults to 20202021.
+            discriminator (int, optional): Discriminator for the provider process. Defaults to 1234.
+            port (int, optional): Port for the provider process. Defaults to 5541.
+            storage_dir (str, optional): Storage dir for the provider proccess. Defaults to '/tmp'.
+            extra_args (list, optional): Extra args to send to the provider process. Defaults to [].
+            kvs_path(str): Str of the path for the kvs path, if not will use temp file.
+            log_file (Optional[str], optional): Destination for the app process logs. Defaults to None.
+            expected_output (str): Expected string to see after a default timeout. Defaults to "Status: Satisfied".
+            timeout (int): Timeout to wait for the expected output. Defaults to 10 seconds
+        """
+        logger.info(f"Launching provider app with with ota image {ota_image_path}")
+        # Image to launch
+        self.provider_app_path = provider_app_path
+        if not path.exists(provider_app_path):
+            raise FileNotFoundError(f"Provider app not found {provider_app_path}")
+
+        if not path.exists(ota_image_path):
+            raise FileNotFoundError(f"Ota image provided does not exists {ota_image_path}")
+
+        # Ota image
+        ota_image_path = OtaImagePath(path=ota_image_path)
+        # Ideally we send the logs to a fixed location to avoid conflicts
+
+        if log_file is None:
+            # Assign the file descriptor to log_file
+            log_file = tempfile.NamedTemporaryFile(
+                dir=storage_dir, prefix='provider_', suffix='.log', mode='ab')
+            logger.info(f"Writing Provider logs at :{log_file.name}")
+        else:
+            logger.info(f"Writing Provider logs at : {log_file}")
+        # Launch the Provider subprocess using the Wrapper
+        proc = OTAProviderSubprocess(
+            provider_app_path,
+            storage_dir=storage_dir,
+            port=port,
+            discriminator=discriminator,
+            passcode=setup_pincode,
+            ota_source=ota_image_path,
+            extra_args=extra_args,
+            kvs_path=kvs_path,
+            log_file=log_file,
+            err_log_file=log_file)
+        proc.start(
+            expected_output=expected_output,
+            timeout=timeout)
+
+        self.current_provider_app_proc = proc
+        logger.info(f"Provider started with PID:  {self.current_provider_app_proc.get_pid()}")
+
+    async def announce_ota_provider(self,
+                                    controller: ChipDeviceCtrl,
+                                    provider_node_id: int,
+                                    requestor_node_id: int,
+                                    reason: Clusters.OtaSoftwareUpdateRequestor.Enums.AnnouncementReasonEnum = Clusters.OtaSoftwareUpdateRequestor.Enums.AnnouncementReasonEnum.kUpdateAvailable,
+                                    vendor_id: int = 0xFFF1,
+                                    endpoint: int = 0):
+        """ Launch the requestor.AnnounceOTAProvider method with the specific configuration.
+            Starts the communication from the requestor to the provider to start a software update.
+        Args:
+            controller (ChipDeviceCtrl): Controller for DUT
+            provider_node_id (int): Node id for the provider
+            requestor_node_id (int): Node id for the requestor
+            reason (Clusters.OtaSoftwareUpdateRequestor.Enums.AnnouncementReasonEnum, optional): Update Reason. Defaults to Clusters.OtaSoftwareUpdateRequestor.Enums.AnnouncementReasonEnum.kUpdateAvailable.
+            vendor_id (int, optional): Vendor id. Defaults to 0xFFF1.
+            endpoint (int, optional): Endpoint id. Defaults to 0.
+
+        Returns:
+            object: Return the data from the OtaSoftwareUpdateRequestor.AnnounceOTAProvider command.
+        """
+        cmd_announce_ota_provider = Clusters.OtaSoftwareUpdateRequestor.Commands.AnnounceOTAProvider(
+            providerNodeID=provider_node_id,
+            vendorID=vendor_id,
+            announcementReason=reason,
+            metadataForNode=None,
+            endpoint=endpoint
+        )
+        logger.info("Sending AnnounceOTA Provider Command")
+        cmd_resp = await self.send_single_cmd(
+            cmd=cmd_announce_ota_provider,
+            dev_ctrl=controller,
+            node_id=requestor_node_id,
+            endpoint=endpoint,
+        )
+        logger.info(f"Announce command sent {cmd_resp}")
+        return cmd_resp
+
+    async def set_default_ota_providers_list(self, controller: ChipDeviceCtrl, provider_node_id: int, requestor_node_id: int, endpoint: int = 0):
+        """Write the provider list in the requestor to initiate the Software Update.
+
+        Args:
+            controller (ChipDeviceCtrl): Controller to write the providers.
+            provider_node_id (int): Node where the provider is localted.
+            requestor_node_id (int): Node of the requestor to write the providers.
+            endpoint (int, optional): Endpoint to write the providerss. Defaults to 0.
+        """
+
+        current_otap_info = await self.read_single_attribute_check_success(
+            dev_ctrl=controller,
+            cluster=Clusters.OtaSoftwareUpdateRequestor,
+            attribute=Clusters.OtaSoftwareUpdateRequestor.Attributes.DefaultOTAProviders
+        )
+        logger.info(f"OTA Providers: {current_otap_info}")
+
+        # Create Provider Location into Requestor
+        provider_location_struct = Clusters.OtaSoftwareUpdateRequestor.Structs.ProviderLocation(
+            providerNodeID=provider_node_id,
+            endpoint=endpoint,
+            fabricIndex=controller.fabricId
+        )
+
+        # Create the OTA Provider Attribute
+        ota_providers_attr = Clusters.OtaSoftwareUpdateRequestor.Attributes.DefaultOTAProviders(value=[provider_location_struct])
+
+        # Write the Attribute
+        resp = await controller.WriteAttribute(
+            attributes=[(endpoint, ota_providers_attr)],
+            nodeid=requestor_node_id,
+        )
+        asserts.assert_equal(resp[0].Status, Status.Success, "Failed to write Default OTA Providers Attribute")
+
+        # Read Updated OTAProviders
+        after_otap_info = await self.read_single_attribute_check_success(
+            dev_ctrl=controller,
+            cluster=Clusters.OtaSoftwareUpdateRequestor,
+            attribute=Clusters.OtaSoftwareUpdateRequestor.Attributes.DefaultOTAProviders
+        )
+        logger.info(f"OTA Providers List: {after_otap_info}")
+
+    async def verify_version_applied_basic_information(self, controller: ChipDeviceCtrl, node_id: int, target_version: int):
+        """Verify the version from the BasicInformationCluster and compares against the provider target version.
+
+        Args:
+            controller (ChipDeviceCtrl): Controller 
+            node_id (int): Node to request
+            target_version (int): Version to compare
+        """
+
+        basicinfo_softwareversion = await self.read_single_attribute_check_success(
+            dev_ctrl=controller,
+            cluster=Clusters.BasicInformation,
+            attribute=Clusters.BasicInformation.Attributes.SoftwareVersion,
+            node_id=node_id)
+        asserts.assert_equal(basicinfo_softwareversion, target_version,
+                             f"Version from basic info cluster is not {target_version}, current cluster version is {basicinfo_softwareversion}")
+
+    def get_downloaded_ota_image_info(self, ota_path='/tmp/test.bin') -> dict:
+        """Return the data of the downloaded image from the provider.
+
+        Args:
+            ota_path (str, optional): _description_. Defaults to '/tmp/test.bin'.
+
+        Returns:
+            dict: Dict with the image info.
+        """
+        ota_image_info = {
+            "path": ota_path,
+            "exists": False,
+            "size": 0,
+        }
+        try:
+            ota_image_info['size'] = path.getsize(ota_path)
+            ota_image_info['exists'] = True
+        except OSError:
+            logger.info(f"OTA IMAGE at {ota_path} does not exists")
+            return ota_image_info
+
+        return ota_image_info
+
+    def verify_state_transition_event(self,
+                                      event_report: Clusters.OtaSoftwareUpdateRequestor.Events.StateTransition,
+                                      expected_previous_state,
+                                      expected_new_state,
+                                      expected_target_version: Optional[int] = None,
+                                      expected_reason: Optional[int] = None):
+        """Verify the values of the StateTransitionEvent from the EventHandler given the provided arguments.
+
+        Args:
+            event_report (Clusters.OtaSoftwareUpdateRequestor.Events.StateTransition): StateTransition Event report to verify.
+            previous_state (UpdateStateEnum:int): Int or UpdateStateEnum value for the previous state.
+            new_state (UpdateStateEnum:int): Int or UpdateStateEnum value for the new or current state.
+            target_version (Optional[int], optional): Software version to verify if not provided ignore this check.. Defaults to None.
+            reason (Optional[int], optional): UpdateStateEnum reason of the event, if not provided ignore. Defaults to None.
+        """
+        logger.info(f"Verifying the event {event_report}")
+        asserts.assert_equal(event_report.previousState, expected_previous_state,
+                             f"Previous state was not {expected_previous_state}")
+        asserts.assert_equal(event_report.newState,  expected_new_state, f"New state is not {expected_new_state}")
+        if expected_target_version is not None:
+            asserts.assert_equal(event_report.targetSoftwareVersion,  expected_target_version,
+                                 f"Target version is not {expected_target_version}")
+        if expected_reason is not None:
+            asserts.assert_equal(event_report.reason,  expected_reason, f"Reason is not {expected_reason}")
+
+    def create_acl_entry(self,
+                         dev_ctrl: ChipDeviceCtrl.ChipDeviceController,
+                         provider_node_id: int,
+                         requestor_node_id: Optional[int] = None,
+                         acl_entries: Optional[list[Clusters.AccessControl.Structs.AccessControlEntryStruct]] = None,
+                         ):
+        """Create ACL entries to allow OTA requestors to access the provider.
+
+        Args:
+            dev_ctrl: Device controller for sending commands
+            provider_node_id: Node ID of the OTA provider
+            requestor_node_id: Optional specific requestor node ID for targeted access
+            acl_entries: Optional[list[Clusters.AccessControl.Structs.AccessControlEntryStruct]]. ACL list to write ino the requestor.
+
+        Returns:
+            Result of the ACL write operation
+        """
+        # Standard ACL entry for OTA Provider cluster
+        admin_node_id = dev_ctrl.nodeId if hasattr(dev_ctrl, 'nodeId') else self.DEFAULT_ADMIN_NODE_ID
+        requestor_subjects = [requestor_node_id] if requestor_node_id else NullValue
+
+        if acl_entries is None:
+            # If there are not ACL entries using proper struct constructors create the default.
+            acl_entries = [
+                # Admin entry
+                Clusters.AccessControl.Structs.AccessControlEntryStruct(  # type: ignore
+                    privilege=Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kAdminister,  # type: ignore
+                    authMode=Clusters.AccessControl.Enums.AccessControlEntryAuthModeEnum.kCase,  # type: ignore
+                    subjects=[admin_node_id],  # type: ignore
+                    targets=NullValue
+                ),
+                # Operate entry
+                Clusters.AccessControl.Structs.AccessControlEntryStruct(  # type: ignore
+                    privilege=Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kOperate,  # type: ignore
+                    authMode=Clusters.AccessControl.Enums.AccessControlEntryAuthModeEnum.kCase,  # type: ignore
+                    subjects=requestor_subjects,  # type: ignore
+                    targets=[
+                        Clusters.AccessControl.Structs.AccessControlTargetStruct(  # type: ignore
+                            cluster=Clusters.OtaSoftwareUpdateProvider.id,  # type: ignore
+                            endpoint=NullValue,
+                            deviceType=NullValue
+                        )
+                    ],
+                )
+            ]
+
+        # Create the attribute descriptor for the ACL attribute
+        acl_attribute = Clusters.AccessControl.Attributes.Acl(acl_entries)
+
+        return dev_ctrl.WriteAttribute(
+            nodeid=provider_node_id,
+            attributes=[(0, acl_attribute)]
+        )