Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 17 additions & 3 deletions samcli/lib/bootstrap/nested_stack/nested_stack_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
from samcli.lib.providers.sam_function_provider import SamFunctionProvider
from samcli.lib.sync.exceptions import InvalidRuntimeDefinitionForFunction
from samcli.lib.utils import osutils
from samcli.lib.utils.cloudformation import is_intrinsic_function
from samcli.lib.utils.osutils import BUILD_DIR_PERMISSIONS
from samcli.lib.utils.packagetype import ZIP
from samcli.lib.utils.resources import AWS_LAMBDA_FUNCTION, AWS_SERVERLESS_FUNCTION
Expand Down Expand Up @@ -118,16 +119,29 @@ def _get_template_folder(self) -> Path:
return Path(self._stack.get_output_template_path(self._build_dir)).parent

def _add_layer(self, dependencies_dir: str, function: Function, resources: Dict):
# Check if Layers is an intrinsic function before creating the layer
function_properties = cast(Dict, resources.get(function.name)).get("Properties", {})
function_layers = function_properties.get("Layers", [])

if is_intrinsic_function(function_layers):
LOG.warning(
"Function %s has Layers defined as an intrinsic function (%s). "
"Auto-dependency layer creation is not supported for functions with dynamic layer configuration. "
"Skipping auto-dependency layer for this function.",
function.name,
list(function_layers.keys())[0],
)
return

# Create the layer
layer_logical_id = NestedStackBuilder.get_layer_logical_id(function.full_path)
layer_location = self.update_layer_folder(
str(self._get_template_folder()), dependencies_dir, layer_logical_id, function.full_path, function.runtime
)

layer_output_key = self._nested_stack_builder.add_function(self._stack_name, layer_location, function)

# add layer reference back to function
function_properties = cast(Dict, resources.get(function.name)).get("Properties", {})
function_layers = function_properties.get("Layers", [])
# Add layer reference back to function
function_layers.append({"Fn::GetAtt": [NESTED_STACK_NAME, f"Outputs.{layer_output_key}"]})
function_properties["Layers"] = function_layers

