[Testing] Convenience property for checking PICS_SDK_CI_ONLY (project-chip#35860)

Author: arkq

docs/testing/python.md

@@ -628,7 +628,10 @@ example DUT on the host and includes factory reset support
 -   Don’t forget to set the PICS file to the ci-pics-values
 -   If there are steps in your test that will fail on CI (e.g. test vendor
     checks), gate them on the PICS_SDK_CI_ONLY
-    -   `is_ci = self.check_pics('PICS_SDK_CI_ONLY')`
+    -   ```python
+        if not self.is_pics_sdk_ci_only:
+            ...  # Step that will fail on CI
+        ```
 
 The CI test runner uses a structured environment setup that can be declared
 using structured comments at the top of the test file. To use this structured

src/python_testing/MinimalRepresentation.py

@@ -129,8 +129,7 @@ def test_MinimalRepresentation(self):
         # Before we can generate a minimal representation, we need to make sure that the device is conformant.
         # Otherwise, the values we extract aren't fully informative.
         ignore_in_progress = self.user_params.get("ignore_in_progress", False)
-        is_ci = self.check_pics('PICS_SDK_CI_ONLY')
-        representation = self.GenerateMinimals(ignore_in_progress, is_ci)
+        representation = self.GenerateMinimals(ignore_in_progress, self.is_pics_sdk_ci_only)
         print(type(representation[0]))
         self.PrettyPrintRepresentation(representation)
 

src/python_testing/TC_BRBINFO_4_1.py

@@ -198,7 +198,6 @@ def resume_th_icd_server(self, check_state):
 
     @async_test_body
     async def test_TC_BRBINFO_4_1(self):
-        self.is_ci = self.check_pics('PICS_SDK_CI_ONLY')
         icdm_cluster = Clusters.Objects.IcdManagement
         icdm_attributes = icdm_cluster.Attributes
         brb_info_cluster = Clusters.Objects.BridgedDeviceBasicInformation

src/python_testing/TC_CCTRL_2_2.py

@@ -127,8 +127,6 @@ def default_timeout(self) -> int:
 
     @run_if_endpoint_matches(has_cluster(Clusters.CommissionerControl))
     async def test_TC_CCTRL_2_2(self):
-        self.is_ci = self.check_pics('PICS_SDK_CI_ONLY')
-
         self.step(1)
         th_server_fabrics = await self.read_single_attribute_check_success(cluster=Clusters.OperationalCredentials, attribute=Clusters.OperationalCredentials.Attributes.Fabrics, dev_ctrl=self.TH_server_controller, node_id=self.server_nodeid, endpoint=0, fabric_filtered=False)
         self.step(2)
@@ -192,7 +190,7 @@ async def test_TC_CCTRL_2_2(self):
         await self.send_single_cmd(cmd)
 
         self.step(12)
-        if not self.is_ci:
+        if not self.is_pics_sdk_ci_only:
             self.wait_for_user_input("Approve Commissioning approval request using manufacturer specified mechanism")
 
         self.step(13)
@@ -248,7 +246,7 @@ async def test_TC_CCTRL_2_2(self):
 
         self.step(19)
         logging.info("Test now waits for 30 seconds")
-        if not self.is_ci:
+        if not self.is_pics_sdk_ci_only:
             time.sleep(30)
 
         self.step(20)
@@ -267,7 +265,7 @@ async def test_TC_CCTRL_2_2(self):
         await self.send_single_cmd(cmd)
 
         self.step(23)
-        if not self.is_ci:
+        if not self.is_pics_sdk_ci_only:
             self.wait_for_user_input("Approve Commissioning approval request using manufacturer specified mechanism")
 
         self.step(24)
@@ -295,13 +293,13 @@ async def test_TC_CCTRL_2_2(self):
         await self.send_single_cmd(cmd, dev_ctrl=self.TH_server_controller, node_id=self.server_nodeid, endpoint=0, timedRequestTimeoutMs=5000)
 
         self.step(27)
-        if not self.is_ci:
+        if not self.is_pics_sdk_ci_only:
             time.sleep(30)
 
         self.step(28)
         th_server_fabrics_new = await self.read_single_attribute_check_success(cluster=Clusters.OperationalCredentials, attribute=Clusters.OperationalCredentials.Attributes.Fabrics, dev_ctrl=self.TH_server_controller, node_id=self.server_nodeid, endpoint=0, fabric_filtered=False)
         # TODO: this should be mocked too.
-        if not self.is_ci:
+        if not self.is_pics_sdk_ci_only:
             asserts.assert_equal(len(th_server_fabrics) + 1, len(th_server_fabrics_new),
                                  "Unexpected number of fabrics on TH_SERVER")
 

src/python_testing/TC_CCTRL_2_3.py

@@ -111,7 +111,6 @@ def default_timeout(self) -> int:
 
     @run_if_endpoint_matches(has_cluster(Clusters.CommissionerControl))
     async def test_TC_CCTRL_2_3(self):
-        self.is_ci = self.check_pics('PICS_SDK_CI_ONLY')
 
         self.step(1)
         th_server_fabrics = await self.read_single_attribute_check_success(cluster=Clusters.OperationalCredentials, attribute=Clusters.OperationalCredentials.Attributes.Fabrics, dev_ctrl=self.TH_server_controller, node_id=self.server_nodeid, endpoint=0, fabric_filtered=False)
@@ -129,7 +128,7 @@ async def test_TC_CCTRL_2_3(self):
         await self.send_single_cmd(cmd=cmd)
 
         self.step(5)
-        if not self.is_ci:
+        if not self.is_pics_sdk_ci_only:
             self.wait_for_user_input("Approve Commissioning approval request using manufacturer specified mechanism")
 
         self.step(6)

src/python_testing/TC_DA_1_2.py

@@ -175,7 +175,6 @@ def steps_TC_DA_1_2(self):
 
     @async_test_body
     async def test_TC_DA_1_2(self):
-        is_ci = self.check_pics('PICS_SDK_CI_ONLY')
         cd_cert_dir = self.user_params.get("cd_cert_dir", 'credentials/development/cd-certs')
         post_cert_test = self.user_params.get("post_cert_test", False)
 
@@ -311,7 +310,7 @@ async def test_TC_DA_1_2(self):
         asserts.assert_equal(format_version, 1, "Format version is incorrect")
         self.step("6.2")
         asserts.assert_equal(vendor_id, basic_info_vendor_id, "Vendor ID is incorrect")
-        if not is_ci:
+        if not self.is_pics_sdk_ci_only:
             asserts.assert_in(vendor_id, range(1, 0xfff0), "Vendor ID is out of range")
         self.step("6.3")
         asserts.assert_true(basic_info_product_id in product_id_array, "Product ID not found in CD product array")
@@ -328,7 +327,7 @@ async def test_TC_DA_1_2(self):
         self.step("6.9")
         if post_cert_test:
             asserts.assert_equal(certification_type, 2, "Certification declaration is not marked as production.")
-        elif is_ci:
+        elif self.is_pics_sdk_ci_only:
             asserts.assert_in(certification_type, [0, 1, 2], "Certification type is out of range")
         else:
             asserts.assert_in(certification_type, [1, 2], "Certification type is out of range")

src/python_testing/TC_DeviceBasicComposition.py

@@ -455,8 +455,7 @@ class RequiredMandatoryAttribute:
 
         self.print_step(
             6, "Validate that none of the global attribute IDs contain values with prefixes outside of the allowed standard or MEI prefix range")
-        is_ci = self.check_pics('PICS_SDK_CI_ONLY')
-        if is_ci:
+        if self.is_pics_sdk_ci_only:
             # test vendor prefixes are allowed in the CI because we use them internally in examples
             bad_prefix_min = 0xFFF5_0000
         else:

src/python_testing/TC_DeviceConformance.py

@@ -356,9 +356,8 @@ def test_TC_IDM_10_2(self):
         # TODO: Turn this off after TE2
         # https://github.com/project-chip/connectedhomeip/issues/34615
         ignore_in_progress = self.user_params.get("ignore_in_progress", True)
-        is_ci = self.check_pics('PICS_SDK_CI_ONLY')
         allow_provisional = self.user_params.get("allow_provisional", False)
-        success, problems = self.check_conformance(ignore_in_progress, is_ci, allow_provisional)
+        success, problems = self.check_conformance(ignore_in_progress, self.is_pics_sdk_ci_only, allow_provisional)
         self.problems.extend(problems)
         if not success:
             self.fail_current_test("Problems with conformance")

src/python_testing/TC_MCORE_FS_1_1.py

@@ -133,7 +133,6 @@ def default_timeout(self) -> int:
 
     @async_test_body
     async def test_TC_MCORE_FS_1_1(self):
-        self.is_ci = self.check_pics('PICS_SDK_CI_ONLY')
         # TODO this value should either be determined or passed in from command line
         dut_commissioning_control_endpoint = 0
         self.step(1)
@@ -152,7 +151,7 @@ async def test_TC_MCORE_FS_1_1(self):
             requestID=good_request_id, vendorID=th_fsa_server_vid, productID=th_fsa_server_pid, label="Test Ecosystem")
         await self.send_single_cmd(cmd, endpoint=dut_commissioning_control_endpoint)
 
