diff --git a/faster_sam/cloudformation.py b/faster_sam/cloudformation.py index f8fa08b6..deccf6b4 100644 --- a/faster_sam/cloudformation.py +++ b/faster_sam/cloudformation.py @@ -1,7 +1,9 @@ import base64 import logging from enum import Enum +import os from pathlib import Path +import re from typing import Any, Dict, List, Optional, Union import yaml @@ -584,7 +586,16 @@ def eval(function: Dict[str, Any], template: Dict[str, Any]) -> Any: NotImplementedError If the intrinsic function is not implemented. """ - implemented = ("Fn::Base64", "Fn::FindInMap", "Ref") + implemented = ( + "Fn::Base64", + "Fn::FindInMap", + "Fn::GetAtt", + "Fn::Join", + "Fn::Select", + "Fn::Split", + "Fn::Sub", + "Ref", + ) fun, val = list(function.items())[0] @@ -597,6 +608,21 @@ def eval(function: Dict[str, Any], template: Dict[str, Any]) -> Any: if "Fn::FindInMap" == fun: return IntrinsicFunctions.find_in_map(val, template) + if "Fn::GetAtt" == fun: + return IntrinsicFunctions.get_att(val, template) + + if "Fn::Join" == fun: + return IntrinsicFunctions.join(val, template) + + if "Fn::Select" == fun: + return IntrinsicFunctions.select(val, template) + + if "Fn::Split" == fun: + return IntrinsicFunctions.split(val, template) + + if "Fn::Sub" == fun: + return IntrinsicFunctions.sub(val, template) + if "Ref" == fun: return IntrinsicFunctions.ref(val, template) @@ -676,3 +702,233 @@ def ref(value: str, template: Dict[str, Any]) -> Optional[str]: # NOTE: this is a partial implementation return None + + @staticmethod + def get_att(value: Union[List[str], str], template: Dict[str, Any]) -> Optional[str]: + """ + Gets the value of an attribute from a CloudFormation template based on a list + of logical name and attribute name. + + Parameters + ---------- + value : List[Any] + List containing the logical name and attribute name + template : Dict[str, Any] + A dictionary representing the CloudFormation template. + + Returns + ------- + Optional[str] + The value of atribute name, or None if the keys are not found. + """ + if isinstance(value, str): + value = value.split(".") + + logical_name, attribute_name = value + + if logical_name not in template["Resources"]: + return None + + if isinstance(attribute_name, dict): + attribute_name = IntrinsicFunctions.eval(attribute_name, template) + + if attribute_name is None: + return None + + if attribute_name not in template["Resources"][logical_name]["Properties"]: + return None + + attribute_value = template["Resources"][logical_name]["Properties"][attribute_name] + + if isinstance(attribute_value, dict): + attribute_value = IntrinsicFunctions.eval(attribute_value, template) + + if attribute_value is None: + return None + + return attribute_value + + @staticmethod + def join(value: List[Any], template: Dict[str, Any]) -> Optional[str]: + """ + Joins elements in a list with a specified delimiter. + + Parameters + ---------- + value : List[Any] + A list containing two elements: the delimiter as the first element, + and the values to join as the second element. + template : Dict[str, Any] + A dictionary representing the CloudFormation template. + + Returns + ------- + Optional[str] + The joined string if successful; otherwise, None. + """ + delimiter, values = value + + for index, element in enumerate(values): + if isinstance(element, dict): + element = IntrinsicFunctions.eval(element, template) + + if element is None: + return None + + values[index] = element + + return delimiter.join(values) + + @staticmethod + def select(value: List[Any], template: Dict[str, Any]) -> Optional[str]: + """ + Selects a value from a list based on the given index. If the value at the index + is a dictionary, it evaluates it using CloudFormation template data. + Parameters + ---------- + value : List[Any] + A list containing values from which to select. + template : Dict[str, Any] + A dictionary representing the CloudFormation template. + Returns + ------- + Optional[str] + The selected value from the list, or None if any of the evaluated + values are None. + """ + index, objects = value + + if isinstance(index, dict): + index = IntrinsicFunctions.eval(index, template) + + if index is None: + return None + + if isinstance(objects, dict): + objects = IntrinsicFunctions.eval(objects, template) + + if objects is None: + return None + else: + for i, obj in enumerate(objects): + if isinstance(obj, dict): + objects[i] = IntrinsicFunctions.eval(obj, template) + + if objects[i] is None: + return None + + return objects[int(index)] + + @staticmethod + def split(value: List[Any], template: Dict[str, Any]) -> Optional[str]: + """ + Splits a list of values using a specified delimiter. + + Parameters + ---------- + value : List[Any] + A tuple containing the delimiter as its first element, followed + by a list of values to split. + template : Dict[str, Any] + A dictionary representing the CloudFormation template. + + Returns + ------- + Optional[str] + A list of strings resulting from splitting using the delimiter. + or None if any of the evaluated values are None. + """ + delimiter, source = value + + if isinstance(source, dict): + source = IntrinsicFunctions.eval(source, template) + + if source is None: + return None + + return source.split(delimiter) + + def replace_placeholders(string: str, matches: List[Any]): + pseudo_parameters = [ + "AWS::AccountId", + "AWS::NotificationARNs", + "AWS::NoValue", + "AWS::Partition", + "AWS::Region", + "AWS::StackId", + "AWS::StackName", + "AWS::URLSuffix", + ] + + result_string = string + + for match in matches: + if match in pseudo_parameters: + if match.replace("::", "_") in os.environ: + env_var = os.environ[match.replace("::", "_")] + result_string = result_string.replace(f"${{{match}}}", env_var) + else: + return None + else: + if match in os.environ: + env_var = os.environ[match] + result_string = result_string.replace(f"${{{match}}}", env_var) + else: + return None + + return result_string + + @staticmethod + def sub(value: List[Any], template: Dict[str, Any]) -> Optional[str]: + """ + Substitutes intrinsic functions and environment variables in the given value. + + Parameters + ---------- + value : List[Any] + A list containing the string to perform substitutions on and a dictionary + containing variable names and their corresponding values. + template : Dict[str, Any] + A dictionary representing the CloudFormation template. + + Returns + ------- + Optional[str] + The resulting string after performing substitutions, or None if any of the + variables or intrinsic functions could not be resolved. + """ + pattern = r"\${(.*?)}" + + if isinstance(value, list): + string, var_list = value + + for var_dict in var_list: + var_name, var_value = list(var_dict.items())[0] + + if isinstance(var_name, dict): + var_name = IntrinsicFunctions.eval(var_name, template) + + if var_name is None: + return None + + if isinstance(var_value, dict): + var_value = IntrinsicFunctions.eval(var_value, template) + + if var_value is None: + return None + + if var_name in string: + result = string.replace(f"${{{var_name}}}", str(var_value)) + else: + return None + + matches = re.findall(pattern, result) + + if not matches: + return result + + return IntrinsicFunctions.replace_placeholders(result, matches) + else: + matches = re.findall(pattern, value) + + return IntrinsicFunctions.replace_placeholders(value, matches) diff --git a/tests/fixtures/templates/example2.yml b/tests/fixtures/templates/example2.yml index 5eff81e0..325c122c 100644 --- a/tests/fixtures/templates/example2.yml +++ b/tests/fixtures/templates/example2.yml @@ -8,6 +8,9 @@ Description: > Parameters: Environment: Type: String + Handler: + type: String + Default: handler Mappings: Environments: @@ -31,6 +34,13 @@ Globals: - Environments - Ref: Environment - LogLevel + HANDLER: + Fn::Join: + - "." + - - fixtures + - handlers + - lambda_handler + - Ref: Handler Resources: ApiGateway: diff --git a/tests/fixtures/templates/example3.yml b/tests/fixtures/templates/example3.yml index ab2ae126..7680a67f 100644 --- a/tests/fixtures/templates/example3.yml +++ b/tests/fixtures/templates/example3.yml @@ -9,6 +9,15 @@ Parameters: Environment: Type: String Default: development + StageName: + Type: String + Default: v1 + AttributeName: + Type: String + Default: StageName + Accounts: + Type: String + Default: dev@gmail|dotz@gmail.com Mappings: Environments: @@ -29,7 +38,10 @@ Resources: Type: AWS::Serverless::Api Properties: Name: sam-api - StageName: v1 + StageName: + Ref: StageName + Tags: + Ref: Tags DefinitionBody: Fn::Transform: Name: AWS::Include @@ -63,3 +75,13 @@ Resources: - Environments - Ref: Environment - LogLevel + STAGE_NAME: + Fn::GetAtt: + - ApiGateway + - Ref: AttributeName + SENDER_ACCOUNT: + Fn::Select: + - 1 + - Fn::Split: + - "|" + - "dev@gmail|dotz@gmail.com" diff --git a/tests/fixtures/templates/example4.yml b/tests/fixtures/templates/example4.yml index 8788f263..a9916d76 100644 --- a/tests/fixtures/templates/example4.yml +++ b/tests/fixtures/templates/example4.yml @@ -9,15 +9,30 @@ Parameters: Environment: Type: String Default: development + Handler: + Type: String + Default: handler2 + HandlerList: + Type: String + Default: [handler, handler2] + Account: + Type: String + Default: 1234 + VarName: + Type: String + Default: account Mappings: Environments: development: LogLevel: DEBUG + Index: 1 staging: LogLevel: WARNING + Index: 1 production: LogLevel: ERROR + Index: 1 Globals: Function: @@ -27,6 +42,16 @@ Globals: Variables: ENVIRONMENT: Ref: Environment + STAGE_NAME: + Fn::GetAtt: ApiGateway.Region + HANDLER: + Fn::Select: + - Fn::FindInMap: + - Environments + - Ref: Environment + - Index + - - handler1 + - Ref: Handler Resources: ApiGateway: diff --git a/tests/test_cloudformation.py b/tests/test_cloudformation.py index d331536c..f9ac0afa 100644 --- a/tests/test_cloudformation.py +++ b/tests/test_cloudformation.py @@ -1,12 +1,13 @@ import io -import unittest +from unittest import mock from contextlib import contextmanager from pathlib import Path +import unittest import yaml import faster_sam.cloudformation as cf -from faster_sam.cloudformation import CloudformationTemplate +from faster_sam.cloudformation import CloudformationTemplate, IntrinsicFunctions @contextmanager @@ -212,14 +213,18 @@ def test_list_environment(self): "tests/fixtures/templates/example2.yml": { "ENVIRONMENT": "development", "LOG_LEVEL": "DEBUG", + "HANDLER": "fixtures.handlers.lambda_handler.handler", }, "tests/fixtures/templates/example3.yml": { "ENVIRONMENT": "development", "LOG_LEVEL": "DEBUG", + "STAGE_NAME": "v1", + "SENDER_ACCOUNT": "dotz@gmail.com", }, "tests/fixtures/templates/example4.yml": { "ENVIRONMENT": "development", "LOG_LEVEL": "DEBUG", + "HANDLER": "handler2", }, } @@ -265,6 +270,248 @@ def test_lambda_handler(self): self.assertEqual(handler_path, "tests.fixtures.handlers.lambda_handler.handler") +class TestIntrinsicFunctions(unittest.TestCase): + def test_getatt_function(self): + scenarios = { + "Resolved Function": { + "template": "tests/fixtures/templates/example3.yml", + "function": {"Fn::GetAtt": ["ApiGateway", {"Ref": "AttributeName"}]}, + "expected": "v1", + }, + "Attribute ID is None": { + "template": "tests/fixtures/templates/example3.yml", + "function": {"Fn::GetAtt": ["ApiFunction", {"Ref": "AttributeName"}]}, + "expected": None, + }, + "Attribute Name is None": { + "template": "tests/fixtures/templates/example3.yml", + "function": {"Fn::GetAtt": ["ApiGateway", "attibute"]}, + "expected": None, + }, + "Unresolved Attribute Function": { + "template": "tests/fixtures/templates/example3.yml", + "function": {"Fn::GetAtt": ["ApiGateway", {"Ref": "Attribute"}]}, + "expected": None, + }, + "Unresolved Function Return Value": { + "template": "tests/fixtures/templates/example3.yml", + "function": {"Fn::GetAtt": ["ApiGateway", "Tags"]}, + "expected": None, + }, + } + + for key, values in scenarios.items(): + with self.subTest(case=key, template=values["template"]): + cloudformation = CloudformationTemplate( + values["template"], parameters={"Environment": "development"} + ) + value = IntrinsicFunctions.eval(values["function"], cloudformation.template) + + self.assertEqual(value, values["expected"]) + + def test_join_function(self): + scenarios = { + "Resolved Function": { + "template": "tests/fixtures/templates/example2.yml", + "function": { + "Fn::Join": [ + ".", + ["fixtures", "handlers", "lambda_handler", {"Ref": "Handler"}], + ] + }, + "expected": "fixtures.handlers.lambda_handler.handler", + }, + "Unresolved Function with Incorrect Reference": { + "template": "tests/fixtures/templates/example2.yml", + "function": { + "Fn::Join": [ + ".", + ["fixtures", "handlers", "lambda_handler", {"Ref": "fixture"}], + ] + }, + "expected": None, + }, + } + + for key, values in scenarios.items(): + with self.subTest(case=key, template=values["template"]): + cloudformation = CloudformationTemplate( + values["template"], parameters={"Environment": "development"} + ) + value = IntrinsicFunctions.eval(values["function"], cloudformation.template) + + self.assertEqual(value, values["expected"]) + + def test_select_function(self): + scenarios = { + "Resolved Function with Correct Index": { + "template": "tests/fixtures/templates/example4.yml", + "function": { + "Fn::Select": [ + {"Fn::FindInMap": ["Environments", {"Ref": "Environment"}, "Index"]}, + ["handler1", {"Ref": "Handler"}], + ] + }, + "expected": "handler2", + }, + "Unresolved Function with Incorrect Reference in List": { + "template": "tests/fixtures/templates/example4.yml", + "function": { + "Fn::Select": [ + {"Fn::FindInMap": ["Environments", {"Ref": "Environment"}, "Index"]}, + ["handler1", {"Ref": "Fixture"}], + ] + }, + "expected": None, + }, + "Resolved Function with Correct Reference in List": { + "template": "tests/fixtures/templates/example4.yml", + "function": { + "Fn::Select": [ + {"Fn::FindInMap": ["Environments", {"Ref": "Environment"}, "Index"]}, + {"Ref": "HandlerList"}, + ] + }, + "expected": "handler2", + }, + "Unresolved Function with Incorrect Index Reference": { + "template": "tests/fixtures/templates/example4.yml", + "function": {"Fn::Select": [{"Ref": "index"}, {"Ref": "HandlerList"}]}, + "expected": None, + }, + "Unresolved Function with Incorrect Reference in Map": { + "template": "tests/fixtures/templates/example4.yml", + "function": { + "Fn::Select": [ + {"Fn::FindInMap": ["Environments", {"Ref": "Environment"}, "Index"]}, + {"Ref": "Fixture"}, + ] + }, + "expected": None, + }, + } + + for key, values in scenarios.items(): + with self.subTest(case=key, template=values["template"]): + cloudformation = CloudformationTemplate( + values["template"], parameters={"Environment": "development"} + ) + value = IntrinsicFunctions.eval(values["function"], cloudformation.template) + + self.assertEqual(value, values["expected"]) + + def test_split_function(self): + scenarios = { + "Resolved Function": { + "template": "tests/fixtures/templates/example3.yml", + "function": {"Fn::Split": ["|", "dev@gmail|dotz@gmail.com"]}, + "expected": ["dev@gmail", "dotz@gmail.com"], + }, + "Resolved Function with Reference": { + "template": "tests/fixtures/templates/example3.yml", + "function": {"Fn::Split": ["|", {"Ref": "Accounts"}]}, + "expected": ["dev@gmail", "dotz@gmail.com"], + }, + "Unresolved Function with Incorrect Reference": { + "template": "tests/fixtures/templates/example3.yml", + "function": {"Fn::Split": ["|", {"Ref": "AccountList"}]}, + "expected": None, + }, + } + + for key, values in scenarios.items(): + with self.subTest(case=key, template=values["template"]): + cloudformation = CloudformationTemplate( + values["template"], parameters={"Environment": "development"} + ) + value = IntrinsicFunctions.eval(values["function"], cloudformation.template) + + self.assertEqual(value, values["expected"]) + + @mock.patch.dict( + "os.environ", + { + "AWS_AccountId": "123456789", + "ENVIRONMENT": "dev", + }, + ) + def test_replace_placeholders(self): + scenarios = [ + { + "name": "with env vars", + "string": "account-${AWS::AccountId}-${ENVIRONMENT}", + "matches": ["AWS::AccountId", "ENVIRONMENT"], + "expected_result": "account-123456789-dev", + }, + { + "name": "missing env vars", + "string": "account-${Id}", + "matches": ["Id"], + "expected_result": None, + }, + { + "name": "no matches", + "string": "account-${AWS::AccountId}-${ENVIRONMENT}", + "matches": ["OTHER_VAR"], + "expected_result": None, + }, + ] + + for scenario in scenarios: + with self.subTest(case=scenario["name"]): + result = IntrinsicFunctions.replace_placeholders( + scenario["string"], scenario["matches"] + ) + self.assertEqual(result, scenario["expected_result"]) + + @mock.patch.dict("os.environ", {"account": "1234"}) + def test_sub_function(self): + scenarios = { + "Resolved Function with list of variables": { + "template": "tests/fixtures/templates/example4.yml", + "function": {"Fn::Sub": ["account-${account}", [{"account": {"Ref": "Account"}}]]}, + "expected": "account-1234", + }, + "Unresolved Function with nonexistent reference": { + "template": "tests/fixtures/templates/example4.yml", + "function": {"Fn::Sub": ["account-${account}", [{"account": {"Ref": "acc"}}]]}, + "expected": None, + }, + "Resolved Function with string value": { + "template": "tests/fixtures/templates/example4.yml", + "function": {"Fn::Sub": "account-${account}"}, + "expected": "account-1234", + }, + "Unresolved Function with nonexistent variable": { + "template": "tests/fixtures/templates/example4.yml", + "function": {"Fn::Sub": "account-${acc}"}, + "expected": None, + }, + "Unresolved Function with nonexistent variable name": { + "template": "tests/fixtures/templates/example4.yml", + "function": {"Fn::Sub": ["account-${account}", [{"value": {"Ref": "Account"}}]]}, + "expected": None, + }, + # "x": { + # "template": "tests/fixtures/templates/example4.yml", + # "function": {"Fn::Sub": + # ["account-${account}", + # [!Ref VarName: {"Ref": "Account"}}]] + # }, + # "expected": "account-1234", + # }, + } + + for key, values in scenarios.items(): + with self.subTest(case=key, template=values["template"]): + cloudformation = CloudformationTemplate( + values["template"], parameters={"Environment": "development"} + ) + value = IntrinsicFunctions.eval(values["function"], cloudformation.template) + + self.assertEqual(value, values["expected"]) + + class TestResource(unittest.TestCase): def test_resource(self): resource_id = "Test"