Add jfds_2.1 and jfds_2.2 test cases (project-chip#39901)

* Add jfds_2.1 and jfds_2.2 test cases

* Restyled by isort

* Comment unused import until steps are available

---------

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

Author: dparausanu

src/python_testing/TC_JFDS_2_1.py

@@ -0,0 +1,193 @@
+#
+#    Copyright (c) 2024 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 test requires a TH_SERVER application. Please specify with --string-arg th_server_app_path:<path_to_app>
+
+# See https://github.com/project-chip/connectedhomeip/blob/master/docs/testing/python.md#defining-the-ci-test-arguments
+# for details about the block below.
+#
+# === BEGIN CI TEST ARGUMENTS ===
+# test-runner-runs:
+#   run1:
+#     script-args: >
+#       --string-arg jfa_server_app:${JF_ADMIN_APP}
+#       --string-arg jfc_server_app:${JF_CONTROL_APP}
+#       --trace-to json:${TRACE_TEST_JSON}.json
+#       --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto
+#     factory-reset: true
+#     quiet: true
+# === END CI TEST ARGUMENTS ===
+
+import base64
+import logging
+import os
+import random
+import tempfile
+from configparser import ConfigParser
+
+import chip.clusters as Clusters
+from chip import CertificateAuthority
+from chip.storage import PersistentStorage
+from chip.testing.apps import AppServerSubprocess, JFControllerSubprocess
+from chip.testing.matter_testing import MatterBaseTest, TestStep, async_test_body, default_matter_test_main
+from mobly import asserts
+
+
+class TC_JFDS_2_1(MatterBaseTest):
+
+    @async_test_body
+    async def setup_class(self):
+        super().setup_class()
+
+        self.fabric_a_ctrl = None
+        self.storage_fabric_a = self.user_params.get("fabric_a_storage", None)
+        self.fabric_a_server_app = None
+
+        jfc_server_app = self.user_params.get("jfc_server_app", None)
+        if not jfc_server_app:
+            asserts.fail("This test requires a Joint Fabric Controller app. Specify app path with --string-arg jfc_server_app:<path_to_app>")
+        if not os.path.exists(jfc_server_app):
+            asserts.fail(f"The path {jfc_server_app} does not exist")
+
+        jfa_server_app = self.user_params.get("jfa_server_app", None)
+        if not jfa_server_app:
+            asserts.fail("This test requires a Joint Fabrics Admin app. Specify app path with --string-arg jfa_server_app:<path_to_app>")
+        if not os.path.exists(jfa_server_app):
+            asserts.fail(f"The path {jfa_server_app} does not exist")
+
+        # Create a temporary storage directory for both ecosystems to keep KVS files if not already provided by user.
+        if self.storage_fabric_a is None:
+            self.storage_directory_ecosystem_a = tempfile.TemporaryDirectory(prefix=self.__class__.__name__+"_A_")
+            self.storage_fabric_a = self.storage_directory_ecosystem_a.name
+            logging.info("Temporary storage directory: %s", self.storage_fabric_a)
+
+        #####################################################################################################################################
+        #
+        # Initialize Ecosystem A
+        #
+        #####################################################################################################################################
+        self.jfadmin_fabric_a_passcode = random.randint(110220011, 110220999)
+        self.jfctrl_fabric_a_vid = random.randint(0x0001, 0xFFF0)
+
+        # Start Fabric A JF-Administrator App
+        self.fabric_a_admin = AppServerSubprocess(
+            jfa_server_app,
+            storage_dir=self.storage_fabric_a,
+            port=random.randint(5001, 5999),
+            discriminator=random.randint(0, 4095),
+            passcode=self.jfadmin_fabric_a_passcode,
+            extra_args=["--capabilities", "0x04", "--rpc-server-port", "33033"])
+        self.fabric_a_admin.start(
+            expected_output="Server initialization complete",
+            timeout=10)
+
+        # Start Fabric A JF-Controller App
+        self.fabric_a_ctrl = JFControllerSubprocess(
+            jfc_server_app,
+            rpc_server_port=33033,
+            storage_dir=self.storage_fabric_a,
+            vendor_id=self.jfctrl_fabric_a_vid)
+        self.fabric_a_ctrl.start(
+            expected_output="CHIP task running",
+            timeout=10)
+
+        # Commission JF-ADMIN app with JF-Controller on Fabric A
+        self.fabric_a_ctrl.send(
+            message=f"pairing onnetwork 1 {self.jfadmin_fabric_a_passcode} --anchor true",
+            expected_output="[JF] Anchor Administrator commissioned with sucess")
+
+        # Extract the Ecosystem A certificates and inject them in the storage that will be provided to a new Python Controller later
+        jfcStorage = ConfigParser()
+        jfcStorage.read(self.storage_fabric_a+'/chip_tool_config.alpha.ini')
+        self.ecoACtrlStorage = {
+            "sdk-config": {
+                "ExampleOpCredsCAKey1": jfcStorage.get("Default", "ExampleOpCredsCAKey0"),
+                "ExampleOpCredsICAKey1": jfcStorage.get("Default", "ExampleOpCredsICAKey0"),
+                "ExampleCARootCert1": jfcStorage.get("Default", "ExampleCARootCert0"),
+                "ExampleCAIntermediateCert1": jfcStorage.get("Default", "ExampleCAIntermediateCert0"),
+            },
+            "repl-config": {
+                "caList": {
+                    "1": [
+                        {
+                            "fabricId": 1,
+                            "vendorId": self.jfctrl_fabric_a_vid
+                        }
+                    ]
+                }
+            }
+        }
+        # Extract CATs to be provided to the Python Controller later
+        self.ecoACATs = base64.b64decode(jfcStorage.get("Default", "CommissionerCATs"))[::-1].hex().strip('0')
+
+    def teardown_class(self):
+        # Stop all Subprocesses that were started in this test case
+        if self.fabric_a_admin is not None:
+            self.fabric_a_admin.terminate()
+        if self.fabric_a_ctrl is not None:
+            self.fabric_a_ctrl.terminate()
+        if self.fabric_a_server_app is not None:
+            self.fabric_a_server_app.terminate()
+
+        super().teardown_class()
+
+    def steps_TC_JFDS_2_1(self) -> list[TestStep]:
+        return [
+            TestStep("1", "TH reads AnchorRootCA attribute from DUT",
+                     "Verify that a RootCA is returned"),
+            TestStep("2", "TH reads AnchorNodeID attribute from DUT",
+                     "Verify that the DUT NodeId is returned"),
+            TestStep("3", "TH reads AnchorVendorID attribute from DUT",
+                     "Verify that the VendorId of the DUT is returned")
+            # TestStep("4", "{PLACEHOLDER_NOT_IMPLEMENTED]TH reads FriendlyName from DUT",
+            #          "Verify that the a valid string is returned")
+        ]
+
+    @async_test_body
+    async def test_TC_JFDS_2_1(self):
+        # Creating a Controller for Ecosystem A
+        _fabric_a_persistent_storage = PersistentStorage(jsonData=self.ecoACtrlStorage)
+        _certAuthorityManagerA = CertificateAuthority.CertificateAuthorityManager(
+            chipStack=self.matter_stack._chip_stack,
+            persistentStorage=_fabric_a_persistent_storage)
+        _certAuthorityManagerA.LoadAuthoritiesFromStorage()
+        devCtrlEcoA = _certAuthorityManagerA.activeCaList[0].adminList[0].NewController(
+            nodeId=101,
+            paaTrustStorePath=str(self.matter_test_config.paa_trust_store_path),
+            catTags=[int(self.ecoACATs, 16)])
+
+        self.step("1")
+        response = await devCtrlEcoA.ReadAttribute(
+            nodeid=1, attributes=[(1, Clusters.JointFabricDatastore.Attributes.AnchorRootCA)],
+            returnClusterObject=True)
+        asserts.assert_greater_equal(len(response[1][Clusters.JointFabricDatastore].anchorRootCA), 0)
+
+        self.step("2")
+        response = await devCtrlEcoA.ReadAttribute(
+            nodeid=1, attributes=[(1, Clusters.JointFabricDatastore.Attributes.AnchorNodeID)],
+            returnClusterObject=True)
+        asserts.assert_greater_equal(response[1][Clusters.JointFabricDatastore].anchorNodeID, 0)
+
+        self.step("3")
+        response = await devCtrlEcoA.ReadAttribute(
+            nodeid=1, attributes=[(1, Clusters.JointFabricDatastore.Attributes.AnchorVendorID)],
+            returnClusterObject=True)
+        asserts.assert_greater_equal(response[1][Clusters.JointFabricDatastore].anchorVendorID, 0)
+
+
+if __name__ == "__main__":
+    default_matter_test_main()