Move CommissionDeviceTest class to commissioning.py (project-chip#39478)

* Moved class CommissionDeviceTest from matter_testing to commission.py. Added class global_stash to have stashed attirbutes outside matter_testing. Updated runner imports.

* fix: removed unused lines

* Added docstring.

* fix: gemini code suggestions.

* Removed clusterobjects from commissioning.py. Renamed get_setup_payload_info for get_setup_payload_info_config at commissioning.py

* Added # chipstack-ok at get_setup_payload_info_config

Author: jtrejoespinoza-grid

src/python_testing/matter_testing_infrastructure/BUILD.gn

@@ -43,6 +43,7 @@ pw_python_package("chip-testing-module") {
     "chip/testing/conversions.py",
     "chip/testing/decorators.py",
     "chip/testing/global_attribute_ids.py",
+    "chip/testing/global_stash.py",
     "chip/testing/matchers.py",
     "chip/testing/matter_asserts.py",
     "chip/testing/matter_testing.py",

src/python_testing/matter_testing_infrastructure/chip/testing/commissioning.py

@@ -21,12 +21,14 @@
 
 import logging
 from dataclasses import dataclass
-from typing import List, Optional
+from typing import Any, List, Optional
 
+import chip.testing.global_stash as global_stash
 from chip import ChipDeviceCtrl, discovery
 from chip.ChipDeviceCtrl import CommissioningParameters
-from chip.clusters import ClusterObjects as ClusterObjects
 from chip.exceptions import ChipStackError
+from chip.setup_payload import SetupPayload
+from mobly import asserts, base_test, signals
 
 logger = logging.getLogger("matter.python_testing")
 logger.setLevel(logging.INFO)
@@ -211,3 +213,100 @@ async def commission_devices(
         commissioned.append(await commission_device(dev_ctrl, node_id, setup_payload, commissioning_info))
 
     return all(commissioned)
+
+
+def get_setup_payload_info_config(matter_test_config: Any) -> List[SetupPayloadInfo]:
+    """
+    Get and builds the payload info provided in the execution.
+
+    Args:
+        matter_test_config: Matter test configuration object
+
+    Returns:
+         List[SetupPayloadInfo]: List of Payload used by the test case
+    """
+    setup_payloads = []
+    for qr_code in matter_test_config.qr_code_content:
+        try:
+            setup_payloads.append(SetupPayload().ParseQrCode(qr_code))
+        except ChipStackError:  # chipstack-ok: This disables ChipStackError linter check. Can not use 'with' because it is not expected to fail
+            asserts.fail(f"QR code '{qr_code} failed to parse properly as a Matter setup code.")
+
+    for manual_code in matter_test_config.manual_code:
+        try:
+            setup_payloads.append(SetupPayload().ParseManualPairingCode(manual_code))
+        except ChipStackError:  # chipstack-ok: This disables ChipStackError linter check. Can not use 'with' because it is not expected to fail
+            asserts.fail(
+                f"Manual code code '{manual_code}' failed to parse properly as a Matter setup code. Check that all digits are correct and length is 11 or 21 characters.")
+
+    infos = []
+    for setup_payload in setup_payloads:
+        info = SetupPayloadInfo()
+        info.passcode = setup_payload.setup_passcode
+        if setup_payload.short_discriminator is not None:
+            info.filter_type = discovery.FilterType.SHORT_DISCRIMINATOR
+            info.filter_value = setup_payload.short_discriminator
+        else:
+            info.filter_type = discovery.FilterType.LONG_DISCRIMINATOR
+            info.filter_value = setup_payload.long_discriminator
+        infos.append(info)
+
+    num_passcodes = 0 if matter_test_config.setup_passcodes is None else len(matter_test_config.setup_passcodes)
+    num_discriminators = 0 if matter_test_config.discriminators is None else len(matter_test_config.discriminators)
+    asserts.assert_equal(num_passcodes, num_discriminators, "Must have same number of discriminators as passcodes")
+    if matter_test_config.discriminators:
+        for idx, discriminator in enumerate(matter_test_config.discriminators):
+            info = SetupPayloadInfo()
+            info.passcode = matter_test_config.setup_passcodes[idx]
+            info.filter_type = DiscoveryFilterType.LONG_DISCRIMINATOR
+            info.filter_value = discriminator
+            infos.append(info)
+
+    return infos
+
+
+class CommissionDeviceTest(base_test.BaseTestClass):
+    """Test class auto-injected at the start of test list to commission a device when requested"""
+
+    def __init__(self, *args):
+        super().__init__(*args)
+        # This class is used to commission the device so is set to True
+        self.is_commissioning = True
+        # Save the stashed values into attributes to avoid mobly conflic with ctypes when mobly performs copy().
+        test_config = args[0]
+        self.default_controller = test_config.user_params['default_controller']
+        meta_config = test_config.user_params['meta_config']
+        self.dut_node_ids: List[int] = meta_config['dut_node_ids']
+        self.commissioning_info: CommissioningInfo = CommissioningInfo(
+            commissionee_ip_address_just_for_testing=meta_config['commissionee_ip_address_just_for_testing'],
+            commissioning_method=meta_config['commissioning_method'],
+            thread_operational_dataset=meta_config['thread_operational_dataset'],
+            wifi_passphrase=meta_config['wifi_passphrase'],
+            wifi_ssid=meta_config['wifi_ssid'],
+            tc_version_to_simulate=meta_config['tc_version_to_simulate'],
+            tc_user_response_to_simulate=meta_config['tc_user_response_to_simulate'],
+        )
+        self.setup_payloads: List[SetupPayloadInfo] = get_setup_payload_info_config(
+            global_stash.unstash_globally(test_config.user_params['matter_test_config']))
+
+    def test_run_commissioning(self):
+        """This method is the test called by mobly, which try to commission the device until is complete or raises an error.
+        Raises:
+            signals.TestAbortAll: Failed to commission node(s)
+        """
+        if not self.event_loop.run_until_complete(commission_devices(
+            dev_ctrl=self.default_controller,
+            dut_node_ids=self.dut_node_ids,
+            setup_payloads=self.setup_payloads,
+            commissioning_info=self.commissioning_info
+        )):
+            raise signals.TestAbortAll("Failed to commission node(s)")
+
+    # Default controller is used by commission_devices
+    @property
+    def default_controller(self) -> ChipDeviceCtrl.ChipDeviceController:
+        return global_stash.unstash_globally(self._default_controller)
+
+    @default_controller.setter
+    def default_controller(self, tmp_default_controller):
+        self._default_controller = tmp_default_controller

src/python_testing/matter_testing_infrastructure/chip/testing/global_stash.py

@@ -0,0 +1,39 @@
+#
+#    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.
+#
+
+"""
+This module contains functions designed to handle config values of ctypes objects as mobly cannot deal with those.
+The methods just use a global dict of uuid -> object to recover items stashed by reference.
+"""
+
+import uuid
+from typing import Any
+
+# Mobly cannot deal with user config passing of ctypes objects,
+# so we use this dict of uuid -> object to recover items stashed
+# by reference.
+_GLOBAL_DATA = {}
+
+
+def stash_globally(o: object) -> str:
+    unique_id = str(uuid.uuid1())
+    _GLOBAL_DATA[unique_id] = o
+    return unique_id
+
+
+def unstash_globally(id: str) -> Any:
+    return _GLOBAL_DATA.get(id)

src/python_testing/matter_testing_infrastructure/chip/testing/matter_testing.py

@@ -32,7 +32,6 @@
 import threading
 import time
 import typing
-import uuid
 from binascii import unhexlify
 from dataclasses import asdict as dataclass_asdict
 from dataclasses import dataclass, field
@@ -59,16 +58,15 @@
 import chip.clusters as Clusters
 import chip.logging
 import chip.native
-from chip import discovery
+import chip.testing.global_stash as global_stash
 from chip.ChipStack import ChipStack
-from chip.clusters import Attribute
-from chip.clusters import ClusterObjects as ClusterObjects
+from chip.clusters import Attribute, ClusterObjects
 from chip.clusters.Attribute import EventReadResult, SubscriptionTransaction, TypedAttributePath
-from chip.exceptions import ChipStackError
 from chip.interaction_model import InteractionModelError, Status
 from chip.setup_payload import SetupPayload
 from chip.storage import PersistentStorage
-from chip.testing.commissioning import CommissioningInfo, CustomCommissioningParameters, SetupPayloadInfo, commission_devices
+from chip.testing.commissioning import (CommissioningInfo, CustomCommissioningParameters, SetupPayloadInfo, commission_devices,
+                                        get_setup_payload_info_config)
 from chip.testing.global_attribute_ids import GlobalAttributeIds
 from chip.testing.pics import read_pics_from_file
 from chip.testing.runner import TestRunnerHooks, TestStep
@@ -90,21 +88,6 @@
 _DEFAULT_DUT_NODE_ID = 0x12344321
 _DEFAULT_TRUST_ROOT_INDEX = 1
 
-# Mobly cannot deal with user config passing of ctypes objects,
-# so we use this dict of uuid -> object to recover items stashed
-# by reference.
-_GLOBAL_DATA = {}
-
-
-def stash_globally(o: object) -> str:
-    id = str(uuid.uuid1())
-    _GLOBAL_DATA[id] = o
-    return id
-
-
-def unstash_globally(id: str) -> Any:
-    return _GLOBAL_DATA.get(id)
-
 
 def default_paa_rootstore_from_root(root_path: pathlib.Path) -> Optional[pathlib.Path]:
     """Attempt to find a PAA trust store following SDK convention at `root_path`
@@ -1040,23 +1023,23 @@ def default_timeout(self) -> int:
 
     @property
     def runner_hook(self) -> TestRunnerHooks:
-        return unstash_globally(self.user_params.get("hooks"))
+        return global_stash.unstash_globally(self.user_params.get("hooks"))
 
     @property
     def matter_test_config(self) -> MatterTestConfig:
-        return unstash_globally(self.user_params.get("matter_test_config"))
+        return global_stash.unstash_globally(self.user_params.get("matter_test_config"))
 
     @property
     def default_controller(self) -> ChipDeviceCtrl.ChipDeviceController:
-        return unstash_globally(self.user_params.get("default_controller"))
+        return global_stash.unstash_globally(self.user_params.get("default_controller"))
 
     @property
     def matter_stack(self) -> MatterStackState:
-        return unstash_globally(self.user_params.get("matter_stack"))
+        return global_stash.unstash_globally(self.user_params.get("matter_stack"))
 
     @property
     def certificate_authority_manager(self) -> chip.CertificateAuthority.CertificateAuthorityManager:
-        return unstash_globally(self.user_params.get("certificate_authority_manager"))
+        return global_stash.unstash_globally(self.user_params.get("certificate_authority_manager"))
 
     @property
     def dut_node_id(self) -> int:
@@ -1612,44 +1595,12 @@ def step(self, step: typing.Union[int, str]):
         self.step_skipped = False
 
     def get_setup_payload_info(self) -> List[SetupPayloadInfo]:
-        setup_payloads = []
-        for qr_code in self.matter_test_config.qr_code_content:
-            try:
-                setup_payloads.append(SetupPayload().ParseQrCode(qr_code))
-            except ChipStackError:  # chipstack-ok: This disables ChipStackError linter check. Can not use 'with' because it is not expected to fail
-                asserts.fail(f"QR code '{qr_code} failed to parse properly as a Matter setup code.")
-
-        for manual_code in self.matter_test_config.manual_code:
-            try:
-                setup_payloads.append(SetupPayload().ParseManualPairingCode(manual_code))
-            except ChipStackError:  # chipstack-ok: This disables ChipStackError linter check. Can not use 'with' because it is not expected to fail
-                asserts.fail(
-                    f"Manual code code '{manual_code}' failed to parse properly as a Matter setup code. Check that all digits are correct and length is 11 or 21 characters.")
-
-        infos = []
-        for setup_payload in setup_payloads:
-            info = SetupPayloadInfo()
-            info.passcode = setup_payload.setup_passcode
-            if setup_payload.short_discriminator is not None:
-                info.filter_type = discovery.FilterType.SHORT_DISCRIMINATOR
-                info.filter_value = setup_payload.short_discriminator
-            else:
-                info.filter_type = discovery.FilterType.LONG_DISCRIMINATOR
-                info.filter_value = setup_payload.long_discriminator
-            infos.append(info)
-
-        num_passcodes = 0 if self.matter_test_config.setup_passcodes is None else len(self.matter_test_config.setup_passcodes)
-        num_discriminators = 0 if self.matter_test_config.discriminators is None else len(self.matter_test_config.discriminators)
-        asserts.assert_equal(num_passcodes, num_discriminators, "Must have same number of discriminators as passcodes")
-        if self.matter_test_config.discriminators:
-            for idx, discriminator in enumerate(self.matter_test_config.discriminators):
-                info = SetupPayloadInfo()
-                info.passcode = self.matter_test_config.setup_passcodes[idx]
-                info.filter_type = DiscoveryFilterType.LONG_DISCRIMINATOR
-                info.filter_value = discriminator
-                infos.append(info)
-
-        return infos
+        """
+        Get and builds the payload info provided in the execution.
+        Returns:
+            List[SetupPayloadInfo]: List of Payload used by the test case
+        """
+        return get_setup_payload_info_config(self.matter_test_config)
 
     def wait_for_user_input(self,
                             prompt_msg: str,
@@ -2196,18 +2147,6 @@ async def _get_all_matching_endpoints(self: MatterBaseTest, accept_function: End
     return matching
 
 
-class CommissionDeviceTest(MatterBaseTest):
-    """Test class auto-injected at the start of test list to commission a device when requested"""
-
-    def __init__(self, *args):
-        super().__init__(*args)
-        self.is_commissioning = True
-
-    def test_run_commissioning(self):
-        if not self.event_loop.run_until_complete(self.commission_devices()):
-            raise signals.TestAbortAll("Failed to commission node(s)")
-
-
 # TODO(#37537): Remove these temporary aliases after transition period
 type_matches = matchers.is_type
 utc_time_in_matter_epoch = timeoperations.utc_time_in_matter_epoch

src/python_testing/matter_testing_infrastructure/chip/testing/runner.py

@@ -27,6 +27,7 @@
 from typing import Optional
 from unittest.mock import MagicMock
 
+import chip.testing.global_stash as global_stash
 from chip.clusters import Attribute
 from mobly import signals
 from mobly.config_parser import ENV_MOBLY_LOGPATH, TestRunConfig
@@ -311,15 +312,15 @@ def run_tests_no_exit(
     # Lazy import to avoid circular dependency
     from typing import TYPE_CHECKING
 
-    from chip.testing.matter_testing import MatterStackState, stash_globally
+    from chip.testing.matter_testing import MatterStackState
     if TYPE_CHECKING:
-        from chip.testing.matter_testing import CommissionDeviceTest
+        from chip.testing.commissioning import CommissionDeviceTest
     else:
         CommissionDeviceTest = None  # Initial placeholder
 
     # Actual runtime import
     if CommissionDeviceTest is None:
-        from chip.testing.matter_testing import CommissionDeviceTest
+        from chip.testing.commissioning import CommissionDeviceTest
 
     # NOTE: It's not possible to pass event loop via Mobly TestRunConfig user params, because the
     #       Mobly deep copies the user params before passing them to the test class and the event
@@ -347,7 +348,7 @@ def run_tests_no_exit(
         for destination in matter_test_config.trace_to:
             tracing_ctx.StartFromString(destination)
 
-        test_config.user_params["matter_stack"] = stash_globally(stack)
+        test_config.user_params["matter_stack"] = global_stash.stash_globally(stack)
 
         # TODO: Steer to right FabricAdmin!
         # TODO: If CASE Admin Subject is a CAT tag range, then make sure to
@@ -360,17 +361,16 @@ def run_tests_no_exit(
                 catTags=matter_test_config.controller_cat_tags,
                 dacRevocationSetPath=matter_test_config.dac_revocation_set_path if matter_test_config.dac_revocation_set_path else ""
             )
-        test_config.user_params["default_controller"] = stash_globally(
+        test_config.user_params["default_controller"] = global_stash.stash_globally(
             default_controller)
-
-        test_config.user_params["matter_test_config"] = stash_globally(
+        test_config.user_params["matter_test_config"] = global_stash.stash_globally(
             matter_test_config)
-        test_config.user_params["hooks"] = stash_globally(hooks)
+        test_config.user_params["hooks"] = global_stash.stash_globally(hooks)
 
         # Execute the test class with the config
         ok = True
 
-        test_config.user_params["certificate_authority_manager"] = stash_globally(
+        test_config.user_params["certificate_authority_manager"] = global_stash.stash_globally(
             stack.certificate_authority_manager)
 
         # Execute the test class with the config