-        if not self.is_ci:
+        if not self.is_pics_sdk_ci_only:
             self.wait_for_user_input("Approve Commissioning approval request on DUT using manufacturer specified mechanism")
 
         if not events:
@@ -181,12 +180,12 @@ async def test_TC_MCORE_FS_1_1(self):
         await self.send_single_cmd(cmd, dev_ctrl=self.TH_server_controller, node_id=self.server_nodeid, endpoint=0, timedRequestTimeoutMs=5000)
 
         self.step("3c")
-        if not self.is_ci:
+        if not self.is_pics_sdk_ci_only:
             time.sleep(30)
 
         th_fsa_server_fabrics_new = await self.read_single_attribute_check_success(cluster=Clusters.OperationalCredentials, attribute=Clusters.OperationalCredentials.Attributes.Fabrics, dev_ctrl=self.TH_server_controller, node_id=self.server_nodeid, endpoint=0, fabric_filtered=False)
         # TODO: this should be mocked too.
-        if not self.is_ci:
+        if not self.is_pics_sdk_ci_only:
             asserts.assert_equal(len(th_fsa_server_fabrics) + 1, len(th_fsa_server_fabrics_new),
                                  "Unexpected number of fabrics on TH_SERVER")
 

src/python_testing/TC_MCORE_FS_1_2.py

