Add unknown clusters for derived

Because the derived clusters may not have the direction noted in
the table, we can't tell if they're accepted or generated commands
until we match them to the base. Keep an unknown list until we merge.
Also WARN on unknown clusters that persist after processing.

Author: cecille

src/python_testing/TestSpecParsingSupport.py

@@ -21,7 +21,8 @@
 from global_attribute_ids import GlobalAttributeIds
 from matter_testing_support import MatterBaseTest, default_matter_test_main
 from mobly import asserts
-from spec_parsing_support import ClusterParser, XmlCluster, add_cluster_data_from_xml, combine_derived_clusters_with_base
+from spec_parsing_support import (ClusterParser, XmlCluster, add_cluster_data_from_xml, check_clusters_for_unknown_commands,
+                                  combine_derived_clusters_with_base)
 
 # TODO: improve the test coverage here
 # https://github.com/project-chip/connectedhomeip/issues/30958
@@ -175,6 +176,21 @@ def get_access_enum_from_string(access_str: str) -> Clusters.AccessControl.Enums
     '</cluster>'
 )
 
+CLUSTER_WITH_UNKNOWN_COMMAND = (
+    '<cluster xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="types types.xsd cluster cluster.xsd" id="0xFFFE" name="Test Unknown Command" revision="1">'
+    '  <revisionHistory>'
+    '    <revision revision="1" summary="Initial version"/>'
+    '  </revisionHistory>'
+    '  <classification hierarchy="base" role="application" picsCode="BASE" scope="Endpoint"/>'
+    '  <commands>'
+    '    <command id="0x00" name="ChangeToMode" direction="commandToClient">'
+    '      <access invokePrivilege="operate"/>'
+    '      <mandatoryConform/>'
+    '    </command>'
+    '  </commands>'
+    '</cluster>'
+)
+
 
 class TestSpecParsingSupport(MatterBaseTest):
     def test_spec_parsing_access(self):
@@ -236,6 +252,14 @@ def test_derived_clusters(self):
         asserts.assert_equal(set(derived_clusters["Test Base"].accepted_commands.keys()), set([0]), "Unexpected accepted commands")
         asserts.assert_equal(set(derived_clusters["Test Base"].generated_commands.keys()),
                              set([1]), "Unexpected generated commands")
+        asserts.assert_equal(str(derived_clusters["Test Base"].accepted_commands[0].conformance),
+                             "M", "Unexpected conformance on base accepted command")
+        asserts.assert_equal(str(derived_clusters["Test Base"].generated_commands[1].conformance),
+                             "M", "Unexpected conformance on base generated command")
+
+        asserts.assert_equal(len(derived_clusters["Test Base"].unknown_commands),
+                             0, "Unexpected number of unknown commands in base")
+        asserts.assert_equal(len(clusters[0xFFFF].unknown_commands), 2, "Unexpected number of unknown commands in derived cluster")
 
         combine_derived_clusters_with_base(clusters, derived_clusters, ids_by_name)
         # Ensure the base-only attribute (1) was added to the derived cluster
@@ -247,8 +271,29 @@ def test_derived_clusters(self):
         asserts.assert_equal(str(clusters[0xFFFF].attributes[2].conformance), "X", "Unexpected conformance on attribute 1")
         asserts.assert_equal(str(clusters[0xFFFF].attributes[3].conformance), "X", "Unexpected conformance on attribute 1")
 
+        # Ensure both the accepted and generated command overrides work
+        asserts.assert_true(set(clusters[0xFFFF].accepted_commands.keys()),
+                            set([0]), "Unexpected accepted command list after merge")
+        asserts.assert_true(set(clusters[0xFFFF].generated_commands.keys()), set([1]),
+                            "Unexpected generated command list after merge")
+        asserts.assert_equal(str(clusters[0xFFFF].accepted_commands[0].conformance),
+                             "X", "Unexpected conformance on accepted commands")
+        asserts.assert_equal(str(clusters[0xFFFF].generated_commands[1].conformance),
+                             "X", "Unexpected conformance on generated commands")
+        asserts.assert_equal(len(clusters[0xFFFF].unknown_commands), 0, "Unexpected number of unknown commands after merge")
+
     def test_missing_command_direction(self):
-        pass
+        clusters: dict[int, XmlCluster] = {}
+        derived_clusters: dict[str, XmlCluster] = {}
+        ids_by_name: dict[str, int] = {}
+        problems: list[ProblemNotice] = []
+        cluster_xml = ElementTree.fromstring(CLUSTER_WITH_UNKNOWN_COMMAND)
+
+        add_cluster_data_from_xml(cluster_xml, clusters, derived_clusters, ids_by_name, problems)
+        check_clusters_for_unknown_commands(clusters, problems)
+        asserts.assert_equal(len(problems), 1, "Unexpected number of problems found")
+        asserts.assert_equal(problems[0].location.cluster_id, 0xFFFE, "Unexpected problem location (cluster id)")
+        asserts.assert_equal(problems[0].location.command_id, 0, "Unexpected problem location (command id)")
 
 
 if __name__ == "__main__":

src/python_testing/spec_parsing_support.py

@@ -76,6 +76,7 @@ def __str__(self):
 
 @dataclass
 class XmlCommand:
+    id: int
     name: str
     conformance: Callable[[uint], ConformanceDecision]
 
@@ -100,6 +101,7 @@ class XmlCluster:
     attributes: dict[uint, XmlAttribute]
     accepted_commands: dict[uint, XmlCommand]
     generated_commands: dict[uint, XmlCommand]
+    unknown_commands: list[XmlCommand]
     events: dict[uint, XmlEvent]
     pics: str
 
