Auto-select data model set based on specification version

.github/workflows/tests.yaml

@@ -537,6 +537,7 @@ jobs:
                   scripts/run_in_python_env.sh out/venv 'python3 src/python_testing/TestIdChecks.py'
                   scripts/run_in_python_env.sh out/venv 'python3 src/python_testing/TestMatterTestingSupport.py'
                   scripts/run_in_python_env.sh out/venv 'python3 src/python_testing/TestSpecParsingDeviceType.py'
+                  scripts/run_in_python_env.sh out/venv 'python3 src/python_testing/TestSpecParsingSelection.py'
                   scripts/run_in_python_env.sh out/venv 'python3 src/python_testing/TestSpecParsingSupport.py'
 
             - name: Run Tests

src/python_testing/TC_AccessChecker.py

@@ -28,7 +28,7 @@
 from chip.testing.global_attribute_ids import GlobalAttributeIds
 from chip.testing.matter_testing import (AttributePathLocation, ClusterPathLocation, MatterBaseTest, TestStep, async_test_body,
                                          default_matter_test_main)
-from chip.testing.spec_parsing import XmlCluster, build_xml_clusters
+from chip.testing.spec_parsing import XmlCluster
 from chip.tlv import uint
 
 
@@ -72,7 +72,8 @@ async def setup_class(self):
         self.user_params["use_pase_only"] = False
         super().setup_class()
         await self.setup_class_helper()
-        self.xml_clusters, self.problems = build_xml_clusters()
+        self.build_spec_xmls()
+
         acl_attr = Clusters.AccessControl.Attributes.Acl
         self.default_acl = await self.read_single_attribute_check_success(cluster=Clusters.AccessControl, attribute=acl_attr)
         self._record_errors()

src/python_testing/TC_DeviceConformance.py

@@ -47,16 +47,14 @@
                                                device_type_id_type, is_valid_device_type_id)
 from chip.testing.matter_testing import (AttributePathLocation, ClusterPathLocation, CommandPathLocation, DeviceTypePathLocation,
                                          MatterBaseTest, ProblemNotice, ProblemSeverity, async_test_body, default_matter_test_main)
-from chip.testing.spec_parsing import CommandType, build_xml_clusters, build_xml_device_types
+from chip.testing.spec_parsing import CommandType
 from chip.tlv import uint
 
 
 class DeviceConformanceTests(BasicCompositionTests):
     async def setup_class_helper(self):
         await super().setup_class_helper()
-        self.xml_clusters, self.problems = build_xml_clusters()
-        self.xml_device_types, problems = build_xml_device_types()
-        self.problems.extend(problems)
+        self.build_spec_xmls()
 
     def _get_device_type_id(self, device_type_name: str) -> int:
         id = [id for id, dt in self.xml_device_types.items() if dt.name.lower() == device_type_name.lower()]

src/python_testing/TC_pics_checker.py

@@ -20,9 +20,9 @@
 from chip.testing.basic_composition import BasicCompositionTests
 from chip.testing.global_attribute_ids import GlobalAttributeIds
 from chip.testing.matter_testing import (AttributePathLocation, ClusterPathLocation, CommandPathLocation, FeaturePathLocation,
-                                         MatterBaseTest, ProblemLocation, TestStep, async_test_body, default_matter_test_main)
+                                         MatterBaseTest, TestStep, UnknownProblemLocation, async_test_body,
+                                         default_matter_test_main)
 from chip.testing.pics import accepted_cmd_pics_str, attribute_pics_str, feature_pics_str, generated_cmd_pics_str
-from chip.testing.spec_parsing import build_xml_clusters
 from mobly import asserts
 
 
@@ -31,10 +31,7 @@ class TC_PICS_Checker(MatterBaseTest, BasicCompositionTests):
     async def setup_class(self):
         super().setup_class()
         await self.setup_class_helper(False)
-        # build_xml_cluster returns a list of issues found when paring the XML
-        # Problems in the XML shouldn't cause test failure, but we want them recorded
-        # so they are added to the list of problems that get output when the test set completes.
-        self.xml_clusters, self.problems = build_xml_clusters()
+        self.build_spec_xmls()
 
     def _check_and_record_errors(self, location, required, pics):
         if required and not self.check_pics(pics):
@@ -178,7 +175,7 @@ def test_TC_IDM_10_4(self):
 
         self.step(7)
         if self.is_pics_sdk_ci_only:
-            self.record_error("PICS check", location=ProblemLocation(),
+            self.record_error("PICS check", location=UnknownProblemLocation(),
                               problem="PICS PICS_SDK_CI_ONLY found in PICS list. This PICS is disallowed for certification.")
             self.success = False
 

src/python_testing/TestConformanceTest.py

@@ -22,7 +22,7 @@
 from chip.testing.conformance import ConformanceDecision
 from chip.testing.global_attribute_ids import GlobalAttributeIds
 from chip.testing.matter_testing import MatterBaseTest, async_test_body, default_matter_test_main
-from chip.testing.spec_parsing import build_xml_clusters, build_xml_device_types
+from chip.testing.spec_parsing import PrebuiltDataModelDirectory, build_xml_clusters, build_xml_device_types
 from mobly import asserts
 from TC_DeviceConformance import DeviceConformanceTests
 
@@ -118,8 +118,10 @@ def is_mandatory(conformance):
 
 class TestConformanceSupport(MatterBaseTest, DeviceConformanceTests):
     def setup_class(self):
-        self.xml_clusters, self.problems = build_xml_clusters()
-        self.xml_device_types, problems = build_xml_device_types()
+        # Latest fully qualified version
+        # TODO: It might be good to find a way to run this against each directory.
+        self.xml_clusters, self.problems = build_xml_clusters(PrebuiltDataModelDirectory.k1_4)
+        self.xml_device_types, problems = build_xml_device_types(PrebuiltDataModelDirectory.k1_4)
         self.problems.extend(problems)
 
     @async_test_body

src/python_testing/TestSpecParsingDeviceType.py

@@ -35,8 +35,9 @@ def test_spec_device_parsing(self):
             print(str(d))
 
     def setup_class(self):
-        self.xml_clusters, self.xml_cluster_problems = build_xml_clusters()
-        self.xml_device_types, self.xml_device_types_problems = build_xml_device_types()
+        # Latest fully qualified release
+        self.xml_clusters, self.xml_cluster_problems = build_xml_clusters(PrebuiltDataModelDirectory.k1_4)
+        self.xml_device_types, self.xml_device_types_problems = build_xml_device_types(PrebuiltDataModelDirectory.k1_4)
 
         self.device_type_id = 0xBBEF
         self.revision = 2

src/python_testing/TestSpecParsingSelection.py

@@ -0,0 +1,149 @@
+#
+#    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 chip.clusters as Clusters
+from chip.testing.conformance import ConformanceDecision, ConformanceException
+from chip.testing.global_attribute_ids import is_standard_attribute_id
+from chip.testing.matter_testing import MatterBaseTest, default_matter_test_main
+from chip.testing.spec_parsing import PrebuiltDataModelDirectory, build_xml_clusters, dm_from_spec_version
+from chip.tlv import uint
+from mobly import asserts, signals
+from TC_DeviceConformance import DeviceConformanceTests
+
+
+class TestSpecParsingSelection(MatterBaseTest, DeviceConformanceTests):
+    def setup_class(self):
+        # Overriding the DeviceConformanceTest setup_class so we don't go out to a real device
+        pass
+
+    def test_dm_from_spec_version(self):
+        asserts.assert_equal(dm_from_spec_version(0x01030000), PrebuiltDataModelDirectory.k1_3,
+                             "Incorrect directory selected for 1.3 with patch 0")
+        asserts.assert_equal(dm_from_spec_version(0x01030100), PrebuiltDataModelDirectory.k1_3,
+                             "Incorrect directory selected for 1.3 with patch 1")
+        asserts.assert_equal(dm_from_spec_version(0x01040100), PrebuiltDataModelDirectory.k1_4_1,
+                             "Incorrect directory selected for 1.4.1")
+        asserts.assert_equal(dm_from_spec_version(0x01040100), PrebuiltDataModelDirectory.k1_4_1,
+                             "Incorrect directory selected for 1.4.1")
+        asserts.assert_equal(dm_from_spec_version(0x01050000), PrebuiltDataModelDirectory.kMaster,
+                             "Incorrect directory selected for 1.5")
+
+        # We don't have data model files for 1.2, so these should error
+        with asserts.assert_raises(ConformanceException, "Expected assertion was not raised for spec version 1.2"):
+            dm_from_spec_version(0x01020000)
+
+        # Any dot release besides 0 and 1 for 1.4 should error
+        with asserts.assert_raises(ConformanceException, "Data model incorrectly identified for 1.4.2"):
+            dm_from_spec_version(0x01040200)
+
+        with asserts.assert_raises(ConformanceException, "Data model incorrectly identified for 1.4.FF"):
+            dm_from_spec_version(0x0104FF00)
+
+        # Any dot release besides 0 for 1.5 should error
+        with asserts.assert_raises(ConformanceException, "Data model incorrectly identified for 1.5.1"):
+            dm_from_spec_version(0x01050100)
+        with asserts.assert_raises(ConformanceException, "Data model incorrectly identified for 1.5.FF"):
+            dm_from_spec_version(0x0105FF00)
+
+        # Any value with stuff in reserved should error
+        with asserts.assert_raises(ConformanceException, "Error not returned for specification revision with non-zero reserved values"):
+            dm_from_spec_version(0x01030001)
+        with asserts.assert_raises(ConformanceException, "Error not returned for specification revision with non-zero reserved values"):
+            dm_from_spec_version(0x01040001)
+        with asserts.assert_raises(ConformanceException, "Error not returned for specification revision with non-zero reserved values"):
+            dm_from_spec_version(0x01040101)
+        with asserts.assert_raises(ConformanceException, "Error not returned for specification revision with non-zero reserved values"):
+            dm_from_spec_version(0x01050001)
+
+    def _create_device(self, spec_version: uint, tc_enabled: bool):
+        # Build at 1.4.1 so we can have TC info
+        xml_clusters, _ = build_xml_clusters(PrebuiltDataModelDirectory.k1_4_1)
+
+        gc_feature_map = Clusters.GeneralCommissioning.Bitmaps.Feature.kTermsAndConditions if tc_enabled else 0
+
+        def create_cluster_globals(cluster, feature_map):
+            spec_attributes = xml_clusters[cluster.id].attributes
+            spec_accepted_commands = xml_clusters[cluster.id].accepted_commands
+            spec_generated_commands = xml_clusters[cluster.id].generated_commands
+            # Build just the lists - basic composition checks the wildcard against the lists, conformance just uses lists
+            attributes = [id for id, a in spec_attributes.items() if a.conformance(
+                feature_map, [], []).decision == ConformanceDecision.MANDATORY]
+            accepted_commands = [id for id, c in spec_accepted_commands.items() if c.conformance(
+                feature_map, [], []).decision == ConformanceDecision.MANDATORY]
+            generated_commands = [id for id, c in spec_generated_commands.items() if c.conformance(
+                feature_map, [], []).decision == ConformanceDecision.MANDATORY]
+            attr = cluster.Attributes
+
+            resp = {}
+            non_global_attrs = [a for a in attributes if is_standard_attribute_id(a)]
+            for attribute_id in non_global_attrs:
+                # We don't use the values in these tests, set them all to 0. The types are wrong, but it shouldn't matter
+                resp[Clusters.ClusterObjects.ALL_ATTRIBUTES[cluster.id][attribute_id]] = 0
+
+            resp[attr.AttributeList] = attributes
+            resp[attr.AcceptedCommandList] = accepted_commands
+            resp[attr.GeneratedCommandList] = generated_commands
+            resp[attr.FeatureMap] = feature_map
+            resp[attr.ClusterRevision] = xml_clusters[cluster.id].revision
+
+            return resp
+
+        def get_tlv(resp):
+            # This only works because there are no structs in here.
+            # structs need special handling. Beware.
+            return {k.attribute_id: v for k, v in resp.items()}
+
+        gc_resp = create_cluster_globals(Clusters.GeneralCommissioning, gc_feature_map)
+        bi_resp = create_cluster_globals(Clusters.BasicInformation, 0)
+        bi_resp[Clusters.BasicInformation.Attributes.SpecificationVersion] = spec_version
+
+        self.endpoints = {0: {Clusters.GeneralCommissioning: gc_resp, Clusters.BasicInformation: bi_resp}}
+        self.endpoints_tlv = {0: {Clusters.GeneralCommissioning.id: get_tlv(
+            gc_resp), Clusters.BasicInformation.id: get_tlv(bi_resp)}}
+
+    def _run_conformance_against_device(self, spec_version: uint, tc_enabled: bool, expect_success_conformance: bool, expect_success_revisions: bool):
+        self._create_device(spec_version, tc_enabled)
+        # build the spec XMLs for the stated version
+        self.build_spec_xmls()
+        success, problems = self.check_conformance(ignore_in_progress=False, is_ci=False, allow_provisional=False)
+        problem_strs = [str(p) for p in problems]
+        problem_str = "\n".join(problem_strs)
+        asserts.assert_equal(success, expect_success_conformance,
+                             f"Improper conformance result for spec version {spec_version:08X}, TC: {tc_enabled} problems: {problem_str}")
+
+        success, problems = self.check_revisions(ignore_in_progress=False)
+        asserts.assert_equal(success, expect_success_revisions,
+                             f"Improper revision result for spec version {spec_version:08X}, TC: {tc_enabled} problems: {problems}")
+
+    def test_conformance(self):
+
+        # 1.4 is OK if TC is off
+        self._run_conformance_against_device(0x01040000, False, expect_success_conformance=True, expect_success_revisions=True)
+        # 1.4.1 is OK if TC is off
+        self._run_conformance_against_device(0x01040100, False, expect_success_conformance=True, expect_success_revisions=True)
+        # 1.4.1 is OK if TC is on
+        self._run_conformance_against_device(0x01040100, True, expect_success_conformance=True, expect_success_revisions=True)
+        # 1.4 is NOT OK if TC is on
+        self._run_conformance_against_device(0x01040000, True, expect_success_conformance=False, expect_success_revisions=True)
+
+        # Check that we get a test failure on a bad spec revision
+        self._create_device(0xFFFFFFFF, False)
+        with asserts.assert_raises(signals.TestFailure, "Exception not properly raised for bad spec type"):
+            self.build_spec_xmls()
+
+
+if __name__ == "__main__":
+    default_matter_test_main()

src/python_testing/TestSpecParsingSupport.py

@@ -253,7 +253,8 @@ def get_access_enum_from_string(access_str: str) -> Clusters.AccessControl.Enums
 class TestSpecParsingSupport(MatterBaseTest):
     def setup_class(self):
         super().setup_class()
-        self.spec_xml_clusters, self.spec_problems = build_xml_clusters()
+        # Latest fully certified build
+        self.spec_xml_clusters, self.spec_problems = build_xml_clusters(PrebuiltDataModelDirectory.k1_4)
         self.all_spec_clusters = set([(id, c.name, c.pics) for id, c in self.spec_xml_clusters.items()])
 
     def test_build_xml_override(self):

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

@@ -31,6 +31,8 @@
 import chip.clusters.ClusterObjects
 import chip.tlv
 from chip.clusters.Attribute import ValueDecodeFailure
+from chip.testing.conformance import ConformanceException
+from chip.testing.spec_parsing import PrebuiltDataModelDirectory, build_xml_clusters, build_xml_device_types, dm_from_spec_version
 from mobly import asserts
 
 
@@ -210,3 +212,23 @@ def fail_current_test(self, msg: Optional[str] = None):
             asserts.fail(msg=self.problems[-1].problem)
         else:
             asserts.fail(msg)
+
+    def _get_dm(self) -> PrebuiltDataModelDirectory:
+        try:
+            spec_version = self.endpoints[0][Clusters.BasicInformation][Clusters.BasicInformation.Attributes.SpecificationVersion]
+        except KeyError:
+            asserts.fail(
+                "Specification Version not found on device - ensure device bas a basic information cluster on EP0 supporting Specification Version")
+        try:
+            return dm_from_spec_version(spec_version)
+        except ConformanceException as e:
+            asserts.fail(f"Unable to identify specification version: {e}")
+
+    def build_spec_xmls(self):
+        dm = self._get_dm()
+        logging.info("----------------------------------------------------------------------------------")
+        logging.info(f"-- Running tests against Specification version {dm.dirname}")
+        logging.info("----------------------------------------------------------------------------------")
+        self.xml_clusters, self.problems = build_xml_clusters(dm)
+        self.xml_device_types, problems = build_xml_device_types(dm)
+        self.problems.extend(problems)

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

@@ -803,7 +803,12 @@ def __str__(self):
         return msg
 
 
-ProblemLocation = typing.Union[ClusterPathLocation, DeviceTypePathLocation]
+class UnknownProblemLocation:
+    def __str__(self):
+        return f'\n      Unknown Locations - see message for more details'
+
+
+ProblemLocation = typing.Union[ClusterPathLocation, DeviceTypePathLocation, UnknownProblemLocation]
 
 # ProblemSeverity is not using StrEnum, but rather Enum, since StrEnum only
 # appeared in 3.11. To make it JSON serializable easily, multiple inheritance

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

@@ -75,6 +75,7 @@ def run_test_with_mock_read(self, read_cache: Attribute.AsyncReadTransaction.Rea
         self.default_controller.Read = AsyncMock(return_value=read_cache)
         # This doesn't need to do anything since we are overriding the read anyway
         self.default_controller.FindOrEstablishPASESession = AsyncMock(return_value=None)
+        self.default_controller.GetConnectedDevice = AsyncMock(return_value=None)
         with asyncio.Runner() as runner:
             return run_tests_no_exit(self.test_class, self.config, runner.get_loop(),
                                      hooks, self.default_controller, self.stack)