@@ -148,7 +148,6 @@ def default_timeout(self) -> int:
 
     @async_test_body
     async def test_TC_MCORE_FS_1_2(self):
-        self.is_ci = self.check_pics('PICS_SDK_CI_ONLY')
 
         min_report_interval_sec = self.user_params.get("min_report_interval_sec", 0)
         max_report_interval_sec = self.user_params.get("max_report_interval_sec", 30)
@@ -176,7 +175,7 @@ async def test_TC_MCORE_FS_1_2(self):
         asserts.assert_true(type_matches(step_1_dut_parts_list, list), "PartsList is expected to be a list")
 
         self.step(2)
-        if not self.is_ci:
+        if not self.is_pics_sdk_ci_only:
             self._ask_for_vendor_commissioning_ux_operation(self.th_server_setup_params)
         else:
             self.dut_fsa_stdin.write(

src/python_testing/TC_MCORE_FS_1_3.py

@@ -124,7 +124,7 @@ async def commission_via_commissioner_control(self, controller_node_id: int, dev
             ),
         )
 
-        if not self.is_ci:
+        if not self.is_pics_sdk_ci_only:
             self.wait_for_user_input("Approve Commissioning Approval Request on DUT using manufacturer specified mechanism")
 
         resp = await self.send_single_cmd(
@@ -152,7 +152,6 @@ async def commission_via_commissioner_control(self, controller_node_id: int, dev
 
     @async_test_body
     async def test_TC_MCORE_FS_1_3(self):
-        self.is_ci = self.check_pics('PICS_SDK_CI_ONLY')
 
         # Commissioning - done
         self.step(0)

src/python_testing/TC_MCORE_FS_1_4.py

@@ -195,7 +195,6 @@ def steps_TC_MCORE_FS_1_4(self) -> list[TestStep]:
 
     @async_test_body
     async def test_TC_MCORE_FS_1_4(self):
-        self.is_ci = self.check_pics('PICS_SDK_CI_ONLY')
 
         # Commissioning - done
         self.step(0)
@@ -298,7 +297,7 @@ async def test_TC_MCORE_FS_1_4(self):
         self.step(4)
 
         # Commissioning TH_FSA_BRIDGE to DUT_FSA fabric.
-        if not self.is_ci:
+        if not self.is_pics_sdk_ci_only:
             self.wait_for_user_input(
                 f"Commission TH_FSA's aggregator on DUT using manufacturer specified mechanism.\n"
                 f"Use the following parameters:\n"
@@ -326,7 +325,7 @@ async def test_TC_MCORE_FS_1_4(self):
         ))
 
         # Synchronize TH_SERVER_NO_UID from TH_FSA to DUT_FSA fabric.
-        if not self.is_ci:
+        if not self.is_pics_sdk_ci_only:
             self.wait_for_user_input(
                 f"Synchronize endpoint from TH_FSA's aggregator to DUT using manufacturer specified mechanism.\n"
                 f"Use the following parameters:\n"

src/python_testing/TC_MCORE_FS_1_5.py

@@ -155,7 +155,6 @@ def default_timeout(self) -> int:
 
     @async_test_body
     async def test_TC_MCORE_FS_1_5(self):
-        self.is_ci = self.check_pics('PICS_SDK_CI_ONLY')
 
         min_report_interval_sec = 0
         max_report_interval_sec = 30
@@ -183,7 +182,7 @@ async def test_TC_MCORE_FS_1_5(self):
         asserts.assert_true(type_matches(step_1_dut_parts_list, list), "PartsList is expected to be a list")
 
         self.step(2)
-        if not self.is_ci:
+        if not self.is_pics_sdk_ci_only:
             self._ask_for_vendor_commissioning_ux_operation(self.th_server_setup_params)
         else:
             self.dut_fsa_stdin.write(

src/python_testing/TC_pics_checker.py

@@ -177,7 +177,7 @@ def test_TC_IDM_10_4(self):
                 self._check_and_record_errors(location, required, pics)
 
         self.step(7)
-        if self.check_pics('PICS_SDK_CI_ONLY'):
+        if self.is_pics_sdk_ci_only:
             self.record_error("PICS check", location=ProblemLocation(),
                               problem="PICS PICS_SDK_CI_ONLY found in PICS list. This PICS is disallowed for certification.")
             self.success = False

src/python_testing/drlk_2_x_common.py

@@ -121,7 +121,6 @@ async def teardown(self):
                                                      credentials=self.createdCredential, userIndex=self.user_index)
 
     async def run_drlk_test_common(self, lockUnlockCommand, lockUnlockCmdRspPICS, lockUnlockText, doAutoRelockTest):
-        is_ci = self.check_pics('PICS_SDK_CI_ONLY')
         self.clear_credential_and_user_flag = True
 
         # Allow for user overrides of these values
@@ -131,7 +130,7 @@ async def run_drlk_test_common(self, lockUnlockCommand, lockUnlockCmdRspPICS, lo
         userCodeTemporaryDisableTime = self.user_params.get("user_code_temporary_disable_time", 15)
         wrongCodeEntryLimit = self.user_params.get("wrong_code_entry_limit", 3)
         autoRelockTime = self.user_params.get("auto_relock_time", 60)
-        if is_ci:
+        if self.is_pics_sdk_ci_only:
             autoRelockTime = 10
 
         cluster = Clusters.Objects.DoorLock

src/python_testing/matter_testing_support.py

@@ -1008,7 +1008,7 @@ def write_to_app_pipe(self, command_dict: dict, app_pipe_name: Optional[str] = N
 
         Use the following environment variables:
 
-         - LINUX_DUT_IP 
+         - LINUX_DUT_IP
             * if not provided, the Matter app is assumed to run on the same machine as the test,
               such as during CI, and the commands are sent to it using a local named pipe
             * if provided, the commands for writing to the named pipe are forwarded to the DUT
@@ -1128,9 +1128,11 @@ def teardown_class(self):
         super().teardown_class()
 
     def check_pics(self, pics_key: str) -> bool:
-        picsd = self.matter_test_config.pics
-        pics_key = pics_key.strip()
-        return pics_key in picsd and picsd[pics_key]
+        return self.matter_test_config.pics.get(pics_key.strip(), False)
+
+    @property
+    def is_pics_sdk_ci_only(self) -> bool:
+        return self.check_pics('PICS_SDK_CI_ONLY')
 
     async def openCommissioningWindow(self, dev_ctrl: ChipDeviceCtrl, node_id: int) -> CustomCommissioningParameters:
         rnd_discriminator = random.randint(0, 4095)