Load registry schemas if type match (#3450)

* Load registry schemas if type match

Author: kddejong

src/cfnlint/rules/resources/PrimaryIdentifiers.py

@@ -145,6 +145,9 @@ def match(self, cfn: Template) -> RuleMatches:
                 # by the customer so if any primary identifiers are read
                 # only we have to skip evaluation
                 primary_ids = schema.schema.get("primaryIdentifier", [])
+                if not primary_ids:
+                    continue
+
                 read_only_ids = schema.schema.get("readOnlyProperties", [])
 
                 if any(id in read_only_ids for id in primary_ids):

src/cfnlint/schema/manager.py

@@ -152,47 +152,49 @@ def get_resource_schema(self, region: str, resource_type: str) -> Schema:
         Returns:
             dict: returns the schema
         """
-        resource_type = self._normalize_resource_type(resource_type)
+        if resource_type not in self._registry_schemas:
+            resource_type = self._normalize_resource_type(resource_type)
 
         if resource_type in self._removed_types:
             raise ResourceNotFoundError(resource_type, region)
 
         reg = ToPy(region)
         rt = ToPy(resource_type)
 
-        schema = self._schemas[reg.name].get(resource_type)
-        if schema is None:
-            # dynamically import the modules as needed
-            self._provider_schema_modules[reg.name] = __import__(
-                f"{self._root.module}.{reg.py}", fromlist=[""]
-            )
-            # check cfn-lint provided schemas
-            if resource_type in self._registry_schemas:
-                self._schemas[reg.name][rt.name] = self._registry_schemas[rt.name]
-                return self._schemas[reg.name][rt.name]
-
-            # load the schema
-            if f"{rt.provider}.json" in self._provider_schema_modules[reg.name].cached:
-                schema_cached = copy(
-                    self.get_resource_schema(
-                        region=self._region_primary.name,
-                        resource_type=rt.name,
-                    )
+        schema = self._schemas[reg.name].get(rt.name)
+        if schema is not None:
+            return schema
+
+        # dynamically import the modules as needed
+        self._provider_schema_modules[reg.name] = __import__(
+            f"{self._root.module}.{reg.py}", fromlist=[""]
+        )
+        # check cfn-lint provided schemas
+        if rt.name in self._registry_schemas:
+            self._schemas[reg.name][rt.name] = self._registry_schemas[rt.name]
+            return self._schemas[reg.name][rt.name]
+
+        # load the schema
+        if f"{rt.provider}.json" in self._provider_schema_modules[reg.name].cached:
+            schema_cached = copy(
+                self.get_resource_schema(
+                    region=self._region_primary.name,
+                    resource_type=rt.name,
                 )
-                schema_cached.is_cached = True
-                self._schemas[reg.name][rt.name] = schema_cached
-                return self._schemas[reg.name][rt.name]
-            try:
-                self._schemas[reg.name][rt.name] = Schema(
-                    load_resource(
-                        self._provider_schema_modules[reg.name],
-                        filename=f"{rt.provider}.json",
-                    )
+            )
+            schema_cached.is_cached = True
+            self._schemas[reg.name][rt.name] = schema_cached
+            return self._schemas[reg.name][rt.name]
+        try:
+            self._schemas[reg.name][rt.name] = Schema(
+                load_resource(
+                    self._provider_schema_modules[reg.name],
+                    filename=f"{rt.provider}.json",
                 )
-            except Exception as e:
-                raise ResourceNotFoundError(rt.name, region) from e
+            )
             return self._schemas[reg.name][rt.name]
-        return schema
+        except Exception as e:
+            raise ResourceNotFoundError(rt.name, region) from e
 
     @lru_cache(maxsize=None)
     def get_resource_types(self, region: str) -> list[str]:

test/fixtures/templates/bad/resources/primary_identifiers.yaml

@@ -154,3 +154,11 @@ Resources:
         - !Ref 'AWS::NoValue'
   BadType:
     Type: !Ref AWS::Region
+  Module1:
+    Type: MyCompany::MODULE
+    Properties:
+      Attribute1: test
+  Module2:
+    Type: MyCompany::MODULE
+    Properties:
+      Attribute2: test

test/unit/module/schema/test_manager.py

@@ -271,3 +271,15 @@ def test_removed_types(self):
 
         with self.assertRaises(ResourceNotFoundError):
             self.manager.get_resource_schema(region, rt)
+
+    def test_type_normalization(self):
+
+        rt = "MyCompany::MODULE"
+        schema = self.manager.get_resource_schema("us-east-1", rt)
+
+        assert schema.schema.get("typeName") == "Module"
+
+        self.manager.get_resource_schema.cache_clear()
+        self.manager._registry_schemas[rt] = True
+        schema = self.manager.get_resource_schema("us-east-1", rt)
+        assert schema is True

tox.ini

@@ -5,9 +5,7 @@ isolated_build = true
 [testenv]
 skip_install = True
 commands =
-  pip install -e .
-  pip install -e .[sarif]
-  pip install -e .[junit]
+  pip install -e .[full]
   coverage run -m pytest {posargs:test}
   coverage xml
 deps =