@@ -346,27 +348,37 @@ def parse_attributes(self) -> dict[uint, XmlAttribute]:
             ), read_access=Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kView, write_access=Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kUnknownEnumValue, write_optional=False)
         return attributes
 
-    def parse_commands(self, command_type: CommandType) -> dict[uint, XmlAttribute]:
-        commands = {}
+    def get_command_direction(self, element: ElementTree.Element) -> CommandType:
+        try:
+            if element.attrib['direction'].lower() == 'responsefromserver':
+                return CommandType.GENERATED
+            if element.attrib['direction'].lower() == 'commandtoclient':
+                return CommandType.UNKNOWN
+        except KeyError:
+            return CommandType.ACCEPTED
+
+    def parse_unknown_commands(self) -> list[XmlCommand]:
+        commands = []
         for element, conformance_xml, access_xml in self.command_elements:
+            if self.get_command_direction(element) != CommandType.UNKNOWN:
+                continue
             code = int(element.attrib['id'], 0)
-            dir = CommandType.ACCEPTED
-            try:
-                if element.attrib['direction'].lower() == 'responsefromserver':
-                    dir = CommandType.GENERATED
-                if element.attrib['direction'].lower() == 'commandtoclient':
-                    dir = CommandType.UNKNOWN
-            except KeyError:
-                pass
-            if dir != command_type:
+            conformance = self.parse_conformance(conformance_xml)
+            commands.append(XmlCommand(id=code, name=element.attrib['name'], conformance=conformance))
+        return commands
+
+    def parse_commands(self, command_type: CommandType) -> dict[uint, XmlCommand]:
+        commands = {}
+        for element, conformance_xml, access_xml in self.command_elements:
+            if self.get_command_direction(element) != command_type:
                 continue
             code = int(element.attrib['id'], 0)
             conformance = self.parse_conformance(conformance_xml)
             if conformance is None:
                 continue
             if code in commands:
                 conformance = or_operation([conformance, commands[code].conformance])
-            commands[code] = XmlCommand(name=element.attrib['name'], conformance=conformance)
+            commands[code] = XmlCommand(id=code, name=element.attrib['name'], conformance=conformance)
         return commands
 
     def parse_events(self) -> dict[uint, XmlAttribute]:
@@ -393,6 +405,7 @@ def create_cluster(self) -> XmlCluster:
                           attributes=self.parse_attributes(),
                           accepted_commands=self.parse_commands(CommandType.ACCEPTED),
                           generated_commands=self.parse_commands(CommandType.GENERATED),
+                          unknown_commands=self.parse_unknown_commands(),
                           events=self.parse_events(), pics=self._pics)
 
     def get_problems(self) -> list[ProblemNotice]:
@@ -423,6 +436,13 @@ def add_cluster_data_from_xml(xml: ElementTree.Element, clusters: dict[int, XmlC
             derived_clusters[name] = new
 
 
+def check_clusters_for_unknown_commands(clusters: dict[int, XmlCluster], problems: list[ProblemNotice]):
+    for id, cluster in clusters.items():
+        for cmd in cluster.unknown_commands:
+            problems.append(ProblemNotice(test_name="Spec XML parsing", location=CommandPathLocation(
+                endpoint_id=0, cluster_id=id, command_id=cmd.id), severity=ProblemSeverity.WARNING, problem="Command with unknown direction"))
+
+
 def build_xml_clusters() -> tuple[list[XmlCluster], list[ProblemNotice]]:
     dir = os.path.join(os.path.dirname(os.path.realpath(__file__)), '..', '..', 'data_model', 'clusters')
     clusters: dict[int, XmlCluster] = {}
@@ -489,6 +509,8 @@ def remove_problem(location: typing.Union[CommandPathLocation, FeaturePathLocati
     clusters[acl_id].attributes[Clusters.AccessControl.Attributes.Acl.attribute_id].write_access = Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kAdminister
     clusters[acl_id].attributes[Clusters.AccessControl.Attributes.Extension.attribute_id].write_access = Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kAdminister
 
+    check_clusters_for_unknown_commands(clusters, problems)
+
     return clusters, problems
 
 
@@ -522,7 +544,6 @@ def combine_attributes(base: dict[uint, XmlAttribute], derived: dict[uint, XmlAt
     for id, c in xml_clusters.items():
         if c.derived:
             base_name = c.derived
-            print(f"fixing cluster {base_name}")
             if base_name in ids_by_name:
                 base = xml_clusters[ids_by_name[c.derived]]
             else:
@@ -543,9 +564,17 @@ def combine_attributes(base: dict[uint, XmlAttribute], derived: dict[uint, XmlAt
             generated_commands.update(c.generated_commands)
             events = deepcopy(base.events)
             events.update(c.events)
+            unknown_commands = deepcopy(base.unknown_commands)
+            for cmd in c.unknown_commands:
+                if cmd.id in accepted_commands.keys() and cmd.name == accepted_commands[cmd.id].name:
+                    accepted_commands[cmd.id].conformance = cmd.conformance
+                elif cmd.id in generated_commands.keys() and cmd.name == generated_commands[cmd.id].name:
+                    generated_commands[cmd.id].conformance = cmd.conformance
+                else:
+                    unknown_commands.append(cmd)
+
             new = XmlCluster(revision=c.revision, derived=c.derived, name=c.name,
                              feature_map=feature_map, attribute_map=attribute_map, command_map=command_map,
                              features=features, attributes=attributes, accepted_commands=accepted_commands,
-                             generated_commands=generated_commands, events=events, pics=c.pics)
-            print(new)
+                             generated_commands=generated_commands, unknown_commands=unknown_commands, events=events, pics=c.pics)
             xml_clusters[id] = new