test_model.py

from typing import Any, List
from unittest import TestCase
from unittest.mock import Mock

import pytest
from samtranslator.intrinsics.resource_refs import SupportedResourceReferences
from samtranslator.model import PropertyType, Resource, ResourceTypeResolver, SamResourceMacro
from samtranslator.model.exceptions import InvalidResourceException
from samtranslator.plugins import LifeCycleEvents


def valid_if_true(value, should_raise=True):
    """Validator that passes if the input is True."""
    if value is True:
        return True

    if should_raise:
        raise TypeError
    return False


class DummyResource(Resource):
    resource_type = "AWS::Dummy::Resource"
    property_types = {
        "RequiredProperty": PropertyType(True, valid_if_true),
        "OptionalProperty": PropertyType(False, valid_if_true),
    }


@pytest.mark.parametrize(
    "logical_id,resource_dict,expected_exception",
    [
        # Valid required property
        ("id", {"Type": "AWS::Dummy::Resource", "Properties": {"RequiredProperty": True}}, None),
        # Valid required property and valid optional property
        (
            "id",
            {"Type": "AWS::Dummy::Resource", "Properties": {"RequiredProperty": True, "OptionalProperty": True}},
            None,
        ),
        # Required property not provided
        ("id", {"Type": "AWS::Dummy::Resource", "Properties": {"OptionalProperty": True}}, InvalidResourceException),
        # Required property provided, but invalid
        ("id", {"Type": "AWS::Dummy::Resource", "Properties": {"RequiredProperty": False}}, InvalidResourceException),
        # Property with invalid name provided
        (
            "id",
            {"Type": "AWS::Dummy::Resource", "Properties": {"RequiredProperty": True, "InvalidProperty": True}},
            InvalidResourceException,
        ),
        # Missing Properties
        ("id", {"Type": "AWS::Other::Other"}, InvalidResourceException),
        # Missing Type
        ("id", {"Properties": {"RequiredProperty": True, "OptionalProperty": True}}, InvalidResourceException),
        # Valid Type with invalid Properties
        ("id", {"Type": "AWS::Dummy::Resource", "Properties": "Invalid"}, InvalidResourceException),
        # Valid Properties with invalid Type
        (
            "id",
            {"Type": "AWS::Invalid::Invalid", "Properties": {"RequiredProperty": True, "OptionalProperty": True}},
            InvalidResourceException,
        ),
        # Invalid logical_id
        (
            "invalid_id",
            {"Type": "AWS::Dummy::Resource", "Properties": {"RequiredProperty": True, "OptionalProperty": True}},
            InvalidResourceException,
        ),
        # intrinsic function
        (
            "id",
            {"Type": "AWS::Dummy::Resource", "Properties": {"RequiredProperty": {"Fn::Any": ["logicalid", "Arn"]}}},
            None,
        ),
    ],
)
def test_resource_type_validation(logical_id, resource_dict, expected_exception):
    if not expected_exception:
        resource = DummyResource.from_dict(logical_id, resource_dict)
        for name, value in resource_dict["Properties"].items():
            assert (
                getattr(resource, name) == value
            ), f"resource did not have expected property attribute {name} with value {value}"

        actual_to_dict = resource.to_dict()
        expected_to_dict = {"id": resource_dict}
        assert (
            actual_to_dict == expected_to_dict
        ), f"to_dict() returned different values from what was passed to from_dict(); expected {expected_to_dict}, got {actual_to_dict}"
    else:
        with pytest.raises(expected_exception):
            resource = DummyResource.from_dict(logical_id, resource_dict)


