Device composition tests: Add option to run tests from file

This PR allows us to injest data dumps from previous runs, and run
tests against them. This works for device basic composition tests,
device conformance tests, and pics tests.

Author: cecille

src/controller/python/BUILD.gn

@@ -67,6 +67,7 @@ shared_library("ChipDeviceCtrl") {
       "ChipDeviceController-ScriptPairingDeviceDiscoveryDelegate.h",
       "ChipDeviceController-StorageDelegate.cpp",
       "ChipDeviceController-StorageDelegate.h",
+      "JsonTlvSupport.cpp",
       "OpCredsBinding.cpp",
       "chip/clusters/attribute.cpp",
       "chip/clusters/command.cpp",
@@ -127,6 +128,7 @@ shared_library("ChipDeviceCtrl") {
     "${chip_root}/src/lib/core",
     "${chip_root}/src/lib/dnssd",
     "${chip_root}/src/lib/support",
+    "${chip_root}/src/lib/support/jsontlv",
     "${chip_root}/src/platform",
     "${chip_root}/src/setup_payload",
     "${chip_root}/src/transport",

src/controller/python/chip/ChipDeviceCtrl.py

@@ -38,7 +38,7 @@
 import threading
 import time
 import typing
-from ctypes import (CDLL, CFUNCTYPE, POINTER, byref, c_bool, c_char, c_char_p, c_int, c_int32, c_size_t, c_uint8, c_uint16,
+from ctypes import (CDLL, CFUNCTYPE, POINTER, byref, c_bool, c_char, c_char_p, c_int, c_int32, c_size_t, c_ubyte, c_uint8, c_uint16,
                     c_uint32, c_uint64, c_void_p, create_string_buffer, pointer, py_object, resize, string_at)
 from dataclasses import dataclass
 
@@ -51,12 +51,14 @@
 from .clusters import ClusterObjects as ClusterObjects
 from .clusters import Command as ClusterCommand
 from .clusters import Objects as GeneratedObjects
+from .clusters.Attribute import AttributeCache, AttributePath
 from .clusters.CHIPClusters import ChipClusters
 from .crypto import p256keypair
 from .exceptions import UnknownAttribute, UnknownCommand
 from .interaction_model import InteractionModelError, SessionParameters, SessionParametersStruct
 from .interaction_model import delegate as im
 from .native import PyChipError
+from .tlv import TLVReader
 
 __all__ = ["ChipDeviceController", "CommissioningParameters"]
 
@@ -2025,3 +2027,58 @@ def __init__(self, operationalKey: p256keypair.P256Keypair, noc: bytes,
         self._set_dev_ctrl(devCtrl)
 
         self._finish_init()
+
+
+class TLVJsonConverter():
+    ''' Converter class used to convert dumped Matter JsonTlv files into a cache. '''
+
+    def __init__(self, name: str = ''):
+        self._ChipStack = builtins.chipStack
+        self._dmLib = None
+
+        self._InitLib()
+
+    def _InitLib(self):
+        if self._dmLib is None:
+            self._dmLib = CDLL(self._ChipStack.LocateChipDLL())
+
+            self._dmLib.pychip_JsonToTlv.argtypes = [c_char_p, POINTER(c_ubyte), c_size_t]
+            self._dmLib.pychip_DeviceController_DeleteDeviceController.restype = c_size_t
+
+    # PER ATTRIBUTE
+    def _attribute_to_tlv(self, json_string: str) -> bytearray:
+        ''' Converts the MatterJsonTlv for one attribute into TLV that can be parsed and put into the cache.'''
+        # We don't currently have a way to size this properly, but we know attributes need to fit into 1 MTU.
+        size = 1280
+        buf = bytearray(size)
+        encoded_bytes = self._dmLib.pychip_JsonToTlv(json_string.encode("utf-8"),  (ctypes.c_ubyte * size).from_buffer(buf), size)
+        return buf[:encoded_bytes]
+
+    def convert_dump_to_cache(self, json_tlv: str) -> AttributeCache:
+        ''' Converts a string containing the MatterJsonTlv dump of an entire device into an AttributeCache object.
+            Input:
+              json_tlv: json string read from the dump file.
+            Returns:
+              AttributeCache with the data from the json_string
+        '''
+        cache = AttributeCache()
+        for endpoint_id_str, endpoint in json_tlv.items():
+            endpoint_id = int(endpoint_id_str, 0)
+            for cluster_id_str, cluster in endpoint.items():
+                s = cluster_id_str.split(':')
+                cluster_id = int(s[0])
+                for attribute_id_str, attribute in cluster.items():
+                    s = attribute_id_str.split(':')
+                    attribute_id = int(s[0])
+                    json_str = json.dumps({attribute_id_str: attribute}, indent=2)
+                    tmp = self._attribute_to_tlv(json_str)
+                    path = AttributePath(EndpointId=endpoint_id, ClusterId=cluster_id, AttributeId=attribute_id)
+                    # Each of these attributes contains only one item
+                    try:
+                        tlvData = next(iter(TLVReader(tmp).get().get("Any", {}).values()))
+                    except StopIteration:
+                        # no data, this is a value decode error
+                        tlvData = ValueDecodeFailure()
+                    cache.UpdateTLV(path=path, dataVersion=0, data=tlvData)
+                    cache.UpdateCachedData(set([path]))
+        return cache

src/python_testing/TC_DeviceBasicComposition.py

@@ -170,6 +170,10 @@ def test_TC_DT_1_1(self):
             self.fail_current_test("At least one endpoint was missing the descriptor cluster.")
 
     async def _read_non_standard_attribute_check_unsupported_read(self, endpoint_id, cluster_id, attribute_id) -> bool:
+        # If we're doing this from file, we don't have a way to assess this. Assume this is OK for now.
+        if self.test_from_file:
+            return True
+
         @dataclass
         class TempAttribute(ClusterAttributeDescriptor):
             @ChipUtility.classproperty

src/python_testing/basic_composition_support.py

@@ -27,8 +27,9 @@
 
 import chip.clusters.ClusterObjects
 import chip.tlv
-from chip.clusters.Attribute import ValueDecodeFailure
+from chip.clusters.Attribute import ValueDecodeFailure, AttributePath, AttributeCache
 from mobly import asserts
+from chip.ChipDeviceCtrl import TLVJsonConverter
 
 
 def MatterTlvToJson(tlv_data: dict[int, Any]) -> dict[str, Any]:
@@ -96,14 +97,33 @@ def ConvertValue(value) -> Any:
     return matter_json_dict
 
 
+def JsonToMatterTlv(json_filename: str) -> dict[int, Any]:
+    converter = TLVJsonConverter()
+    with open(json_filename, "r") as fin:
+        json_tlv = json.load(fin)
+        return converter.convert_dump_to_cache(json_tlv)
+
+
 class BasicCompositionTests:
     async def setup_class_helper(self, default_to_pase: bool = True):
         dev_ctrl = self.default_controller
         self.problems = []
-
+        self.test_from_file = self.user_params.get("test_from_file", None)
         do_test_over_pase = self.user_params.get("use_pase_only", default_to_pase)
         dump_device_composition_path: Optional[str] = self.user_params.get("dump_device_composition_path", None)
 
+        def log_test_start():
+            logging.info("###########################################################")
+            logging.info("Start of actual tests")
+            logging.info("###########################################################")
+
+        if self.test_from_file:
+            cache = JsonToMatterTlv(self.test_from_file)
+            self.endpoints = cache.attributeCache
+            self.endpoints_tlv = cache.attributeTLVCache
+            log_test_start()
+            return
+
         if do_test_over_pase:
             setupCode = self.matter_test_config.qr_code_content if self.matter_test_config.qr_code_content is not None else self.matter_test_config.manual_code
             asserts.assert_true(setupCode, "Require either --qr-code or --manual-code.")
@@ -125,9 +145,7 @@ async def setup_class_helper(self, default_to_pase: bool = True):
             with open(pathlib.Path(dump_device_composition_path).with_suffix(".txt"), "wt+") as outfile:
                 pprint(wildcard_read.attributes, outfile, indent=1, width=200, compact=True)
 
-        logging.info("###########################################################")
-        logging.info("Start of actual tests")
-        logging.info("###########################################################")
+        log_test_start()
 
         # ======= State kept for use by all tests =======