Expand Down
15 changes: 14 additions & 1 deletion samcli/lib/providers/sam_function_provider.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
from samcli.commands.local.cli_common.user_exceptions import InvalidLayerVersionArn
from samcli.lib.build.exceptions import MissingFunctionHandlerException
from samcli.lib.providers.exceptions import InvalidLayerReference, MissingFunctionNameException
from samcli.lib.utils.cloudformation import is_intrinsic_function
from samcli.lib.utils.colors import Colored, Colors
from samcli.lib.utils.file_observer import FileObserver
from samcli.lib.utils.packagetype import IMAGE, ZIP
Expand Down Expand Up @@ -571,7 +572,19 @@ def _parse_layer_info(

I.E: list_of_layers = ["layer1", "layer2"] the return would be [Layer("layer1"), Layer("layer2")]
"""
layers = []
layers: List[LayerVersion] = []

# Check if list_of_layers is an intrinsic function
if is_intrinsic_function(list_of_layers):
# At this point we know it's a dict with one key (the intrinsic function name)
intrinsic_dict = cast(Dict, list_of_layers)
intrinsic_name = next(iter(intrinsic_dict.keys())) if intrinsic_dict else "unknown"
LOG.debug(
"Layers property is defined as an intrinsic function (%s). "
"Skipping layer parsing as the actual layer list cannot be determined at build time.",
intrinsic_name,
)
return layers

if locate_layer_nested and stacks and function_id:
# The layer can be a parameter pass from parent stack, we need to locate to where the
Expand Down
44 changes: 44 additions & 0 deletions samcli/lib/utils/cloudformation.py
Original file line number Diff line number Diff line change
Expand Up @@ -223,3 +223,47 @@ def list_active_stack_names(boto_client_provider: BotoProviderType, show_nested_
continue
yield stack_summary.get("StackName")
next_token = list_stacks_result.get("NextToken")


# CloudFormation intrinsic function names
# https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/intrinsic-function-reference.html
CLOUDFORMATION_INTRINSIC_FUNCTIONS = {
"Fn::Base64",
"Fn::Cidr",
"Fn::FindInMap",
"Fn::ForEach",
"Fn::GetAtt",
"Fn::GetAZs",
"Fn::ImportValue",
"Fn::Join",
"Fn::Length",
"Fn::Select",
"Fn::Split",
"Fn::Sub",
"Fn::ToJsonString",
"Fn::Transform",
"Fn::And",
"Fn::Equals",
"Fn::If",
"Fn::Not",
"Fn::Or",
"Ref",
}


def is_intrinsic_function(value: Any) -> bool:
"""
Checks if a value is a CloudFormation intrinsic function.

When YAML templates are parsed, intrinsic functions are represented as OrderedDict
with a single key that matches one of the CloudFormation intrinsic function names.
"""

if not isinstance(value, dict):
return False

if len(value) != 1:
return False

key = next(iter(value.keys()))
return key in CLOUDFORMATION_INTRINSIC_FUNCTIONS
22 changes: 22 additions & 0 deletions tests/unit/commands/local/lib/test_sam_function_provider.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import os
import posixpath
from unittest import TestCase
from collections import OrderedDict
from unittest.mock import patch, PropertyMock, Mock, call

from parameterized import parameterized
Expand Down Expand Up @@ -2039,6 +2040,27 @@ def test_return_empty_list_on_no_layers(self):

self.assertEqual(actual, [])

def test_return_empty_list_on_intrinsic_function_layers(self):
"""Test that intrinsic functions in Layers property return empty list"""

resources = {"Function": {"Type": "AWS::Serverless::Function", "Properties": {}}}

# Test with Fn::If intrinsic function
list_of_layers = OrderedDict(
[("Fn::If", ["UseLayer", ["arn:aws:lambda:us-east-1:123456789012:layer:MyLayer:1"], []])]
)
actual = SamFunctionProvider._parse_layer_info(
Mock(stack_path=STACK_PATH, location="template.yaml", resources=resources), list_of_layers
)
self.assertEqual(actual, [])

# Test with Ref intrinsic function
list_of_layers = OrderedDict([("Ref", "LayerParameter")])
actual = SamFunctionProvider._parse_layer_info(
Mock(stack_path=STACK_PATH, location="template.yaml", resources=resources), list_of_layers
)
self.assertEqual(actual, [])

@patch.object(SamFunctionProvider, "_locate_layer_from_nested")
def test_layers_with_search_layer(self, locate_layer_mock):
layer = {"Ref", "layer"}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -244,3 +244,53 @@ def test_skipping_dependency_copy_when_function_has_no_dependencies(
@parameterized.expand([("python3.8", True), ("ruby3.2", False)])
def test_is_runtime_supported(self, runtime, supported):
self.assertEqual(NestedStackManager.is_runtime_supported(runtime), supported)

@patch("samcli.lib.bootstrap.nested_stack.nested_stack_manager.move_template")
@patch("samcli.lib.bootstrap.nested_stack.nested_stack_manager.osutils")
@patch("samcli.lib.bootstrap.nested_stack.nested_stack_manager.os.path.isdir")
def test_with_intrinsic_function_layers_skips_auto_layer(
self, patched_isdir, patched_osutils, patched_move_template
):
"""Test that functions with intrinsic function Layers are skipped with a warning"""
resources = {
"MyFunction": {
"Type": AWS_SERVERLESS_FUNCTION,
"Properties": {
"Runtime": "python3.8",
"Handler": "FakeHandler",
"Layers": {"Fn::If": ["HasLayer", [{"Ref": "MyLayer"}], []]},
},
}
}
self.stack.resources = resources
template = {"Resources": resources}

# prepare build graph
dependencies_dir = Mock()
function = Mock()
function.name = "MyFunction"
functions = [function]
build_graph = Mock()
function_definition_mock = Mock(dependencies_dir=dependencies_dir, functions=functions)
build_graph.get_function_build_definition_with_logical_id.return_value = function_definition_mock
app_build_result = ApplicationBuildResult(build_graph, {"MyFunction": "path/to/build/dir"})
patched_isdir.return_value = True

nested_stack_manager = NestedStackManager(
self.stack, self.stack_name, self.build_dir, template, app_build_result
)

with patch.object(nested_stack_manager, "_add_layer_readme_info"):
result = nested_stack_manager.generate_auto_dependency_layer_stack()

# Should not create nested stack since function was skipped
patched_move_template.assert_not_called()

# Template should remain unchanged
self.assertEqual(template, result)

# Layers should still be the intrinsic function (not modified)
self.assertEqual(
result.get("Resources", {}).get("MyFunction", {}).get("Properties", {}).get("Layers"),
{"Fn::If": ["HasLayer", [{"Ref": "MyLayer"}], []]},
)
41 changes: 41 additions & 0 deletions tests/unit/lib/utils/test_cloudformation.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,15 @@
from collections import OrderedDict
from unittest import TestCase
from unittest.mock import patch, Mock, ANY, call

from botocore.exceptions import ClientError
from parameterized import parameterized

from samcli.lib.utils.cloudformation import (
CloudFormationResourceSummary,
get_resource_summaries,
get_resource_summary,
is_intrinsic_function,
list_active_stack_names,
get_resource_summary_from_physical_id,
)
Expand Down Expand Up @@ -215,3 +218,41 @@ def test_get_resource_summary_from_physical_id_fail(self, patched_log):
resource_summary = get_resource_summary_from_physical_id(patched_cfn_client_provider, "invalid_physical_id")
self.assertIsNone(resource_summary)
patched_log.debug.assert_called_once()


class TestIsIntrinsicFunction(TestCase):
"""Tests for the is_intrinsic_function utility"""

@parameterized.expand(
[
("Fn::If", ["Condition", "Value1", "Value2"]),
("Ref", "MyParameter"),
("Fn::GetAtt", ["Resource", "Attribute"]),
("Fn::Sub", "${AWS::StackName}-bucket"),
("Fn::ForEach", ["Item", ["a", "b"], {"Key": "Value"}]),
("Fn::Length", ["item1", "item2", "item3"]),
("Fn::ToJsonString", {"key": "value"}),
("Fn::Base64", "value"),
("Fn::Join", ["-", ["a", "b"]]),
("Fn::Select", [0, ["a", "b"]]),
]
)
def test_intrinsic_functions_detected(self, function_name, function_value):
"""Test that CloudFormation intrinsic functions are correctly detected"""
value = OrderedDict([(function_name, function_value)])
self.assertTrue(is_intrinsic_function(value))

@parameterized.expand(
[
(["item1", "item2", "item3"], "list"),
({"key1": "value1", "key2": "value2"}, "multi_key_dict"),
({"NotAnIntrinsic": "value"}, "single_key_non_intrinsic"),
("just a string", "string"),
(None, "none"),
({}, "empty_dict"),
(123, "number"),
]
)
def test_non_intrinsic_values_not_detected(self, value, description):
"""Test that non-intrinsic values are not detected as intrinsic functions"""
self.assertFalse(is_intrinsic_function(value))
Loading