class TestResourceAttributes(TestCase):
    class MyResource(Resource):
        resource_type = "foo"
        property_types = {}

    def test_to_dict(self):
        """Tests if resource attributes are correctly set and converted to dictionary"""

        empty_resource_dict = {"id": {"Type": "foo", "Properties": {}}}
        dict_with_attributes = {
            "id": {"Type": "foo", "Properties": {}, "UpdatePolicy": "update", "DeletionPolicy": {"foo": "bar"}}
        }
        dict_with_attributes2 = {
            "id": {
                "Type": "foo",
                "Properties": {},
                "UpdateReplacePolicy": "update",
                "Metadata": {"foo": "bar"},
                "Condition": "con",
            }
        }

        r = self.MyResource("id")
        self.assertEqual(r.to_dict(), empty_resource_dict)

        r = self.MyResource("id", attributes={"UpdatePolicy": "update", "DeletionPolicy": {"foo": "bar"}})
        self.assertEqual(r.to_dict(), dict_with_attributes)

        r = self.MyResource(
            "id", attributes={"UpdateReplacePolicy": "update", "Metadata": {"foo": "bar"}, "Condition": "con"}
        )
        self.assertEqual(r.to_dict(), dict_with_attributes2)

    def test_invalid_attr(self):
        with pytest.raises(KeyError):
            # Unsupported attributes cannot be added to the resource
            self.MyResource("id", attributes={"foo": "bar"})

        # But unsupported properties will silently be ignored when deserialization from dictionary
        with_unsupported_attributes = {
            "Type": "foo",
            "Properties": {},
            "DeletionPolicy": "foo",
            "UnsupportedPolicy": "bar",
        }
        r = self.MyResource.from_dict("id", resource_dict=with_unsupported_attributes)
        self.assertEqual(r.get_resource_attribute("DeletionPolicy"), "foo")

        with pytest.raises(KeyError):
            r.get_resource_attribute("UnsupportedPolicy")

    def test_from_dict(self):
        no_attribute = {"Type": "foo", "Properties": {}}
        all_supported_attributes = {
            "Type": "foo",
            "Properties": {},
            "UpdatePolicy": "update",
            "DeletionPolicy": [1, 2, 3],
            "UpdateReplacePolicy": "update",
            "Metadata": {"foo": "bar"},
            "Condition": "con",
        }

        r = self.MyResource.from_dict("id", resource_dict=no_attribute)
        self.assertEqual(r.logical_id, "id")  # Just making sure the resource got created

        r = self.MyResource.from_dict("id", resource_dict=all_supported_attributes)
        self.assertEqual(r.get_resource_attribute("DeletionPolicy"), [1, 2, 3])
        self.assertEqual(r.get_resource_attribute("UpdatePolicy"), "update")
        self.assertEqual(r.get_resource_attribute("UpdateReplacePolicy"), "update")
        self.assertEqual(r.get_resource_attribute("Metadata"), {"foo": "bar"})
        self.assertEqual(r.get_resource_attribute("Condition"), "con")


class TestResourceRuntimeAttributes(TestCase):
    def test_resource_must_override_runtime_attributes(self):
        class NewResource(Resource):
            resource_type = "foo"
            property_types = {}
            runtime_attrs = {"attr1": Mock(), "attr2": Mock()}

            runtime_attrs["attr1"].return_value = "value1"
            runtime_attrs["attr2"].return_value = "value2"

        logical_id = "SomeId"
        resource = NewResource(logical_id)
        self.assertEqual("value1", resource.get_runtime_attr("attr1"))
        self.assertEqual("value2", resource.get_runtime_attr("attr2"))

        with self.assertRaises(KeyError):
            resource.get_runtime_attr("invalid_attribute")

    def test_resource_default_runtime_attributes(self):
        # There are no attributes by default
        class NewResource(Resource):
            resource_type = "foo"
            property_types = {}

        resource = NewResource("SomeId")
        self.assertEqual(0, len(resource.runtime_attrs))


