Validate a SAM resource property using schema models

Author: GavinZZ

requirements/base.txt

@@ -1,3 +1,6 @@
 boto3>=1.19.5,==1.*
 jsonschema<5,>=3.2  # TODO: evaluate risk of removing jsonschema 3.x support
 typing_extensions>=4.4,<5 # 3.7 doesn't have Literal
+
+# resource validation & schema generation, requiring features in >=1.10
+pydantic~=1.10

requirements/dev.txt

@@ -29,6 +29,3 @@ mypy~=1.0.0
 boto3-stubs[appconfig,serverlessrepo]>=1.19.5,==1.*
 types-PyYAML~=5.4
 types-jsonschema~=3.2
-
-# schema generation, requiring features in >=1.10
-pydantic~=1.10

samtranslator/validator/resource_validator.py

@@ -0,0 +1,40 @@
+"""A resource validator to help validate resource properties and raise exception when some value is unexpected."""
+from typing import Any, Dict, Type
+
+from pydantic import BaseModel
+
+
+class ResourceModel:
+    """
+    Wrapper class around a SAM schema BaseModel to with a new functional "get" method
+    """
+
+    def __init__(self, model: BaseModel) -> None:
+        self.model = model
+
+    def get(self, attr: str, default_value: Any = None) -> Any:
+        attr_value = self.model.__dict__.get(attr, default_value)
+        if isinstance(attr_value, BaseModel):
+            if "__root__" in attr_value.__dict__:
+                return attr_value.__dict__["__root__"]
+            return ResourceModel(attr_value)
+        return attr_value
+
+    def __getitem__(self, attr):
+        attr_value = self.model.__dict__[attr]
+        if isinstance(attr_value, BaseModel):
+            if "__root__" in attr_value.__dict__:
+                return attr_value.__dict__["__root__"]
+            return ResourceModel(attr_value)
+        return attr_value
+
+
+def to_resource_model(resource_properties: Dict[Any, Any], cls: Type[BaseModel]) -> ResourceModel:
+    """
+    Given properties of a SAM resource return a typed object from the definitions of SAM schema model
+
+    param:
+        resource_properties: properties from input template
+        cls: SAM schema models
+    """
+    return ResourceModel(cls(**resource_properties))

tests/validator/input/connector/error_connector.yaml

@@ -0,0 +1,55 @@
+Resources:
+  NoSourceConnector:
+    Type: AWS::Serverless::Connector
+    Properties:
+      Destination:
+        Id: MyQueue1
+      Permissions:
+      - Write
+
+  NoDestConnector:
+    Type: AWS::Serverless::Connector
+    Properties:
+      Source:
+        Id: MyQueue1
+      Permissions:
+      - Write
+
+  NoPermissionConnector:
+    Type: AWS::Serverless::Connector
+    Properties:
+      Source:
+        Id: MyQueue1
+      Destination:
+        Id: MyTable
+
+  InvalidPermissionConnector:
+    Type: AWS::Serverless::Connector
+    Properties:
+      Source:
+        Id: MyQueue1
+      Destination:
+        Id: MyTable
+      Permissions:
+      - Invoke
+
+  InvalidPermissionTypeConnector:
+    Type: AWS::Serverless::Connector
+    Properties:
+      Source:
+        Id: MyQueue1
+      Destination:
+        Id: MyTable
+      Permissions:
+        Hello: World
+
+  InvalidIdTypeConnector:
+    Type: AWS::Serverless::Connector
+    Properties:
+      Source:
+        Id:
+          Key: Value
+      Destination:
+        Id: MyTable
+      Permissions:
+      - Read

tests/validator/test_resource_validator.py

@@ -0,0 +1,52 @@
+import os.path
+from unittest import TestCase
+
+from pydantic.error_wrappers import ValidationError
+from samtranslator.validator.resource_validator import to_resource_model
+from samtranslator.yaml_helper import yaml_parse
+from schema_source.aws_serverless_connector import Properties as ConnectorProperties
+
+BASE_PATH = os.path.dirname(__file__)
+CONNECTOR_INPUT_FOLDER = os.path.join(BASE_PATH, "input", "connector")
+
+
+class TestResourceModel(TestCase):
+    def setUp(self) -> None:
+        self.connector_template = {
+            "Source": {
+                "Arn": "random-arn",
+                "Type": "random-type",
+            },
+            "Destination": {"Id": "MyTable"},
+            "Permissions": ["Read"],
+        }
+
+    def test_resource_model_get(self):
+        connector_model = to_resource_model(
+            self.connector_template,
+            ConnectorProperties,
+        )
+        self.assertEqual(connector_model.get("Source").get("Arn"), "random-arn")
+        self.assertEqual(connector_model.get("Source").get("Type"), "random-type")
+        self.assertEqual(connector_model.get("Source").get("Id"), None)
+        self.assertEqual(connector_model.get("Destination").get("Id"), "MyTable")
+        self.assertEqual(connector_model.get("Permissions"), ["Read"])
+        self.assertEqual(connector_model.get("FakeProperty"), None)
+
+        self.assertEqual(connector_model["Source"]["Arn"], "random-arn")
+        self.assertEqual(connector_model["Source"]["Type"], "random-type")
+        self.assertEqual(connector_model["Destination"]["Id"], "MyTable")
+        self.assertEqual(connector_model["Permissions"], ["Read"])
+
+
+class TestResourceValidatorFailure(TestCase):
+    def test_to_resource_model_error_connector_template(self):
+        manifest = yaml_parse(open(os.path.join(CONNECTOR_INPUT_FOLDER, "error_connector.yaml")))
+        for _, resource in manifest["Resources"].items():
+            properties = resource["Properties"]
+
+            with self.assertRaisesRegex(
+                ValidationError,
+                "[1-9] validation error for Properties(.|\n)+(Source|Destination|Permissions)(.|\n)*(field required)|(unexpected value)|(str type expected)|(value is not a valid list).+",
+            ):
+                to_resource_model(properties, ConnectorProperties)