Support 'reject server cluster' in matterlint rules

scripts/py_matter_idl/matter_idl/lint/lint_rules_grammar.lark

@@ -6,12 +6,14 @@ load_xml: "load" ESCAPED_STRING ";"
 
 all_endpoint_rule: "all" "endpoints" "{" required_global_attribute* "}"
 
-specific_endpoint_rule: "endpoint" integer "{" required_server_cluster* "}"
+specific_endpoint_rule: "endpoint" integer "{" (required_server_cluster|rejected_server_cluster)* "}"
 
 required_global_attribute: "require" "global" "attribute" id "=" integer ";"
 
 required_server_cluster: "require" "server" "cluster" (id|POSITIVE_INTEGER|HEX_INTEGER) ";"
 
+rejected_server_cluster: "reject" "server" "cluster" (id|POSITIVE_INTEGER|HEX_INTEGER) ";"
+
 integer: positive_integer | negative_integer
 
 positive_integer: POSITIVE_INTEGER | HEX_INTEGER

scripts/py_matter_idl/matter_idl/lint/lint_rules_parser.py

@@ -4,21 +4,22 @@
 import os
 import xml.etree.ElementTree
 from dataclasses import dataclass
-from typing import List, MutableMapping
+from enum import Enum, auto
+from typing import List, MutableMapping, Tuple, Union
 
 from lark import Lark
 from lark.visitors import Discard, Transformer, v_args
 
 try:
-    from .types import (AttributeRequirement, ClusterCommandRequirement, ClusterRequirement, RequiredAttributesRule,
-                        RequiredCommandsRule)
+    from .types import (AttributeRequirement, ClusterCommandRequirement, ClusterRequirement, ClusterValidationRule,
+                        RequiredAttributesRule, RequiredCommandsRule)
 except ImportError:
     import sys
 
     sys.path.append(os.path.join(os.path.abspath(
         os.path.dirname(__file__)), "..", ".."))
-    from matter_idl.lint.types import (AttributeRequirement, ClusterCommandRequirement, ClusterRequirement, RequiredAttributesRule,
-                                       RequiredCommandsRule)
+    from matter_idl.lint.types import (AttributeRequirement, ClusterCommandRequirement, ClusterRequirement, ClusterValidationRule,
+                                       RequiredAttributesRule, RequiredCommandsRule)
 
 
 class ElementNotFoundError(Exception):
@@ -53,6 +54,17 @@ class DecodedCluster:
     required_commands: List[RequiredCommand]
 
 
+class ClusterActionEnum(Enum):
+    REQUIRE = auto()
+    REJECT = auto()
+
+
+@dataclass
+class ServerClusterRequirement:
+    action: ClusterActionEnum
+    id: Union[str, int]
+
+
 def DecodeClusterFromXml(element: xml.etree.ElementTree.Element):
     if element.tag != 'cluster':
         logging.error("Not a cluster element: %r" % element)
@@ -141,35 +153,58 @@ class LintRulesContext:
     def __init__(self):
         self._required_attributes_rule = RequiredAttributesRule(
             "Required attributes")
+        self._cluster_validation_rule = ClusterValidationRule(
+            "Cluster validation")
         self._required_commands_rule = RequiredCommandsRule(
             "Required commands")
 
         # Map cluster names to the underlying code
         self._cluster_codes: MutableMapping[str, int] = {}
 
     def GetLinterRules(self):
-        return [self._required_attributes_rule, self._required_commands_rule]
+        return [self._required_attributes_rule, self._required_commands_rule, self._cluster_validation_rule]
 
     def RequireAttribute(self, r: AttributeRequirement):
         self._required_attributes_rule.RequireAttribute(r)
 
-    def RequireClusterInEndpoint(self, name: str, code: int):
-        """Mark that a specific cluster is always required in the given endpoint
-        """
+    def FindClusterCode(self, name: str) -> Tuple[str, int]:
         if name not in self._cluster_codes:
             # Name may be a number. If this can be parsed as a number, accept it anyway
             try:
-                cluster_code = parseNumberString(name)
-                name = "ID_%s" % name
+                return "ID_%s" % name, parseNumberString(name)
             except ValueError:
                 logging.error("UNKNOWN cluster name %s" % name)
                 logging.error("Known names: %s" %
                               (",".join(self._cluster_codes.keys()), ))
-                return
+                return None
         else:
-            cluster_code = self._cluster_codes[name]
+            return name, self._cluster_codes[name]
+
+    def RequireClusterInEndpoint(self, name: str, code: int):
+        """Mark that a specific cluster is always required in the given endpoint
+        """
+        cluster_info = self.FindClusterCode(name)
+        if not cluster_info:
+            return
 
-        self._required_attributes_rule.RequireClusterInEndpoint(ClusterRequirement(
+        name, cluster_code = cluster_info
+
+        self._cluster_validation_rule.RequireClusterInEndpoint(ClusterRequirement(
+            endpoint_id=code,
+            cluster_code=cluster_code,
+            cluster_name=name,
+        ))
+
+    def RejectClusterInEndpoint(self, name: str, code: int):
+        """Mark that a specific cluster is always rejected in the given endpoint
+        """
+        cluster_info = self.FindClusterCode(name)
+        if not cluster_info:
+            return
+
+        name, cluster_code = cluster_info
+
+        self._cluster_validation_rule.RejectClusterInEndpoint(ClusterRequirement(
             endpoint_id=code,
             cluster_code=cluster_code,
             cluster_name=name,
@@ -265,14 +300,25 @@ def required_global_attribute(self, name, code):
         return AttributeRequirement(code=code, name=name)
 
     @v_args(inline=True)
-    def specific_endpoint_rule(self, code, *names):
-        for name in names:
-            self.context.RequireClusterInEndpoint(name, code)
+    def specific_endpoint_rule(self, code, *requirements):
+        for requirement in requirements:
+            if requirement.action == ClusterActionEnum.REQUIRE:
+                self.context.RequireClusterInEndpoint(requirement.id, code)
+            elif requirement.action == ClusterActionEnum.REJECT:
+                self.context.RejectClusterInEndpoint(requirement.id, code)
+            else:
+                raise Exception("Unexpected requirement action %r" %
+                                requirement.action)
+
         return Discard
 
     @v_args(inline=True)
     def required_server_cluster(self, id):
-        return id
+        return ServerClusterRequirement(ClusterActionEnum.REQUIRE, id)
+
+    @v_args(inline=True)
+    def rejected_server_cluster(self, id):
+        return ServerClusterRequirement(ClusterActionEnum.REJECT, id)
 
 
 class Parser:

scripts/py_matter_idl/matter_idl/lint/types.py

@@ -115,12 +115,93 @@ def _LintImpl(self):
         pass
 
 
+class ClusterValidationRule(ErrorAccumulatingRule):
+    def __init__(self, name):
+        super().__init__(name)
+        self._mandatory_clusters: List[ClusterRequirement] = []
+        self._rejected_clusters: List[ClusterRequirement] = []
+
+    def __repr__(self):
+        result = "ClusterValidationRule{\n"
+
+        if self._mandatory_clusters:
+            result += "  mandatory_clusters:\n"
+            for cluster in self._mandatory_clusters:
+                result += "    - %r\n" % cluster
+
+        if self._rejected_clusters:
+            result += "   rejected_clusters:\n"
+            for cluster in self._rejected_clusters:
+                result += "    - %r\n" % cluster
+
+        result += "}"
+
+        return result
+
+    def RequireClusterInEndpoint(self, requirement: ClusterRequirement):
+        self._mandatory_clusters.append(requirement)
+
+    def RejectClusterInEndpoint(self, requirement: ClusterRequirement):
+        self._rejected_clusters.append(requirement)
+
+    def _ClusterCode(self, name: str, location: Optional[LocationInFile]):
+        """Finds the server cluster definition with the given name.
+
+        On error returns None and _lint_errors is updated internlly
+        """
+        if not self._idl:
+            raise MissingIdlError()
+
+        cluster_definition = [
+            c for c in self._idl.clusters if c.name == name and c.side == ClusterSide.SERVER
+        ]
+        if not cluster_definition:
+            self._AddLintError(
+                "Cluster definition for %s not found" % name, location)
+            return None
+
+        if len(cluster_definition) > 1:
+            self._AddLintError(
+                "Multiple cluster definitions found for %s" % name, location)
+            return None
+
+        return cluster_definition[0].code
+
+    def _LintImpl(self):
+        if not self._idl:
+            raise MissingIdlError()
+
+        for endpoint in self._idl.endpoints:
+            cluster_codes = set()
+            for cluster in endpoint.server_clusters:
+                cluster_code = self._ClusterCode(
+                    cluster.name, self._ParseLocation(cluster.parse_meta))
+                if not cluster_code:
+                    continue
+
+                cluster_codes.add(cluster_code)
+
+            for requirement in self._mandatory_clusters:
+                if requirement.endpoint_id != endpoint.number:
+                    continue
+
+                if requirement.cluster_code not in cluster_codes:
+                    self._AddLintError("Endpoint %d DOES NOT expose cluster %s (%d)" %
+                                       (requirement.endpoint_id, requirement.cluster_name, requirement.cluster_code), location=None)
+
+            for requirement in self._rejected_clusters:
+                if requirement.endpoint_id != endpoint.number:
+                    continue
+
+                if requirement.cluster_code in cluster_codes:
+                    self._AddLintError("Endpoint %d EXPOSES cluster %s (%d)" %
+                                       (requirement.endpoint_id, requirement.cluster_name, requirement.cluster_code), location=None)
+
+
 class RequiredAttributesRule(ErrorAccumulatingRule):
     def __init__(self, name):
-        super(RequiredAttributesRule, self).__init__(name)
-        # Map attribute code to name
+        super().__init__(name)
         self._mandatory_attributes: List[AttributeRequirement] = []
-        self._mandatory_clusters: List[ClusterRequirement] = []
 
     def __repr__(self):
         result = "RequiredAttributesRule{\n"
@@ -130,21 +211,13 @@ def __repr__(self):
             for attr in self._mandatory_attributes:
                 result += "    - %r\n" % attr
 
-        if self._mandatory_clusters:
-            result += "  mandatory_clusters:\n"
-            for cluster in self._mandatory_clusters:
-                result += "    - %r\n" % cluster
-
         result += "}"
         return result
 
     def RequireAttribute(self, attr: AttributeRequirement):
         """Mark an attribute required"""
         self._mandatory_attributes.append(attr)
 
-    def RequireClusterInEndpoint(self, requirement: ClusterRequirement):
-        self._mandatory_clusters.append(requirement)
-
     def _ServerClusterDefinition(self, name: str, location: Optional[LocationInFile]):
         """Finds the server cluster definition with the given name.
 
@@ -213,14 +286,6 @@ def _LintImpl(self):
                                             check.name, check.code),
                                            self._ParseLocation(cluster.parse_meta))
 
-            for requirement in self._mandatory_clusters:
-                if requirement.endpoint_id != endpoint.number:
-                    continue
-
-                if requirement.cluster_code not in cluster_codes:
-                    self._AddLintError("Endpoint %d does not expose cluster %s (%d)" %
-                                       (requirement.endpoint_id, requirement.cluster_name, requirement.cluster_code), location=None)
-
 
 @dataclass
 class ClusterCommandRequirement:

scripts/rules.matterlint

@@ -112,6 +112,11 @@ endpoint 0 {
   require server cluster OperationalCredentials;
   require server cluster GeneralDiagnostics;
 
+  // Example rejection of clusters:
+  //
+  // reject server cluster Scenes;
+  // reject server cluster Groups;
+
   // Required only if !CustomNetworkConfig. 
   // require server cluster NetworkCommissioning;