class TestSamResourceReferableProperties(TestCase):
    class ResourceType1(Resource):
        resource_type = "resource_type1"
        property_types = {}

    class ResourceType2(Resource):
        resource_type = "resource_type2"
        property_types = {}

    class ResourceType3(Resource):
        resource_type = "resource_type3"
        property_types = {}

    def setUp(self):
        self.supported_resource_refs = SupportedResourceReferences()

    def test_must_get_property_for_available_resources(self):
        class NewSamResource(SamResourceMacro):
            resource_type = "foo"
            property_types = {}
            referable_properties = {"prop1": "resource_type1", "prop2": "resource_type2", "prop3": "resource_type3"}

            def to_cloudformation(self, **kwargs: Any) -> List[Any]:
                return []

        sam_resource = NewSamResource("SamLogicalId")

        cfn_resources = [self.ResourceType1("logicalId1"), self.ResourceType2("logicalId2")]

        self.supported_resource_refs = sam_resource.get_resource_references(cfn_resources, self.supported_resource_refs)

        self.assertEqual("logicalId1", self.supported_resource_refs.get("SamLogicalId", "prop1"))
        self.assertEqual("logicalId2", self.supported_resource_refs.get("SamLogicalId", "prop2"))

        # there is no cfn resource of for "prop3" in the cfn_resources list
        self.assertEqual(None, self.supported_resource_refs.get("SamLogicalId", "prop3"))

        # Must add only for the given SAM resource
        self.assertEqual(1, len(self.supported_resource_refs))

    def test_must_work_with_two_resources_of_same_type(self):
        class NewSamResource(SamResourceMacro):
            resource_type = "foo"
            property_types = {}
            referable_properties = {"prop1": "resource_type1", "prop2": "resource_type2", "prop3": "resource_type3"}

            def to_cloudformation(self, **kwargs: Any) -> List[Any]:
                return []

        sam_resource1 = NewSamResource("SamLogicalId1")
        sam_resource2 = NewSamResource("SamLogicalId2")

        cfn_resources = [self.ResourceType1("logicalId1"), self.ResourceType2("logicalId2")]

        self.supported_resource_refs = sam_resource1.get_resource_references(
            cfn_resources, self.supported_resource_refs
        )

        self.supported_resource_refs = sam_resource2.get_resource_references(
            cfn_resources, self.supported_resource_refs
        )

        self.assertEqual("logicalId1", self.supported_resource_refs.get("SamLogicalId1", "prop1"))
        self.assertEqual("logicalId2", self.supported_resource_refs.get("SamLogicalId1", "prop2"))
        self.assertEqual("logicalId1", self.supported_resource_refs.get("SamLogicalId2", "prop1"))
        self.assertEqual("logicalId2", self.supported_resource_refs.get("SamLogicalId2", "prop2"))

        self.assertEqual(2, len(self.supported_resource_refs))

    def test_must_skip_unknown_resource_types(self):
        class NewSamResource(SamResourceMacro):
            resource_type = "foo"
            property_types = {}
            referable_properties = {"prop1": "foo", "prop2": "bar"}

            def to_cloudformation(self, **kwargs: Any) -> List[Any]:
                return []

        sam_resource = NewSamResource("SamLogicalId")

        # None of the CFN resource types are in the referable list
        cfn_resources = [self.ResourceType1("logicalId1"), self.ResourceType2("logicalId2")]

        self.supported_resource_refs = sam_resource.get_resource_references(cfn_resources, self.supported_resource_refs)

        self.assertEqual(0, len(self.supported_resource_refs))

    def test_must_skip_if_no_supported_properties(self):
        class NewSamResource(SamResourceMacro):
            resource_type = "foo"
            property_types = {}
            referable_properties = {}

            def to_cloudformation(self, **kwargs: Any) -> List[Any]:
                return []

        sam_resource = NewSamResource("SamLogicalId")

        cfn_resources = [self.ResourceType1("logicalId1"), self.ResourceType2("logicalId2")]

        self.supported_resource_refs = sam_resource.get_resource_references(cfn_resources, self.supported_resource_refs)

        self.assertEqual(0, len(self.supported_resource_refs))

    def test_must_skip_if_no_resources(self):
        class NewSamResource(SamResourceMacro):
            resource_type = "foo"
            property_types = {}
            referable_properties = {"prop1": "resource_type1"}

            def to_cloudformation(self, **kwargs: Any) -> List[Any]:
                return []

        sam_resource = NewSamResource("SamLogicalId")

        cfn_resources = []

        self.supported_resource_refs = sam_resource.get_resource_references(cfn_resources, self.supported_resource_refs)

        self.assertEqual(0, len(self.supported_resource_refs))

    def test_must_raise_if_input_is_absent(self):
        class NewSamResource(SamResourceMacro):
            resource_type = "foo"
            property_types = {}
            referable_properties = {"prop1": "resource_type1"}

            def to_cloudformation(self, **kwargs: Any) -> List[Any]:
                return []

        sam_resource = NewSamResource("SamLogicalId")

        cfn_resources = [self.ResourceType1("logicalId1")]

        with self.assertRaises(ValueError):
            sam_resource.get_resource_references(cfn_resources, None)


class TestResourceTypeResolver(TestCase):
    def test_can_resolve_must_handle_null_resource_dict(self):
        resolver = ResourceTypeResolver()

        self.assertFalse(resolver.can_resolve(None))

    def test_can_resolve_must_handle_non_dict(self):
        resolver = ResourceTypeResolver()

        self.assertFalse(resolver.can_resolve("some string value"))

    def test_can_resolve_must_handle_dict_without_type(self):
        resolver = ResourceTypeResolver()

        self.assertFalse(resolver.can_resolve({"a": "b"}))

    def test_can_resolve_must_handle_known_types(self):
        resolver = ResourceTypeResolver()
        resolver.resource_types = {"type1": DummyResource("id")}

        self.assertTrue(resolver.can_resolve({"Type": "type1"}))

    def test_can_resolve_must_handle_unknown_types(self):
        resolver = ResourceTypeResolver()
        resolver.resource_types = {"type1": DummyResource("id")}

        self.assertFalse(resolver.can_resolve({"Type": "AWS::Lambda::Function"}))


class TestSamPluginsInResource(TestCase):
    def test_must_act_on_plugins_before_resource_creation(self):
        resource_type = "AWS::Dummy::Resource"
        resource_dict = {"Type": resource_type, "Properties": {"RequiredProperty": True}}
        expected_properties = {"RequiredProperty": True}

        mock_sam_plugins = Mock()
        DummyResource.from_dict("logicalId", resource_dict, sam_plugins=mock_sam_plugins)

        mock_sam_plugins.act.assert_called_once_with(
            LifeCycleEvents.before_transform_resource, "logicalId", resource_type, expected_properties
        )

    def test_must_act_on_plugins_for_resource_having_no_properties(self):
        resource_type = "MyResourceType"

        class MyResource(Resource):
            property_types = {}
            resource_type = "MyResourceType"

        # No Properties for this resource
        resource_dict = {"Type": resource_type}
        expected_properties = {}

        mock_sam_plugins = Mock()
        MyResource.from_dict("logicalId", resource_dict, sam_plugins=mock_sam_plugins)

        mock_sam_plugins.act.assert_called_once_with(
            LifeCycleEvents.before_transform_resource, "logicalId", resource_type, expected_properties
        )