diff --git a/samcli/commands/list/cli_common/options.py b/samcli/commands/list/cli_common/options.py index ecaadc97087..b214ec9da7e 100644 --- a/samcli/commands/list/cli_common/options.py +++ b/samcli/commands/list/cli_common/options.py @@ -24,6 +24,7 @@ def stack_name_option(f): def output_click_option(): return click.option( "--output", + default="table", help="Output the results from the command in a given " "output format (json or table). ", type=click.Choice(["json", "table"], case_sensitive=False), ) diff --git a/samcli/commands/list/table_consumer.py b/samcli/commands/list/table_consumer.py new file mode 100644 index 00000000000..cbbdd52f971 --- /dev/null +++ b/samcli/commands/list/table_consumer.py @@ -0,0 +1,42 @@ +""" +The table consumer for 'sam list' +""" +from typing import Dict, Any +from samcli.lib.list.list_interfaces import ListInfoPullerConsumer +from samcli.commands._utils.table_print import pprint_column_names, pprint_columns + + +class StringConsumerTableOutput(ListInfoPullerConsumer): + """ + Outputs data in table format + """ + + def consume(self, data: Dict[Any, Any]) -> None: + """ + Outputs the data in a table format + Parameters + ---------- + data: Dict[Any, Any] + The data to be outputted + """ + + @pprint_column_names( + format_string=data["format_string"], + format_kwargs=data["format_args"], + table_header=data["table_name"], + ) + def print_table_rows(**kwargs): + """ + Prints the rows of the table based on the data provided + """ + for entry in data["data"]: + pprint_columns( + columns=entry, + width=kwargs["width"], + margin=kwargs["margin"], + format_string=data["format_string"], + format_args=kwargs["format_args"], + columns_dict=data["format_args"].copy(), + ) + + print_table_rows() diff --git a/samcli/commands/list/testable_resources/cli.py b/samcli/commands/list/testable_resources/cli.py index 255d6c17e40..f00bdaf5bc1 100644 --- a/samcli/commands/list/testable_resources/cli.py +++ b/samcli/commands/list/testable_resources/cli.py @@ -8,6 +8,8 @@ from samcli.cli.main import pass_context, common_options, aws_creds_options, print_cmdline_args from samcli.lib.utils.version_checker import check_newer_version from samcli.lib.telemetry.metric import track_command +from samcli.commands._utils.options import template_option_without_build +from samcli.cli.cli_config_file import configuration_option, TomlProvider HELP_TEXT = """ @@ -19,22 +21,30 @@ @click.command(name="testable-resources", help=HELP_TEXT) +@configuration_option(provider=TomlProvider(section="parameters")) @stack_name_option @output_option +@template_option_without_build @aws_creds_options @common_options @pass_context @track_command @check_newer_version @print_cmdline_args -def cli(self, stack_name, output): +def cli(self, stack_name, output, template_file, config_file, config_env): """ `sam list testable-resources` command entry point """ - do_cli(stack_name=stack_name, output=output, region=self.region, profile=self.profile) + do_cli(stack_name=stack_name, output=output, region=self.region, profile=self.profile, template_file=template_file) -def do_cli(stack_name, output, region, profile): +def do_cli(stack_name, output, region, profile, template_file): """ Implementation of the ``cli`` method """ + from samcli.commands.list.testable_resources.testable_resources_context import TestableResourcesContext + + with TestableResourcesContext( + stack_name=stack_name, output=output, region=region, profile=profile, template_file=template_file + ) as testable_resources_context: + testable_resources_context.run() diff --git a/samcli/commands/list/testable_resources/testable_resources_context.py b/samcli/commands/list/testable_resources/testable_resources_context.py new file mode 100644 index 00000000000..df2ca7b8e6f --- /dev/null +++ b/samcli/commands/list/testable_resources/testable_resources_context.py @@ -0,0 +1,84 @@ +""" +Display of the Testable Resources of a SAM stack +""" +import logging +from typing import Optional + +from samcli.commands.list.cli_common.list_common_context import ListContext +from samcli.lib.list.testable_resources.testable_resources_producer import TestableResourcesProducer +from samcli.lib.list.mapper_consumer_factory import MapperConsumerFactory +from samcli.lib.list.list_interfaces import ProducersEnum + +LOG = logging.getLogger(__name__) + + +class TestableResourcesContext(ListContext): + """ + Context class for testable resources + """ + + def __init__( + self, stack_name: str, output: str, region: Optional[str], profile: Optional[str], template_file: Optional[str] + ): + """ + Parameters + ---------- + stack_name: str + The name of the stack + output: str + The format of the output, either json or table + region: Optional[str] + The region of the stack + profile: Optional[str] + Optional profile to be used + template_file: Optional[str] + The location of the template file. If one is not specified, the default will be "template.yaml" in the CWD + """ + super().__init__() + self.stack_name = stack_name + self.output = output + self.region = region + self.profile = profile + self.template_file = template_file + self.iam_client = None + self.cloudcontrol_client = None + self.apigateway_client = None + self.apigatewayv2_client = None + + def __enter__(self): + self.init_clients() + return self + + def __exit__(self, *args): + pass + + def init_clients(self) -> None: + """ + Initialize the clients being used by sam list. + """ + super().init_clients() + self.iam_client = self.client_provider("iam") + self.cloudcontrol_client = self.client_provider("cloudcontrol") + self.apigateway_client = self.client_provider("apigateway") + self.apigatewayv2_client = self.client_provider("apigatewayv2") + + def run(self) -> None: + """ + Get the resources for a stack + """ + factory = MapperConsumerFactory() + container = factory.create(producer=ProducersEnum.TESTABLE_RESOURCES_PRODUCER, output=self.output) + testable_resource_producer = TestableResourcesProducer( + stack_name=self.stack_name, + region=self.region, + profile=self.profile, + template_file=self.template_file, + cloudformation_client=self.cloudformation_client, + iam_client=self.iam_client, + cloudcontrol_client=self.cloudcontrol_client, + apigateway_client=self.apigateway_client, + apigatewayv2_client=self.apigatewayv2_client, + mapper=container.mapper, + consumer=container.consumer, + ) + testable_resource_producer.produce() diff --git a/samcli/lib/list/mapper_consumer_factory.py b/samcli/lib/list/mapper_consumer_factory.py index 4bfa137d6e6..565a023a3f7 100644 --- a/samcli/lib/list/mapper_consumer_factory.py +++ b/samcli/lib/list/mapper_consumer_factory.py @@ -4,14 +4,48 @@ from samcli.lib.list.list_interfaces import MapperConsumerFactoryInterface from samcli.lib.list.data_to_json_mapper import DataToJsonMapper from samcli.commands.list.json_consumer import StringConsumerJsonOutput +from samcli.commands.list.table_consumer import StringConsumerTableOutput from samcli.lib.list.mapper_consumer_container import MapperConsumerContainer -from samcli.lib.list.list_interfaces import ProducersEnum +from samcli.lib.list.stack_outputs.stack_output_to_table_mapper import StackOutputToTableMapper +from samcli.lib.list.resources.resources_to_table_mapper import ResourcesToTableMapper +from samcli.lib.list.testable_resources.testable_resources_to_table_mapper import TestableResourcesToTableMapper +from samcli.lib.list.list_interfaces import ProducersEnum, Mapper class MapperConsumerFactory(MapperConsumerFactoryInterface): + """ + Factory class to create factory objects that map a given producer and output format to a mapper and a consumer + """ + def create(self, producer: ProducersEnum, output: str) -> MapperConsumerContainer: - # Will add conditions here to return different sorts of containers later on - data_to_json_mapper = DataToJsonMapper() - json_consumer = StringConsumerJsonOutput() - container = MapperConsumerContainer(data_to_json_mapper, json_consumer) + """ + Creates a MapperConsumerContainer that contains the resulting mapper and consumer given + the producer and output format + + Parameters + ---------- + producer: ProducersEnum + An enum representing the producers (stack-outputs, resources, or testable-resources producer) + output: str + The output format, either json or table + + Returns + ------- + container: MapperConsumerContainer + A MapperConsumerContainer containing the resulting mapper and consumer to be used by the producer + """ + if output == "json": + data_to_json_mapper = DataToJsonMapper() + json_consumer = StringConsumerJsonOutput() + container = MapperConsumerContainer(data_to_json_mapper, json_consumer) + return container + table_mapper: Mapper + table_consumer = StringConsumerTableOutput() + if producer == ProducersEnum.STACK_OUTPUTS_PRODUCER: + table_mapper = StackOutputToTableMapper() + elif producer == ProducersEnum.RESOURCES_PRODUCER: + table_mapper = ResourcesToTableMapper() + elif producer == ProducersEnum.TESTABLE_RESOURCES_PRODUCER: + table_mapper = TestableResourcesToTableMapper() + container = MapperConsumerContainer(table_mapper, table_consumer) return container diff --git a/samcli/lib/list/resources/resources_to_table_mapper.py b/samcli/lib/list/resources/resources_to_table_mapper.py new file mode 100644 index 00000000000..4cc209809b9 --- /dev/null +++ b/samcli/lib/list/resources/resources_to_table_mapper.py @@ -0,0 +1,48 @@ +""" +Implementation of the resources to table mapper +""" +from typing import Dict, Any +from collections import OrderedDict +from samcli.lib.list.list_interfaces import Mapper + + +class ResourcesToTableMapper(Mapper): + """ + Mapper class for mapping resources data for table output + """ + + def map(self, data: list) -> Dict[Any, Any]: + """ + Maps data to the format needed for consumption by the table consumer + + Parameters + ---------- + data: list + List of dictionaries containing the entries of the resources data + + Returns + ------- + table_data: Dict[Any, Any] + Dictionary containing the information and data needed for the table + consumer to output the data in table format + """ + entry_list = [] + for resource in data: + entry_list.append( + [ + resource.get("LogicalResourceId", "-"), + resource.get("PhysicalResourceId", "-"), + ] + ) + table_data = { + "format_string": "{Logical ID:<{0}} {Physical ID:<{1}}", + "format_args": OrderedDict( + { + "Logical ID": "Logical ID", + "Physical ID": "Physical ID", + } + ), + "table_name": "Resources", + "data": entry_list, + } + return table_data diff --git a/samcli/lib/list/stack_outputs/stack_output_to_table_mapper.py b/samcli/lib/list/stack_outputs/stack_output_to_table_mapper.py new file mode 100644 index 00000000000..3e0cc846f65 --- /dev/null +++ b/samcli/lib/list/stack_outputs/stack_output_to_table_mapper.py @@ -0,0 +1,46 @@ +""" +Implementation of the stack output to table mapper +""" +from typing import Dict, Any +from collections import OrderedDict +from samcli.lib.list.list_interfaces import Mapper + + +class StackOutputToTableMapper(Mapper): + """ + Mapper class for mapping stack-outputs data for table output + """ + + def map(self, data: list) -> Dict[Any, Any]: + """ + Maps data to the format needed for consumption by the table consumer + + Parameters + ---------- + data: list + List of dictionaries containing the entries of the stack outputs data + + Returns + ------- + table_data: Dict[Any, Any] + Dictionary containing the information and data needed for the table consumer + to output the data in table format + """ + entry_list = [] + for stack_output in data: + entry_list.append( + [ + stack_output.get("OutputKey", "-"), + stack_output.get("OutputValue", "-"), + stack_output.get("Description", "-"), + ] + ) + table_data = { + "format_string": "{OutputKey:<{0}} {OutputValue:<{1}} {Description:<{2}}", + "format_args": OrderedDict( + {"OutputKey": "OutputKey", "OutputValue": "OutputValue", "Description": "Description"} + ), + "table_name": "Stack Outputs", + "data": entry_list, + } + return table_data diff --git a/samcli/lib/list/stack_outputs/stack_outputs_producer.py b/samcli/lib/list/stack_outputs/stack_outputs_producer.py index 888df6975ca..a97b35856e3 100644 --- a/samcli/lib/list/stack_outputs/stack_outputs_producer.py +++ b/samcli/lib/list/stack_outputs/stack_outputs_producer.py @@ -58,11 +58,13 @@ def get_stack_info(self) -> Optional[Any]: def produce(self): response = self.get_stack_info() + output_list = [] for stack_output in response: stack_output_data = StackOutputs( OutputKey=stack_output["OutputKey"], OutputValue=stack_output["OutputValue"], Description=stack_output["Description"], ) - mapped_output = self.mapper.map(dataclasses.asdict(stack_output_data)) - self.consumer.consume(mapped_output) + output_list.append(dataclasses.asdict(stack_output_data)) + mapped_output = self.mapper.map(output_list) + self.consumer.consume(data=mapped_output) diff --git a/samcli/lib/list/testable_resources/__init__.py b/samcli/lib/list/testable_resources/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/samcli/lib/list/testable_resources/testable_res_def.py b/samcli/lib/list/testable_resources/testable_res_def.py new file mode 100644 index 00000000000..0d89e632e9d --- /dev/null +++ b/samcli/lib/list/testable_resources/testable_res_def.py @@ -0,0 +1,17 @@ +""" +The container for Testable Resources +""" +from typing import Any +from dataclasses import dataclass + + +@dataclass +class TestableResDef: + """ + Dataclass for containing entries of testable resources data + """ + + LogicalResourceId: str + PhysicalResourceId: str + CloudEndpointOrFunctionURL: Any + Methods: Any diff --git a/samcli/lib/list/testable_resources/testable_resources_producer.py b/samcli/lib/list/testable_resources/testable_resources_producer.py new file mode 100644 index 00000000000..be2a6441d48 --- /dev/null +++ b/samcli/lib/list/testable_resources/testable_resources_producer.py @@ -0,0 +1,472 @@ +""" +The producer for the 'sam list testable-resources' command +""" +import dataclasses +import logging +from typing import Dict, List, Any +from enum import Enum +import json +from botocore.exceptions import ClientError, BotoCoreError +from samcli.commands.list.exceptions import ( + SamListUnknownBotoCoreError, + SamListLocalResourcesNotFoundError, + SamListUnknownClientError, +) +from samcli.lib.list.list_interfaces import Producer +from samcli.lib.providers.sam_stack_provider import SamLocalStackProvider +from samcli.lib.providers.provider import Stack +from samcli.commands._utils.template import get_template_data +from samcli.lib.list.testable_resources.testable_res_def import TestableResDef +from samcli.lib.list.resources.resource_mapping_producer import ResourceMappingProducer +from samcli.lib.utils.boto_utils import get_client_error_code + +LOG = logging.getLogger(__name__) +TESTABLE_RESOURCE_TYPES = {"AWS::Lambda::Function", "AWS::ApiGateway::RestApi", "AWS::ApiGatewayV2::Api"} +RESOURCE_DESCRIPTION = "ResourceDescription" +PROPERTIES = "Properties" +FUNCTION_URL = "FunctionUrl" +STACK_RESOURCES = "StackResources" +RESOURCE_TYPE = "ResourceType" +PHYSICAL_RESOURCE_ID = "PhysicalResourceId" +LOGICAL_RESOURCE_ID = "LogicalResourceId" +REST_API_ID = "RestApiId" +API_ID = "ApiId" +DOMAIN_NAME = "DomainName" +BODY = "Body" +PATHS = "paths" + + +class APIGatewayEnum(Enum): + API_GATEWAY = 1 + API_GATEWAY_V2 = 2 + + +class TestableResourcesProducer(ResourceMappingProducer, Producer): + def __init__( + self, + stack_name, + region, + profile, + template_file, + cloudformation_client, + iam_client, + cloudcontrol_client, + apigateway_client, + apigatewayv2_client, + mapper, + consumer, + ): + """ + Parameters + ---------- + stack_name: str + The name of the stack + region: Optional[str] + The region of the stack + profile: Optional[str] + Optional profile to be used + template_file: Optional[str] + The location of the template file. If one is not specified, the default will be "template.yaml" in the CWD + cloudformation_client: CloudFormation + The CloudFormation client + iam_client: IAM + The IAM client + cloudcontrol_client: CloudControl + The CloudControl client + apigateway_client: APIGateway + The APIGateway client + apigatewayv2_client: APIGatewayV2 + The APIGatewayV2 client + mapper: Mapper + The mapper used to map data to the format needed for the consumer provided + consumer: ListInfoPullerConsumer + The consumer used to output the data + """ + super().__init__( + stack_name, region, profile, template_file, cloudformation_client, iam_client, mapper, consumer + ) + self.stack_name = stack_name + self.region = region + self.profile = profile + self.template_file = template_file + self.cloudformation_client = cloudformation_client + self.iam_client = iam_client + self.cloudcontrol_client = cloudcontrol_client + self.apigateway_client = apigateway_client + self.apigatewayv2_client = apigatewayv2_client + self.mapper = mapper + self.consumer = consumer + + def get_function_url(self, identifier: str) -> Any: + """ + Gets the function url of a Lambda Function + + Parameters + ---------- + identifier: str + The identifier or physical ID + + Returns + ------- + furl: str + The function url in the form of a string + """ + try: + response = self.cloudcontrol_client.get_resource(TypeName="AWS::Lambda::Url", Identifier=identifier) + if not response.get(RESOURCE_DESCRIPTION, {}).get(PROPERTIES, {}): + return "-" + response_dict = json.loads(response.get(RESOURCE_DESCRIPTION, {}).get(PROPERTIES, {})) + furl = response_dict.get(FUNCTION_URL, "-") + return furl + except ClientError as e: + if get_client_error_code(e) == "ResourceNotFoundException": + return "-" + LOG.error("ClientError Exception : %s", str(e)) + raise SamListUnknownClientError(msg=str(e)) from e + + def get_stage_list(self, api_id: str, api_type: APIGatewayEnum) -> List[Any]: + """ + Gets a list of stages for a given api of type AWS::ApiGateway::RestApi or AWS::ApiGatewayV2::Api + + Parameters + ---------- + api_id: str + The api id or rest api id of the api + api_type: APIGatewayEnum + The type of api, AWS::ApiGateway::RestApi or AWS::ApiGatewayV2::Api + + Returns + ------- + response_list: List[Any] + A list of stages for the api + """ + response_list: List[Any] + try: + response_list = [] + response: dict + search_key: str + stage_name_key: str + if api_type == APIGatewayEnum.API_GATEWAY: + response = self.apigateway_client.get_stages(restApiId=api_id) + search_key = "item" + stage_name_key = "stageName" + elif api_type == APIGatewayEnum.API_GATEWAY_V2: + response = self.apigatewayv2_client.get_stages(ApiId=api_id) + search_key = "Items" + stage_name_key = "StageName" + if not response.get(search_key, []): + return response_list + for item in response.get(search_key, []): + if item.get(stage_name_key, None): + response_list.append(item.get(stage_name_key, "")) + return response_list + except ClientError as e: + if get_client_error_code(e) == "NotFoundException": + return [] + LOG.error("ClientError Exception : %s", str(e)) + raise SamListUnknownClientError(msg=str(e)) from e + except BotoCoreError as e: + LOG.error("Botocore Exception : %s", str(e)) + raise SamListUnknownBotoCoreError(msg=str(e)) from e + + def build_api_gw_endpoints(self, physical_id: str, stages: list) -> list: + """ + Builds the default api gateway endpoints + + Parameters + ---------- + physical_id: str + The physical ID of the api resource + stages: list + A list of stages for the api resource + + Returns + ------- + api_list: List[Any] + The list of default api gateway endpoints + """ + api_list = [] + for stage in stages: + + api_list.append(f"https://{physical_id}.execute-api.{self.region}.amazonaws.com/{stage}") + return api_list + + def get_api_gateway_endpoint( + self, deployed_resource: Dict[Any, Any], custom_domain_substitute_dict: Dict[Any, Any] + ) -> Any: + """ + Gets the API gateway endpoints for APIGateway and APIGatewayV2 APIs + + Parameters + ---------- + deployed_resource: Dict[Any, Any] + Dictionary containing the resource info of the deployed API + custom_domain_substitute_dict: Dict[Any, Any] + Dictionary containing the mappings of the custom domains for APIs + + Returns + ------- + endpoint: Any + The endpoint(s) of the current API resource + """ + endpoint: Any + stages = self.get_stage_list( + deployed_resource.get(PHYSICAL_RESOURCE_ID, ""), + get_api_type_enum(deployed_resource.get(RESOURCE_TYPE, "")), + ) + if deployed_resource.get(LOGICAL_RESOURCE_ID, "") in custom_domain_substitute_dict: + endpoint = custom_domain_substitute_dict.get(deployed_resource.get(LOGICAL_RESOURCE_ID, ""), "-") + else: + endpoint = self.build_api_gw_endpoints(deployed_resource.get(PHYSICAL_RESOURCE_ID, ""), stages) + return endpoint + + def get_cloud_testable_resources(self, stacks: list) -> list: + """ + Gets a list of cloud testable resources + + Parameters + ---------- + stacks: list + A list containing the local stack + + Returns + ------- + testable_resources_list: List[Any] + A list of cloud testable resources + """ + testable_resources_list = [] + local_stack = stacks[0] + local_stack_resources = local_stack.resources + seen_testable_resources = set() + response = self.get_resources_info() + response_domain_dict = get_response_domain_dict(response) + custom_domain_substitute_dict = get_custom_domain_substitute_list(response, stacks, response_domain_dict) + + for deployed_resource in response.get(STACK_RESOURCES, {}): + if deployed_resource.get(RESOURCE_TYPE, "") in TESTABLE_RESOURCE_TYPES: + endpoint_function_url: Any + paths_and_methods: Any + endpoint_function_url = "-" + paths_and_methods = "-" + if deployed_resource.get(RESOURCE_TYPE, "") == "AWS::Lambda::Function": + endpoint_function_url = self.get_function_url(deployed_resource.get(PHYSICAL_RESOURCE_ID, "")) + + elif deployed_resource.get(RESOURCE_TYPE, "") in ("AWS::ApiGateway::RestApi", "AWS::ApiGatewayV2::Api"): + endpoint_function_url = self.get_api_gateway_endpoint( + deployed_resource, custom_domain_substitute_dict + ) + paths_and_methods = get_methods_and_paths( + deployed_resource.get(LOGICAL_RESOURCE_ID, ""), local_stack + ) + + testable_resource_data = TestableResDef( + LogicalResourceId=deployed_resource.get(LOGICAL_RESOURCE_ID, "-"), + PhysicalResourceId=deployed_resource.get(PHYSICAL_RESOURCE_ID, "-"), + CloudEndpointOrFunctionURL=endpoint_function_url, + Methods=paths_and_methods, + ) + testable_resources_list.append(dataclasses.asdict(testable_resource_data)) + seen_testable_resources.add(deployed_resource.get(LOGICAL_RESOURCE_ID, "")) + for local_resource in local_stack_resources: + local_resource_type = local_stack_resources.get(local_resource, {}).get("Type", "") + paths_and_methods = "-" + if local_resource_type in TESTABLE_RESOURCE_TYPES and local_resource not in seen_testable_resources: + if local_resource_type in ("AWS::ApiGateway::RestApi", "AWS::ApiGatewayV2::Api"): + paths_and_methods = get_methods_and_paths(local_resource, local_stack) + testable_resource_data = TestableResDef( + LogicalResourceId=local_resource, + PhysicalResourceId="-", + CloudEndpointOrFunctionURL="-", + Methods=paths_and_methods, + ) + testable_resources_list.append(dataclasses.asdict(testable_resource_data)) + + return testable_resources_list + + def produce(self): + """ + The producer function for the testable resources command + """ + sam_template = get_template_data(self.template_file) + + translated_dict = self.get_translated_dict(template_file_dict=sam_template) + stacks, _ = SamLocalStackProvider.get_stacks(template_file="", template_dictionary=translated_dict) + validate_stack(stacks) + + testable_resources_list: list + + if self.stack_name: + testable_resources_list = self.get_cloud_testable_resources(stacks) + else: + testable_resources_list = get_local_testable_resources(stacks) + mapped_output = self.mapper.map(testable_resources_list) + self.consumer.consume(mapped_output) + + +def validate_stack(stacks: list): + """ + Checks if the stack non-empty and contains stack resources and raises exceptions accordingly + + Parameters + ---------- + stacks: list + A list containing the stack + """ + + if not stacks or not hasattr(stacks[0], "resources") or not stacks[0].resources: + raise SamListLocalResourcesNotFoundError(msg="No local resources found.") + + +def get_local_testable_resources(stacks: list) -> list: + """ + Gets a list of local testable resources based on the local stack + + Parameters + ---------- + stacks: list + A list containing the stack + + Returns + ------- + testable_resources_list: list + A list containing the testable resources and their information + """ + testable_resources_list = [] + paths_and_methods: Any + local_stack = stacks[0] + local_stack_resources = local_stack.resources + for local_resource in local_stack_resources: + local_resource_type = local_stack_resources.get(local_resource, {}).get("Type", "") + if local_resource_type in TESTABLE_RESOURCE_TYPES: + paths_and_methods = "-" + if local_resource_type in ("AWS::ApiGateway::RestApi", "AWS::ApiGatewayV2::Api"): + paths_and_methods = get_methods_and_paths(local_resource, local_stack) + # Set the PhysicalID to "-" if there is no corresponding PhysicalID + testable_resource_data = TestableResDef( + LogicalResourceId=local_resource, + PhysicalResourceId="-", + CloudEndpointOrFunctionURL="-", + Methods=paths_and_methods, + ) + testable_resources_list.append(dataclasses.asdict(testable_resource_data)) + return testable_resources_list + + +def get_api_type_enum(resource_type: str) -> APIGatewayEnum: + """ + Gets the APIGatewayEnum associated with the input resource type + + Parameters + ---------- + resource_type: str + The type of the resource + + Returns + ------- + The APIGatewayEnum associated with the input resource type + """ + if resource_type == "AWS::ApiGatewayV2::Api": + return APIGatewayEnum.API_GATEWAY_V2 + return APIGatewayEnum.API_GATEWAY + + +def get_custom_domain_substitute_list( + response: Dict[Any, Any], stacks: list, response_domain_dict: Dict[str, str] +) -> Dict[Any, Any]: + """ + Gets a dictionary containing the custom domain lists that map back to the original api + + Parameters + ---------- + response: Dict[Any, Any] + The response containing the cloud stack resources information + stacks: list + A list containing the local stack + response_domain_dict: Dict + A dictionary containing the custom domains + Returns + ------- + custom_domain_substitute_dict: Dict[Any, Any] + A dict containing the custom domain lists mapped to the original apis + """ + custom_domain_substitute_dict = {} + local_stack = stacks[0] + local_stack_resources = local_stack.resources + for resource in response.get(STACK_RESOURCES, {}): + if resource.get(RESOURCE_TYPE, "") == "AWS::ApiGateway::BasePathMapping": + local_mapping = local_stack_resources.get(resource.get(LOGICAL_RESOURCE_ID, ""), {}).get(PROPERTIES, {}) + rest_api_id = local_mapping.get(REST_API_ID, "") + domain_id = local_mapping.get(DOMAIN_NAME, "") + if domain_id in response_domain_dict: + if rest_api_id not in custom_domain_substitute_dict: + custom_domain_substitute_dict[rest_api_id] = [response_domain_dict.get(domain_id, None)] + else: + custom_domain_substitute_dict[rest_api_id].append(response_domain_dict.get(domain_id, None)) + elif resource.get(RESOURCE_TYPE, "") == "AWS::ApiGatewayV2::ApiMapping": + local_mapping = local_stack_resources.get(resource.get(LOGICAL_RESOURCE_ID, ""), {}).get(PROPERTIES, {}) + rest_api_id = local_mapping.get(API_ID, "") + domain_id = local_mapping.get(DOMAIN_NAME, "") + if domain_id in response_domain_dict: + if rest_api_id not in custom_domain_substitute_dict: + custom_domain_substitute_dict[rest_api_id] = [response_domain_dict.get(domain_id, None)] + else: + custom_domain_substitute_dict[rest_api_id].append(response_domain_dict.get(domain_id, None)) + return custom_domain_substitute_dict + + +def get_response_domain_dict(response: Dict[Any, Any]) -> Dict[str, str]: + """ + Gets a dictionary containing the custom domains + + Parameters + ---------- + response: Dict[Any, Any] + The response containing the cloud stack resources information + + Returns + ------- + response_domain_dict: Dict[str, str] + A dict containing the custom domains + """ + response_domain_dict = {} + for resource in response.get(STACK_RESOURCES, {}): + if ( + resource.get(RESOURCE_TYPE, "") == "AWS::ApiGateway::DomainName" + or resource.get(RESOURCE_TYPE, "") == "AWS::ApiGatewayV2::DomainName" + ): + response_domain_dict[ + resource.get(LOGICAL_RESOURCE_ID, "") + ] = f'https://{resource.get(PHYSICAL_RESOURCE_ID, "")}' + return response_domain_dict + + +def get_methods_and_paths(logical_id: str, stack: Stack) -> list: + """ + Gets the methods and paths for apis based on the stack and the logical ID + + Parameters + ---------- + logical_id: str + The logical ID of the api + stack: Stack + The stack to retrieve the methods and paths from + + Returns + ------- + method_paths_list: list + A list containing the methods and paths of the api + """ + method_paths_list: List[Any] + method_paths_list = [] + if not stack.resources: + raise SamListLocalResourcesNotFoundError(msg="No local resources found.") + if not stack.resources.get(logical_id, {}).get(PROPERTIES, {}).get(BODY, {}).get(PATHS, {}): + return method_paths_list + paths_dict = stack.resources.get(logical_id, {}).get(PROPERTIES, {}).get(BODY, {}).get(PATHS, {}) + for path in paths_dict: + method_list = [] + for method in paths_dict.get(path, ""): + method_list.append(method) + path_item = path + f"{method_list}" + method_paths_list.append(path_item) + return method_paths_list diff --git a/samcli/lib/list/testable_resources/testable_resources_to_table_mapper.py b/samcli/lib/list/testable_resources/testable_resources_to_table_mapper.py new file mode 100644 index 00000000000..e55ce950a51 --- /dev/null +++ b/samcli/lib/list/testable_resources/testable_resources_to_table_mapper.py @@ -0,0 +1,67 @@ +""" +Implementation of the testable resources to table mapper +""" +from typing import Dict, Any +from collections import OrderedDict +from samcli.lib.list.list_interfaces import Mapper + + +class TestableResourcesToTableMapper(Mapper): + """ + Mapper class for mapping testable-resources data for table output + """ + + def map(self, data: list) -> Dict[Any, Any]: + """ + Maps data to the format needed for consumption by the table consumer + + Parameters + ---------- + data: list + List of dictionaries containing the entries of the testable resources data + + Returns + ------- + table_data: Dict[Any, Any] + Dictionary containing the information and data needed for the table consumer + to output the data in table format + """ + entry_list = [] + for testable_resource in data: + cloud_endpoint_furl_string = testable_resource.get("CloudEndpointOrFunctionURL", "-") + methods_string = "-" + cloud_endpoint_furl_multi_list = [] + if isinstance(testable_resource.get("CloudEndpointOrFunctionURL", "-"), list) and testable_resource.get( + "CloudEndpointOrFunctionURL", [] + ): + cloud_endpoint_furl_string = testable_resource.get("CloudEndpointOrFunctionURL", ["-"])[0] + if len(testable_resource.get("CloudEndpointOrFunctionURL", [])) > 1: + cloud_endpoint_furl_multi_list = testable_resource.get("CloudEndpointOrFunctionURL", ["", ""])[1:] + if isinstance(testable_resource.get("Methods", "-"), list) and testable_resource.get("Methods", []): + methods_string = "; ".join(testable_resource.get("Methods", [])) + + entry_list.append( + [ + testable_resource.get("LogicalResourceId", "-"), + testable_resource.get("PhysicalResourceId", "-"), + cloud_endpoint_furl_string, + methods_string, + ] + ) + if cloud_endpoint_furl_multi_list: + for url in cloud_endpoint_furl_multi_list: + entry_list.append(["", "", url, ""]) + table_data = { + "format_string": "{Resource ID:<{0}} {Physical ID:<{1}} {Cloud Endpoint/FURL:<{2}} {Methods:<{3}}", + "format_args": OrderedDict( + { + "Resource ID": "Resource ID", + "Physical ID": "Physical ID", + "Cloud Endpoint/FURL": "Cloud Endpoint/Function URL", + "Methods": "Methods", + } + ), + "table_name": "Testable Resources", + "data": entry_list, + } + return table_data diff --git a/tests/integration/list/resources/test_resources_command.py b/tests/integration/list/resources/test_resources_command.py index 63f019213b1..df7eeef09a5 100644 --- a/tests/integration/list/resources/test_resources_command.py +++ b/tests/integration/list/resources/test_resources_command.py @@ -9,12 +9,12 @@ from tests.testing_utils import RUNNING_ON_CI, RUNNING_TEST_FOR_MASTER_ON_CI, RUN_BY_CANARY from tests.testing_utils import run_command, run_command_with_input, method_to_stack_name -SKIP_STACK_OUTPUTS_TESTS = RUNNING_ON_CI and RUNNING_TEST_FOR_MASTER_ON_CI and not RUN_BY_CANARY +SKIP_RESOURCES_TESTS = RUNNING_ON_CI and RUNNING_TEST_FOR_MASTER_ON_CI and not RUN_BY_CANARY CFN_SLEEP = 3 CFN_PYTHON_VERSION_SUFFIX = os.environ.get("PYTHON_VERSION", "0.0.0").replace(".", "-") -@skipIf(SKIP_STACK_OUTPUTS_TESTS, "Skip stack-outputs tests in CI/CD only") +@skipIf(SKIP_RESOURCES_TESTS, "Skip resources tests in CI/CD only") class TestResources(DeployIntegBase, ResourcesIntegBase): @classmethod def setUpClass(cls): diff --git a/tests/integration/list/stack_outputs/test_stack_outputs_command.py b/tests/integration/list/stack_outputs/test_stack_outputs_command.py index 4043a590f49..54016784d3d 100644 --- a/tests/integration/list/stack_outputs/test_stack_outputs_command.py +++ b/tests/integration/list/stack_outputs/test_stack_outputs_command.py @@ -56,30 +56,30 @@ def test_stack_output_exists(self): self.assertTrue( re.search( """{ - "OutputKey": "HelloWorldFunctionIamRole", - "OutputValue": "arn:aws:iam::.*:role/.*-HelloWorldFunctionRole\-.*", - "Description": "Implicit IAM Role created for Hello World function" -}""", + "OutputKey": "HelloWorldFunctionIamRole", + "OutputValue": "arn:aws:iam::.*:role/.*-HelloWorldFunctionRole\-.*", + "Description": "Implicit IAM Role created for Hello World function" + }""", command_result.stdout.decode(), ) ) self.assertTrue( re.search( - """{ - "OutputKey": "HelloWorldApi", - "OutputValue": "https://.*execute.*.amazonaws.com/Prod/hello/", - "Description": "API Gateway endpoint URL for Prod stage for Hello World function" -}""", + """ { + "OutputKey": "HelloWorldApi", + "OutputValue": "https://.*execute.*.amazonaws.com/Prod/hello/", + "Description": "API Gateway endpoint URL for Prod stage for Hello World function" + }""", command_result.stdout.decode(), ) ) self.assertTrue( re.search( - """{ - "OutputKey": "HelloWorldFunction", - "OutputValue": "arn:aws:lambda:.*:.*:function:.*-HelloWorldFunction\-.*", - "Description": "Hello World Lambda Function ARN" -}""", + """ { + "OutputKey": "HelloWorldFunction", + "OutputValue": "arn:aws:lambda:.*:.*:function:.*-HelloWorldFunction\-.*", + "Description": "Hello World Lambda Function ARN" + }""", command_result.stdout.decode(), ) ) diff --git a/tests/integration/list/testable_resources/test_testable_resources_command.py b/tests/integration/list/testable_resources/test_testable_resources_command.py index a1a13e5c336..aef9240839f 100644 --- a/tests/integration/list/testable_resources/test_testable_resources_command.py +++ b/tests/integration/list/testable_resources/test_testable_resources_command.py @@ -1,12 +1,135 @@ +import os +import time +import boto3 +import re +from unittest import skipIf +from tests.integration.deploy.deploy_integ_base import DeployIntegBase from tests.integration.list.testable_resources.testable_resources_integ_base import TestableResourcesIntegBase from samcli.commands.list.testable_resources.cli import HELP_TEXT from tests.testing_utils import run_command +from tests.testing_utils import RUNNING_ON_CI, RUNNING_TEST_FOR_MASTER_ON_CI, RUN_BY_CANARY +from tests.testing_utils import run_command, run_command_with_input, method_to_stack_name +SKIP_TESTABLE_RESOURCES_TESTS = RUNNING_ON_CI and RUNNING_TEST_FOR_MASTER_ON_CI and not RUN_BY_CANARY +CFN_SLEEP = 3 +CFN_PYTHON_VERSION_SUFFIX = os.environ.get("PYTHON_VERSION", "0.0.0").replace(".", "-") + + +@skipIf(SKIP_TESTABLE_RESOURCES_TESTS, "Skip testable resources tests in CI/CD only") +class TestTestableResources(DeployIntegBase, TestableResourcesIntegBase): + @classmethod + def setUpClass(cls): + DeployIntegBase.setUpClass() + TestableResourcesIntegBase.setUpClass() + + def setUp(self): + self.cf_client = boto3.client("cloudformation") + time.sleep(CFN_SLEEP) + super().setUp() -class TestTestableResources(TestableResourcesIntegBase): def test_testable_resources_help_message(self): cmdlist = self.get_testable_resources_command_list(help=True) command_result = run_command(cmdlist) from_command = "".join(command_result.stdout.decode().split()) from_help = "".join(HELP_TEXT.split()) self.assertIn(from_help, from_command, "Testable-resources help text should have been printed") + + def test_no_stack_name(self): + template_path = self.list_test_data_path.joinpath("test_testable_resources_template.yaml") + region = boto3.Session().region_name + cmdlist = self.get_testable_resources_command_list( + stack_name=None, region=region, output="json", template_file=template_path + ) + command_result = run_command(cmdlist, cwd=self.working_dir) + expected_output = [ + """{ + "LogicalResourceId": "HelloWorldFunction", + "PhysicalResourceId": "-", + "CloudEndpointOrFunctionURL": "-", + "Methods": "-" + }""", + """{ + "LogicalResourceId": "TestAPI", + "PhysicalResourceId": "-", + "CloudEndpointOrFunctionURL": "-", + "Methods": [] + }""", + """{ + "LogicalResourceId": "ServerlessRestApi", + "PhysicalResourceId": "-", + "CloudEndpointOrFunctionURL": "-", + "Methods": [ + "/hello2['get']", + "/hello['get']" + ] + }""", + ] + for expression in expected_output: + self.assertIn(expression, command_result.stdout.decode()) + + def test_has_stack_name(self): + template_path = self.list_test_data_path.joinpath("test_testable_resources_template.yaml") + stack_name = method_to_stack_name(self.id()) + config_file_name = stack_name + ".toml" + region = boto3.Session().region_name + deploy_command_list = self.get_deploy_command_list( + template_file=template_path, + guided=True, + config_file=config_file_name, + region=region, + confirm_changeset=True, + disable_rollback=True, + ) + deploy_process_execute = run_command_with_input( + deploy_command_list, "{}\n{}\nY\nY\nY\nY\nY\nY\n\n\nY\n".format(stack_name, region).encode() + ) + cmdlist = self.get_testable_resources_command_list( + stack_name=stack_name, region=region, output="json", template_file=template_path + ) + command_result = run_command(cmdlist, cwd=self.working_dir) + expected_output = [ + """{ + "LogicalResourceId": "HelloWorldFunction", + "PhysicalResourceId": "test-has-stack-name.*", + "CloudEndpointOrFunctionURL": "https://.*.lambda-url..*.on.aws/", + "Methods": "-" + }""", + """ { + "LogicalResourceId": "ServerlessRestApi", + "PhysicalResourceId": ".*", + "CloudEndpointOrFunctionURL": .* + "https://.*.execute-api..*.amazonaws.com/Prod", + "https://.*.execute-api..*.amazonaws.com/Stage" + .*, + "Methods": .* + "/hello2.'get'.", + "/hello.'get'." + . + }""", + """ { + "LogicalResourceId": "TestAPI", + "PhysicalResourceId": ".*", + "CloudEndpointOrFunctionURL": . + "https://.*.execute-api..*.amazonaws.com/Test2" + ., + "Methods": .. + }""", + ] + for expression in expected_output: + self.assertTrue(re.search(expression, command_result.stdout.decode())) + + def test_stack_does_not_exist(self): + template_path = self.list_test_data_path.joinpath("test_testable_resources_template.yaml") + stack_name = method_to_stack_name(self.id()) + config_file_name = stack_name + ".toml" + region = boto3.Session().region_name + cmdlist = self.get_testable_resources_command_list( + stack_name=stack_name, region=region, output="json", template_file=template_path + ) + command_result = run_command(cmdlist, cwd=self.working_dir) + expected_output = ( + f"Error: The input stack {stack_name} does" f" not exist on Cloudformation in the region {region}" + ) + self.assertIn( + expected_output, command_result.stderr.decode(), "Should have raised error that outputs do not exist" + ) diff --git a/tests/integration/list/testable_resources/testable_resources_integ_base.py b/tests/integration/list/testable_resources/testable_resources_integ_base.py index 2a050a31e2e..0b814841c3a 100644 --- a/tests/integration/list/testable_resources/testable_resources_integ_base.py +++ b/tests/integration/list/testable_resources/testable_resources_integ_base.py @@ -2,7 +2,9 @@ class TestableResourcesIntegBase(ListIntegBase): - def get_testable_resources_command_list(self, stack_name=None, output=None, region=None, profile=None, help=False): + def get_testable_resources_command_list( + self, stack_name=None, output=None, region=None, profile=None, template_file=None, help=False + ): command_list = [self.base_command(), "list", "testable-resources"] if stack_name: command_list += ["--stack-name", str(stack_name)] @@ -16,6 +18,9 @@ def get_testable_resources_command_list(self, stack_name=None, output=None, regi if profile: command_list += ["--profile", str(profile)] + if template_file: + command_list += ["--template-file", str(template_file)] + if help: command_list += ["--help"] diff --git a/tests/integration/testdata/list/test_testable_resources_template.yaml b/tests/integration/testdata/list/test_testable_resources_template.yaml new file mode 100644 index 00000000000..7d8a81fec70 --- /dev/null +++ b/tests/integration/testdata/list/test_testable_resources_template.yaml @@ -0,0 +1,44 @@ +AWSTemplateFormatVersion: '2010-09-09' +Transform: AWS::Serverless-2016-10-31 +Description: > + sam-app-hello + + Sample SAM Template for sam-app-hello + +# More info about Globals: https://github.com/awslabs/serverless-application-model/blob/master/docs/globals.rst +Globals: + Function: + Timeout: 3 + Tracing: Active + +Resources: + HelloWorldFunction: + Type: AWS::Serverless::Function # More info about Function Resource: https://github.com/awslabs/serverless-application-model/blob/master/versions/2016-10-31.md#awsserverlessfunction + Properties: + CodeUri: hello_world/ + Handler: app.lambda_handler + Runtime: python3.8 + FunctionUrlConfig: + AuthType: AWS_IAM + Architectures: + - x86_64 + Events: + HelloWorld: + Type: Api # More info about API Event Source: https://github.com/awslabs/serverless-application-model/blob/master/versions/2016-10-31.md#api + Properties: + Path: /hello + Method: get + HelloWorld2: + Type: Api # More info about API Event Source: https://github.com/awslabs/serverless-application-model/blob/master/versions/2016-10-31.md#api + Properties: + Path: /hello2 + Method: get + TestAPI: + Type: AWS::Serverless::HttpApi + Properties: + Description: "Test resources" + StageName: Test2 + + + + diff --git a/tests/unit/commands/list/stack_outputs/test_stack_outputs_context.py b/tests/unit/commands/list/stack_outputs/test_stack_outputs_context.py index 06021b7328e..721e1d9eeb1 100644 --- a/tests/unit/commands/list/stack_outputs/test_stack_outputs_context.py +++ b/tests/unit/commands/list/stack_outputs/test_stack_outputs_context.py @@ -23,7 +23,9 @@ def test_stack_outputs_stack_exists( stack_output_context.run() expected_click_echo_calls = [ - call('{\n "OutputKey": "HelloWorldTest",\n "OutputValue": "TestVal",\n "Description": "Test"\n}') + call( + '[\n {\n "OutputKey": "HelloWorldTest",\n "OutputValue": "TestVal",\n "Description": "Test"\n }\n]' + ) ] self.assertEqual( expected_click_echo_calls, patched_click_echo.call_args_list, "Stack and stack outputs should exist" diff --git a/tests/unit/commands/list/test_list_mappers.py b/tests/unit/commands/list/test_list_mappers.py new file mode 100644 index 00000000000..8d0550573f2 --- /dev/null +++ b/tests/unit/commands/list/test_list_mappers.py @@ -0,0 +1,105 @@ +from unittest import TestCase +from unittest.mock import patch, call +from collections import OrderedDict +from samcli.lib.list.resources.resources_to_table_mapper import ResourcesToTableMapper +from samcli.lib.list.stack_outputs.stack_output_to_table_mapper import StackOutputToTableMapper +from samcli.lib.list.data_to_json_mapper import DataToJsonMapper +from samcli.commands.list.json_consumer import StringConsumerJsonOutput +from samcli.lib.list.testable_resources.testable_resources_to_table_mapper import TestableResourcesToTableMapper +from samcli.lib.list.mapper_consumer_factory import MapperConsumerFactory +from samcli.lib.list.list_interfaces import ProducersEnum +from samcli.commands.list.table_consumer import StringConsumerTableOutput + + +class TestStackOutputsToTableMapper(TestCase): + def test_map(self): + data = [{"OutputKey": "outputkey1", "OutputValue": "outputvalue1", "Description": "sample description"}] + stack_outputs_to_table_mapper = StackOutputToTableMapper() + output = stack_outputs_to_table_mapper.map(data) + self.assertEqual(output.get("table_name", ""), "Stack Outputs") + + +class TestResourcesToTableMapper(TestCase): + def test_map(self): + data = [{"LogicalResourceId": "LID_1", "PhysicalResourceId": "PID_1"}] + resources_to_table_mapper = ResourcesToTableMapper() + output = resources_to_table_mapper.map(data) + self.assertEqual(output.get("table_name", ""), "Resources") + + +class TestTestableResourcesToTableMapper(TestCase): + def test_map(self): + data = [ + { + "LogicalResourceId": "LID_1", + "PhysicalResourceId": "PID_1", + "CloudEndpointOrFunctionURL": "test.url", + "Methods": "-", + }, + { + "LogicalResourceId": "LID_1", + "PhysicalResourceId": "PID_1", + "CloudEndpointOrFunctionURL": "-", + "Methods": "-", + }, + { + "LogicalResourceId": "LID_1", + "PhysicalResourceId": "PID_1", + "CloudEndpointOrFunctionURL": ["api.url1"], + "Methods": "-", + }, + { + "LogicalResourceId": "LID_1", + "PhysicalResourceId": "PID_1", + "CloudEndpointOrFunctionURL": ["api.url1", "api.url2", "api.url3"], + "Methods": ["/hello2['get, put']", "/hello['get']"], + }, + ] + testable_resources_to_table_mapper = TestableResourcesToTableMapper() + output = testable_resources_to_table_mapper.map(data) + self.assertEqual(output.get("table_name", ""), "Testable Resources") + + +class TestMapperConsumerFactory(TestCase): + def test_create_json_output(self): + factory = MapperConsumerFactory() + container = factory.create(ProducersEnum.STACK_OUTPUTS_PRODUCER, "json") + self.assertIsInstance(container.mapper, DataToJsonMapper) + self.assertIsInstance(container.consumer, StringConsumerJsonOutput) + + def test_create_stack_outputs_table_output(self): + factory = MapperConsumerFactory() + container = factory.create(ProducersEnum.STACK_OUTPUTS_PRODUCER, "table") + self.assertIsInstance(container.mapper, StackOutputToTableMapper) + self.assertIsInstance(container.consumer, StringConsumerTableOutput) + + def test_create_resources_table_output(self): + factory = MapperConsumerFactory() + container = factory.create(ProducersEnum.RESOURCES_PRODUCER, "table") + self.assertIsInstance(container.mapper, ResourcesToTableMapper) + self.assertIsInstance(container.consumer, StringConsumerTableOutput) + + def test_create_testable_resources_table_output(self): + factory = MapperConsumerFactory() + container = factory.create(ProducersEnum.TESTABLE_RESOURCES_PRODUCER, "table") + self.assertIsInstance(container.mapper, TestableResourcesToTableMapper) + self.assertIsInstance(container.consumer, StringConsumerTableOutput) + + +class TestTableConsumer(TestCase): + @patch("samcli.commands.list.json_consumer.click.secho") + @patch("samcli.commands.list.json_consumer.click.get_current_context") + def test_consume(self, patched_click_get_current_context, patched_click_echo): + consumer = StringConsumerTableOutput() + data = { + "format_string": "{OutputKey:<{0}} {OutputValue:<{1}} {Description:<{2}}", + "format_args": OrderedDict( + {"OutputKey": "OutputKey", "OutputValue": "OutputValue", "Description": "Description"} + ), + "table_name": "Stack Outputs", + "data": [], + } + consumer.consume(data) + print(patched_click_echo.call_args_list) + self.assertTrue(patched_click_echo.call_args_list) + self.assertEqual(call("Stack Outputs"), patched_click_echo.call_args_list[0]) diff --git a/tests/unit/commands/list/testable_resources/test_cli.py b/tests/unit/commands/list/testable_resources/test_cli.py index 83b441b6217..b55734a0774 100644 --- a/tests/unit/commands/list/testable_resources/test_cli.py +++ b/tests/unit/commands/list/testable_resources/test_cli.py @@ -9,13 +9,28 @@ def setUp(self): self.output = "json" self.region = None self.profile = None + self.template_file = None @patch("samcli.commands.list.testable_resources.cli.click") - def test_cli_base_command(self, mock_testable_resources_click): + @patch("samcli.commands.list.testable_resources.testable_resources_context.TestableResourcesContext") + def test_cli_base_command(self, mock_testable_resources_context, mock_testable_resources_click): context_mock = Mock() + mock_testable_resources_context.return_value.__enter__.return_value = context_mock do_cli( stack_name=self.stack_name, output=self.output, region=self.region, profile=self.profile, + template_file=self.template_file, ) + + mock_testable_resources_context.assert_called_with( + stack_name=self.stack_name, + output=self.output, + region=self.region, + profile=self.profile, + template_file=self.template_file, + ) + + context_mock.run.assert_called_with() + self.assertEqual(context_mock.run.call_count, 1) diff --git a/tests/unit/commands/list/testable_resources/test_testable_resources_context.py b/tests/unit/commands/list/testable_resources/test_testable_resources_context.py new file mode 100644 index 00000000000..aa6a503a1b1 --- /dev/null +++ b/tests/unit/commands/list/testable_resources/test_testable_resources_context.py @@ -0,0 +1,1017 @@ +from unittest import TestCase +from unittest.mock import patch, call, Mock +from botocore.exceptions import ClientError, EndpointConnectionError, NoCredentialsError, BotoCoreError +from samtranslator.translator.arn_generator import NoRegionFound + +from samcli.commands.list.testable_resources.testable_resources_context import TestableResourcesContext +from samcli.commands.local.cli_common.user_exceptions import InvalidSamTemplateException +from samcli.commands.validate.lib.exceptions import InvalidSamDocumentException +from samcli.commands.exceptions import RegionError, UserException +from samcli.commands.list.exceptions import ( + SamListLocalResourcesNotFoundError, + SamListUnknownClientError, + StackDoesNotExistInRegionError, + SamListUnknownBotoCoreError, +) +from samcli.lib.providers.sam_stack_provider import SamLocalStackProvider +from samtranslator.public.exceptions import InvalidDocumentException +from samcli.lib.translate.sam_template_validator import SamTemplateValidator +from samcli.lib.list.testable_resources.testable_resources_producer import TestableResourcesProducer, APIGatewayEnum +from samcli.lib.list.data_to_json_mapper import DataToJsonMapper +from samcli.commands.list.json_consumer import StringConsumerJsonOutput +from samcli.lib.providers.provider import Stack +from samcli.lib.list.data_to_json_mapper import DataToJsonMapper +from samcli.commands.list.json_consumer import StringConsumerJsonOutput +from samcli.commands.list.table_consumer import StringConsumerTableOutput +from samcli.lib.list.testable_resources.testable_resources_to_table_mapper import TestableResourcesToTableMapper + + +TRANSLATED_DICT_RETURN = { + "AWSTemplateFormatVersion": "2010-09-09", + "Description": "sam-app-hello\nSample SAM Template for sam-app-hello\n", + "Resources": { + "HelloWorldFunction": { + "Properties": { + "Architectures": ["x86_64"], + "Code": {"S3Bucket": "bucket", "S3Key": "value"}, + "Handler": "app.lambda_handler", + "Role": {"Fn::GetAtt": ["HelloWorldFunctionRole", "Arn"]}, + "Runtime": "python3.8", + "Tags": [{"Key": "lambda:createdBy", "Value": "SAM"}], + "Timeout": 3, + "TracingConfig": {"Mode": "Active"}, + }, + "Type": "AWS::Lambda::Function", + }, + "HelloWorldFunctionRole": { + "Properties": { + "AssumeRolePolicyDocument": { + "Statement": [ + { + "Action": ["sts:AssumeRole"], + "Effect": "Allow", + "Principal": {"Service": ["lambda.amazonaws.com"]}, + } + ], + "Version": "2012-10-17", + }, + "ManagedPolicyArns": [ + "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole", + "arn:aws:iam::aws:policy/AWSXrayWriteOnlyAccess", + ], + "Tags": [{"Key": "lambda:createdBy", "Value": "SAM"}], + }, + "Type": "AWS::IAM::Role", + }, + "HelloWorldFunctionHelloWorldPermissionProd": { + "Properties": { + "Action": "lambda:InvokeFunction", + "FunctionName": {"Ref": "HelloWorldFunction"}, + "Principal": "apigateway.amazonaws.com", + "SourceArn": { + "Fn::Sub": [ + "arn:aws:execute-api:${AWS::Region}:${AWS::AccountId}:${__ApiId__}/${__Stage__}/GET/hello", + {"__ApiId__": {"Ref": "ServerlessRestApi"}, "__Stage__": "*"}, + ] + }, + }, + "Type": "AWS::Lambda::Permission", + }, + "ServerlessRestApi": { + "Properties": { + "Body": { + "info": {"version": "1.0", "title": {"Ref": "AWS::StackName"}}, + "paths": { + "/hello": { + "get": { + "x-amazon-apigateway-integration": { + "httpMethod": "POST", + "type": "aws_proxy", + "uri": { + "Fn::Sub": "arn:aws:apigateway:${AWS::Region}:lambda:path/2015-03-31/functions/${HelloWorldFunction.Arn}/invocations" + }, + }, + "responses": {}, + } + } + }, + "swagger": "2.0", + } + }, + "Type": "AWS::ApiGateway::RestApi", + }, + "ServerlessRestApiDeploymentf5716dc08b": { + "Properties": { + "Description": "RestApi deployment id: f5716dc08b0d213bd0f2dfb686579c351b09ae49", + "RestApiId": {"Ref": "ServerlessRestApi"}, + "StageName": "Stage", + }, + "Type": "AWS::ApiGateway::Deployment", + }, + "ServerlessRestApiProdStage": { + "Properties": { + "DeploymentId": {"Ref": "ServerlessRestApiDeploymentf5716dc08b"}, + "RestApiId": {"Ref": "ServerlessRestApi"}, + "StageName": "Prod", + }, + "Type": "AWS::ApiGateway::Stage", + }, + }, +} + +TRANSLATED_DICT_RETURN_WITH_APIS = { + "AWSTemplateFormatVersion": "2010-09-09", + "Description": "sam-app-hello\nSample SAM Template for sam-app-hello\n", + "Resources": { + "customDomainCert": { + "Type": "AWS::CertificateManager::Certificate", + "Properties": {"DomainName": "api7.zhandr.people.aws.dev", "ValidationMethod": "DNS"}, + }, + "BPMapping1": { + "Type": "AWS::ApiGateway::BasePathMapping", + "Properties": {"DomainName": "apigw_dm_mapping_LID", "RestApiId": "test_apigw_restapi", "Stage": "String"}, + }, + "HelloWorldFunction": { + "Properties": { + "Architectures": ["x86_64"], + "Code": {"S3Bucket": "bucket", "S3Key": "value"}, + "Handler": "app.lambda_handler", + "Role": {"Fn::GetAtt": ["HelloWorldFunctionRole", "Arn"]}, + "Runtime": "python3.8", + "Tags": [{"Key": "lambda:createdBy", "Value": "SAM"}], + "Timeout": 3, + "TracingConfig": {"Mode": "Active"}, + }, + "Type": "AWS::Lambda::Function", + }, + "HelloWorldFunctionUrl": { + "Properties": {"AuthType": "AWS_IAM", "TargetFunctionArn": {"Ref": "HelloWorldFunction"}}, + "Type": "AWS::Lambda::Url", + }, + "HelloWorldFunctionRole": { + "Properties": { + "AssumeRolePolicyDocument": { + "Statement": [ + { + "Action": ["sts:AssumeRole"], + "Effect": "Allow", + "Principal": {"Service": ["lambda.amazonaws.com"]}, + } + ], + "Version": "2012-10-17", + }, + "ManagedPolicyArns": [ + "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole", + "arn:aws:iam::aws:policy/AWSXrayWriteOnlyAccess", + ], + "Tags": [{"Key": "lambda:createdBy", "Value": "SAM"}], + }, + "Type": "AWS::IAM::Role", + }, + "HelloWorldFunctionHelloWorldPermissionProd": { + "Properties": { + "Action": "lambda:InvokeFunction", + "FunctionName": {"Ref": "HelloWorldFunction"}, + "Principal": "apigateway.amazonaws.com", + "SourceArn": { + "Fn::Sub": [ + "arn:aws:execute-api:${AWS::Region}:${AWS::AccountId}:${__ApiId__}/${__Stage__}/GET/hello", + {"__ApiId__": {"Ref": "ServerlessRestApi"}, "__Stage__": "*"}, + ] + }, + }, + "Type": "AWS::Lambda::Permission", + }, + "HelloWorldFunctionHelloWorld2PermissionProd": { + "Properties": { + "Action": "lambda:InvokeFunction", + "FunctionName": {"Ref": "HelloWorldFunction"}, + "Principal": "apigateway.amazonaws.com", + "SourceArn": { + "Fn::Sub": [ + "arn:aws:execute-api:${AWS::Region}:${AWS::AccountId}:${__ApiId__}/${__Stage__}/GET, PUT/hello2", + {"__ApiId__": {"Ref": "ServerlessRestApi"}, "__Stage__": "*"}, + ] + }, + }, + "Type": "AWS::Lambda::Permission", + }, + "TestResource2": { + "Properties": { + "Body": { + "info": {"version": "1.0", "description": "Test resources", "title": {"Ref": "AWS::StackName"}}, + "paths": {}, + "openapi": "3.0.1", + "tags": [{"name": "httpapi:createdBy", "x-amazon-apigateway-tag-value": "SAM"}], + } + }, + "Type": "AWS::ApiGatewayV2::Api", + }, + "TestResource5": { + "Properties": { + "Body": { + "info": {"version": "1.0", "description": "Test resources", "title": {"Ref": "AWS::StackName"}}, + "paths": {}, + "openapi": "3.0.1", + "tags": [{"name": "httpapi:createdBy", "x-amazon-apigateway-tag-value": "SAM"}], + } + }, + "Type": "AWS::ApiGatewayV2::Api", + }, + "ApiGatewayDomainNameV28437445d28": { + "Properties": { + "DomainName": "api7.zhandr.people.aws.dev", + "DomainNameConfigurations": [ + {"CertificateArn": {"Ref": "customDomainCert"}, "EndpointType": "REGIONAL"} + ], + "Tags": {"httpapi:createdBy": "SAM"}, + }, + "Type": "AWS::ApiGatewayV2::DomainName", + }, + "TestResource2ApiMapping": { + "Properties": { + "ApiId": {"Ref": "TestResource2"}, + "DomainName": {"Ref": "ApiGatewayDomainNameV28437445d28"}, + "Stage": {"Ref": "TestResource2Test2Stage"}, + }, + "Type": "AWS::ApiGatewayV2::ApiMapping", + }, + "TestResource2Test2Stage": { + "Properties": { + "ApiId": {"Ref": "TestResource2"}, + "AutoDeploy": True, + "StageName": "Test2", + "Tags": {"httpapi:createdBy": "SAM"}, + }, + "Type": "AWS::ApiGatewayV2::Stage", + }, + "TestResource4": { + "Properties": { + "Body": { + "info": {"version": "1.0", "description": "Test resources", "title": {"Ref": "AWS::StackName"}}, + "paths": {}, + "openapi": "3.0.1", + "tags": [{"name": "httpapi:createdBy", "x-amazon-apigateway-tag-value": "SAM"}], + } + }, + "Type": "AWS::ApiGatewayV2::Api", + }, + "TestResource4Test2Stage": { + "Properties": { + "ApiId": {"Ref": "TestResource4"}, + "AutoDeploy": True, + "StageName": "Test2", + "Tags": {"httpapi:createdBy": "SAM"}, + }, + "Type": "AWS::ApiGatewayV2::Stage", + }, + "ServerlessRestApi": { + "Properties": { + "Body": { + "info": {"version": "1.0", "title": {"Ref": "AWS::StackName"}}, + "paths": { + "/hello2": { + "get, put": { + "x-amazon-apigateway-integration": { + "httpMethod": "POST", + "type": "aws_proxy", + "uri": { + "Fn::Sub": "arn:aws:apigateway:${AWS::Region}:lambda:path/2015-03-31/functions/${HelloWorldFunction.Arn}/invocations" + }, + }, + "responses": {}, + } + }, + "/hello": { + "get": { + "x-amazon-apigateway-integration": { + "httpMethod": "POST", + "type": "aws_proxy", + "uri": { + "Fn::Sub": "arn:aws:apigateway:${AWS::Region}:lambda:path/2015-03-31/functions/${HelloWorldFunction.Arn}/invocations" + }, + }, + "responses": {}, + } + }, + }, + "swagger": "2.0", + } + }, + "Type": "AWS::ApiGateway::RestApi", + }, + "ServerlessRestApiDeployment88d73b1fc4": { + "Properties": { + "Description": "RestApi deployment id: 88d73b1fc436b53afc5f54ce63096d44e97b741b", + "RestApiId": {"Ref": "ServerlessRestApi"}, + "StageName": "Stage", + }, + "Type": "AWS::ApiGateway::Deployment", + }, + "ServerlessRestApiProdStage": { + "Properties": { + "DeploymentId": {"Ref": "ServerlessRestApiDeployment88d73b1fc4"}, + "RestApiId": {"Ref": "ServerlessRestApi"}, + "StageName": "Prod", + }, + "Type": "AWS::ApiGateway::Stage", + }, + }, +} + +SAM_APP_HELLO_RETURN_RESPONSE = { + "StackResources": [ + { + "StackName": "sam-app-hello6", + "LogicalResourceId": "ApiGatewayDomainName1", + "PhysicalResourceId": "test.custom.domain1", + "ResourceType": "AWS::ApiGatewayV2::DomainName", + "ResourceStatus": "CREATE_COMPLETE", + "DriftInformation": {"StackResourceDriftStatus": "NOT_CHECKED"}, + }, + { + "StackName": "sam-app-hello6", + "LogicalResourceId": "HelloWorldFunction", + "PhysicalResourceId": "sam-app-hello6-HelloWorldFunction-testID", + "ResourceType": "AWS::Lambda::Function", + "ResourceStatus": "CREATE_COMPLETE", + "DriftInformation": {"StackResourceDriftStatus": "NOT_CHECKED"}, + }, + { + "StackName": "sam-app-hello6", + "LogicalResourceId": "HelloWorldFunctionUrl", + "PhysicalResourceId": "arn:aws:lambda:us-east-1:function:sam-app-hello6-HelloWorldFunction-testID", + "ResourceType": "AWS::Lambda::Url", + "ResourceStatus": "CREATE_COMPLETE", + "DriftInformation": {"StackResourceDriftStatus": "NOT_CHECKED"}, + }, + { + "StackName": "sam-app-hello6", + "LogicalResourceId": "ServerlessRestApi", + "PhysicalResourceId": "jwompba769", + "ResourceType": "AWS::ApiGateway::RestApi", + "ResourceStatus": "CREATE_COMPLETE", + "DriftInformation": {"StackResourceDriftStatus": "NOT_CHECKED"}, + }, + { + "StackName": "sam-app-hello6", + "LogicalResourceId": "ServerlessRestApiDeployment78c5316093", + "PhysicalResourceId": "lulx9h", + "ResourceType": "AWS::ApiGateway::Deployment", + "ResourceStatus": "CREATE_COMPLETE", + "DriftInformation": {"StackResourceDriftStatus": "NOT_CHECKED"}, + }, + { + "StackName": "sam-app-hello6", + "LogicalResourceId": "ServerlessRestApiProdStage", + "PhysicalResourceId": "Prod", + "ResourceType": "AWS::ApiGateway::Stage", + "ResourceStatus": "CREATE_COMPLETE", + "DriftInformation": {"StackResourceDriftStatus": "NOT_CHECKED"}, + }, + { + "StackName": "sam-app-hello6", + "LogicalResourceId": "TestResource2", + "PhysicalResourceId": "erj31jdyw5", + "ResourceType": "AWS::ApiGatewayV2::Api", + "ResourceStatus": "CREATE_COMPLETE", + "DriftInformation": {"StackResourceDriftStatus": "NOT_CHECKED"}, + }, + { + "StackName": "sam-app-hello6", + "LogicalResourceId": "TestResource2ApiMapping", + "PhysicalResourceId": "rut5pp", + "ResourceType": "AWS::ApiGatewayV2::ApiMapping", + "ResourceStatus": "CREATE_COMPLETE", + "DriftInformation": {"StackResourceDriftStatus": "NOT_CHECKED"}, + }, + { + "StackName": "sam-app-hello6", + "LogicalResourceId": "TestResource2Test2Stage", + "PhysicalResourceId": "Test2", + "ResourceType": "AWS::ApiGatewayV2::Stage", + "ResourceStatus": "CREATE_COMPLETE", + "DriftInformation": {"StackResourceDriftStatus": "NOT_CHECKED"}, + }, + { + "StackName": "sam-app-hello6", + "LogicalResourceId": "TestResource4", + "PhysicalResourceId": "5u9ekr1d32", + "ResourceType": "AWS::ApiGatewayV2::Api", + "ResourceStatus": "CREATE_COMPLETE", + "DriftInformation": {"StackResourceDriftStatus": "NOT_CHECKED"}, + }, + { + "StackName": "sam-app-hello6", + "LogicalResourceId": "TestResource4Test2Stage", + "PhysicalResourceId": "Test2", + "ResourceType": "AWS::ApiGatewayV2::Stage", + "ResourceStatus": "CREATE_COMPLETE", + "DriftInformation": {"StackResourceDriftStatus": "NOT_CHECKED"}, + }, + { + "StackName": "sam-app-hello6", + "LogicalResourceId": "customDomainCert", + "PhysicalResourceId": "arn:aws:acm:us-east-1:certificate", + "ResourceType": "AWS::CertificateManager::Certificate", + "ResourceStatus": "CREATE_COMPLETE", + "DriftInformation": {"StackResourceDriftStatus": "NOT_CHECKED"}, + }, + { + "StackName": "sam-app-hello6", + "LogicalResourceId": "test_apigw_restapi", + "PhysicalResourceId": "testPID", + "ResourceType": "AWS::ApiGateway::RestApi", + "ResourceStatus": "CREATE_COMPLETE", + "DriftInformation": {"StackResourceDriftStatus": "NOT_CHECKED"}, + }, + { + "StackName": "sam-app-hello6", + "LogicalResourceId": "apigw_dm_mapping_LID", + "PhysicalResourceId": "test.custom.bpmapping.domain", + "ResourceType": "AWS::ApiGateway::DomainName", + }, + { + "StackName": "sam-app-hello6", + "LogicalResourceId": "BPMapping1", + "PhysicalResourceId": "bp_mapping_PID", + "ResourceType": "AWS::ApiGateway::BasePathMapping", + }, + ], + "ResponseMetadata": { + "RequestId": "b15914d5-009b-46ce-aab8-458efc09f34d", + "HTTPStatusCode": 200, + "HTTPHeaders": { + "x-amzn-requestid": "b15914d5-009b-46ce-aab8-458efc09f34d", + "content-type": "text/xml", + "content-length": "10370", + "date": "Mon, 25 Jul 2022 20:27:05 GMT", + }, + "RetryAttempts": 0, + }, +} + +SAM_FILE_READER_RETURN = { + "AWSTemplateFormatVersion": "2010-09-09", + "Transform": "AWS::Serverless-2016-10-31", + "Description": "sam-app-hello\nSample SAM Template for sam-app-hello\n", + "Globals": {"Function": {"Tracing": "Active", "Timeout": 3}}, + "Resources": { + "HelloWorldFunction": { + "Type": "AWS::Serverless::Function", + "Properties": { + "CodeUri": "hello_world/", + "Handler": "app.lambda_handler", + "Architectures": ["x86_64"], + "Runtime": "python3.8", + "Events": {"HelloWorld": {"Type": "Api", "Properties": {"Path": "/hello", "Method": "get"}}}, + }, + } + }, +} + + +class TestTestableResourcesInitClients(TestCase): + @patch("samcli.commands.list.json_consumer.click.echo") + @patch("samcli.commands.list.json_consumer.click.get_current_context") + @patch("boto3.Session.region_name", None) + def test_init_clients_no_region(self, patched_click_get_current_context, patched_click_echo): + with self.assertRaises(RegionError): + with TestableResourcesContext( + stack_name="test", output="json", region=None, profile=None, template_file=None + ) as testable_resources_context: + testable_resources_context.init_clients() + + @patch("samcli.commands.list.json_consumer.click.echo") + @patch("samcli.commands.list.json_consumer.click.get_current_context") + @patch("boto3.Session.region_name", "us-east-1") + def test_init_clients_no_input_region_get_region_from_session( + self, patched_click_get_current_context, patched_click_echo + ): + with TestableResourcesContext( + stack_name="test", output="json", region=None, profile=None, template_file=None + ) as testable_resources_context: + testable_resources_context.init_clients() + self.assertEqual(testable_resources_context.region, "us-east-1") + + +class TestGetFunctionUrl(TestCase): + @patch("samcli.commands.list.json_consumer.click.echo") + @patch("samcli.commands.list.json_consumer.click.get_current_context") + @patch("samcli.commands.list.cli_common.list_common_context.get_boto_client_provider_with_config") + def test_clienterror_resource_not_found( + self, + mock_client_provider, + patched_click_get_current_context, + patched_click_echo, + ): + mock_client_provider.return_value.return_value.get_resource.side_effect = ClientError( + {"Error": {"Code": "ResourceNotFoundException", "Message": "The resource you requested does not exist"}}, + "GetResources", + ) + testable_resource_producer = TestableResourcesProducer( + stack_name=None, + region="us-east-1", + profile=None, + template_file=None, + cloudformation_client=None, + iam_client=None, + cloudcontrol_client=mock_client_provider.return_value.return_value, + apigateway_client=None, + apigatewayv2_client=None, + mapper=None, + consumer=None, + ) + response = testable_resource_producer.get_function_url("testID") + self.assertEqual(response, "-") + + @patch("samcli.commands.list.json_consumer.click.echo") + @patch("samcli.commands.list.json_consumer.click.get_current_context") + @patch("samcli.commands.list.cli_common.list_common_context.get_boto_client_provider_with_config") + def test_clienterror_others( + self, + mock_client_provider, + patched_click_get_current_context, + patched_click_echo, + ): + mock_client_provider.return_value.return_value.get_resource.side_effect = ClientError( + {"Error": {"Code": "ExpiredToken", "Message": "The security token included in the request is expired"}}, + "DescribeStacks", + ) + with self.assertRaises(SamListUnknownClientError): + testable_resource_producer = TestableResourcesProducer( + stack_name=None, + region="us-east-1", + profile=None, + template_file=None, + cloudformation_client=None, + iam_client=None, + cloudcontrol_client=mock_client_provider.return_value.return_value, + apigateway_client=None, + apigatewayv2_client=None, + mapper=None, + consumer=None, + ) + testable_resource_producer.get_function_url("testID") + + @patch("samcli.commands.list.json_consumer.click.echo") + @patch("samcli.commands.list.json_consumer.click.get_current_context") + @patch("samcli.commands.list.cli_common.list_common_context.get_boto_client_provider_with_config") + def test_properties_not_in_response( + self, + mock_client_provider, + patched_click_get_current_context, + patched_click_echo, + ): + mock_client_provider.return_value.return_value.get_resource.return_value = {} + testable_resource_producer = TestableResourcesProducer( + stack_name=None, + region="us-east-1", + profile=None, + template_file=None, + cloudformation_client=None, + iam_client=None, + cloudcontrol_client=mock_client_provider.return_value.return_value, + apigateway_client=None, + apigatewayv2_client=None, + mapper=None, + consumer=None, + ) + response = testable_resource_producer.get_function_url("testID") + self.assertEqual(response, "-") + + @patch("samcli.commands.list.json_consumer.click.echo") + @patch("samcli.commands.list.json_consumer.click.get_current_context") + @patch("samcli.commands.list.cli_common.list_common_context.get_boto_client_provider_with_config") + def test_properties_in_response( + self, + mock_client_provider, + patched_click_get_current_context, + patched_click_echo, + ): + mock_client_provider.return_value.return_value.get_resource.return_value = { + "TypeName": "AWS::Lambda::Url", + "ResourceDescription": { + "Identifier": "testid", + "Properties": '{"FunctionArn":"arn:aws:lambda:sam-app-hello-HelloWorldFunction","FunctionUrl":"https://test.lambda-url.us-east-1.on.aws/","AuthType":"AWS_IAM"}', + }, + "ResponseMetadata": { + "RequestId": "testID", + "HTTPStatusCode": 200, + "HTTPHeaders": { + "x-amzn-requestid": "testID", + "date": "testDate", + "content-type": "application/x-amz-json-1.0", + "content-length": "408", + }, + "RetryAttempts": 0, + }, + } + testable_resource_producer = TestableResourcesProducer( + stack_name=None, + region="us-east-1", + profile=None, + template_file=None, + cloudformation_client=None, + iam_client=None, + cloudcontrol_client=mock_client_provider.return_value.return_value, + apigateway_client=None, + apigatewayv2_client=None, + mapper=None, + consumer=None, + ) + response = testable_resource_producer.get_function_url("testID") + self.assertEqual(response, "https://test.lambda-url.us-east-1.on.aws/") + + +class TestGetStages(TestCase): + @patch("samcli.commands.list.json_consumer.click.echo") + @patch("samcli.commands.list.json_consumer.click.get_current_context") + @patch("samcli.commands.list.cli_common.list_common_context.get_boto_client_provider_with_config") + def test_apigw_v2_stages( + self, + mock_client_provider, + patched_click_get_current_context, + patched_click_echo, + ): + mock_client_provider.return_value.return_value.get_stages.return_value = { + "ResponseMetadata": { + "RequestId": "testid", + "HTTPStatusCode": 200, + "HTTPHeaders": { + "date": "Mon, 18 Jul 2022 20:59:15 GMT", + "content-type": "application/json", + "content-length": "762", + "connection": "keep-alive", + "x-amzn-requestid": "testid", + "access-control-allow-origin": "*", + "x-amz-apigw-id": "testid", + "access-control-expose-headers": "x-amzn-RequestId,x-amzn-ErrorType,x-amzn-ErrorMessage,Date", + "x-amzn-trace-id": "Root=testid", + }, + "RetryAttempts": 0, + }, + "Items": [ + { + "AutoDeploy": True, + "DefaultRouteSettings": {"DetailedMetricsEnabled": False}, + "RouteSettings": {}, + "StageName": "$default", + "StageVariables": {}, + } + ], + } + testable_resource_producer = TestableResourcesProducer( + stack_name=None, + region="us-east-1", + profile=None, + template_file=None, + cloudformation_client=None, + iam_client=None, + cloudcontrol_client=None, + apigateway_client=None, + apigatewayv2_client=mock_client_provider.return_value.return_value, + mapper=None, + consumer=None, + ) + response = testable_resource_producer.get_stage_list("testID", APIGatewayEnum.API_GATEWAY_V2) + self.assertEqual(response, ["$default"]) + + @patch("samcli.commands.list.json_consumer.click.echo") + @patch("samcli.commands.list.json_consumer.click.get_current_context") + @patch("samcli.commands.list.cli_common.list_common_context.get_boto_client_provider_with_config") + def test_apigw_stages( + self, + mock_client_provider, + patched_click_get_current_context, + patched_click_echo, + ): + mock_client_provider.return_value.return_value.get_stages.return_value = { + "ResponseMetadata": { + "RequestId": "testID", + "HTTPStatusCode": 200, + "HTTPHeaders": { + "date": "Mon, 18 Jul 2022 21:15:06 GMT", + "content-type": "application/json", + "content-length": "679", + "connection": "keep-alive", + "x-amzn-requestid": "testID", + "x-amz-apigw-id": "testID", + }, + "RetryAttempts": 0, + }, + "item": [ + { + "deploymentId": "t50nmu", + "stageName": "Prod", + "cacheClusterEnabled": False, + "cacheClusterStatus": "NOT_AVAILABLE", + "methodSettings": {}, + "tracingEnabled": False, + "tags": {"aws:cloudformation:logical-id": "testID", "aws:cloudformation:stack-name": "testStack"}, + }, + { + "deploymentId": "t50nmu", + "stageName": "Stage", + "cacheClusterEnabled": False, + "cacheClusterStatus": "NOT_AVAILABLE", + "methodSettings": {}, + "tracingEnabled": False, + }, + ], + } + testable_resource_producer = TestableResourcesProducer( + stack_name=None, + region="us-east-1", + profile=None, + template_file=None, + cloudformation_client=None, + iam_client=None, + cloudcontrol_client=None, + apigateway_client=mock_client_provider.return_value.return_value, + apigatewayv2_client=None, + mapper=None, + consumer=None, + ) + response = testable_resource_producer.get_stage_list("testID", APIGatewayEnum.API_GATEWAY) + self.assertEqual(response, ["Prod", "Stage"]) + + @patch("samcli.commands.list.json_consumer.click.echo") + @patch("samcli.commands.list.json_consumer.click.get_current_context") + @patch("samcli.commands.list.cli_common.list_common_context.get_boto_client_provider_with_config") + def test_apigw_stages_empty_return( + self, + mock_client_provider, + patched_click_get_current_context, + patched_click_echo, + ): + mock_client_provider.return_value.return_value.get_stages.return_value = { + "ResponseMetadata": { + "RequestId": "testID", + "HTTPStatusCode": 200, + "HTTPHeaders": { + "date": "Mon, 18 Jul 2022 21:15:06 GMT", + "content-type": "application/json", + "content-length": "679", + "connection": "keep-alive", + "x-amzn-requestid": "testID", + "x-amz-apigw-id": "testID", + }, + "RetryAttempts": 0, + }, + } + testable_resource_producer = TestableResourcesProducer( + stack_name=None, + region="us-east-1", + profile=None, + template_file=None, + cloudformation_client=None, + iam_client=None, + cloudcontrol_client=None, + apigateway_client=mock_client_provider.return_value.return_value, + apigatewayv2_client=None, + mapper=None, + consumer=None, + ) + response = testable_resource_producer.get_stage_list("testID", APIGatewayEnum.API_GATEWAY) + self.assertEqual(response, []) + + @patch("samcli.commands.list.json_consumer.click.echo") + @patch("samcli.commands.list.json_consumer.click.get_current_context") + @patch("samcli.commands.list.cli_common.list_common_context.get_boto_client_provider_with_config") + def test_get_stage_list_unknown_clienterror( + self, + mock_client_provider, + patched_click_get_current_context, + patched_click_echo, + ): + mock_client_provider.return_value.return_value.get_stages.side_effect = ClientError( + {"Error": {"Code": "ExpiredToken", "Message": "The security token included in the request is expired"}}, + "DescribeStacks", + ) + with self.assertRaises(SamListUnknownClientError): + testable_resource_producer = TestableResourcesProducer( + stack_name=None, + region="us-east-1", + profile=None, + template_file=None, + cloudformation_client=None, + iam_client=None, + cloudcontrol_client=None, + apigateway_client=mock_client_provider.return_value.return_value, + apigatewayv2_client=mock_client_provider.return_value.return_value, + mapper=None, + consumer=None, + ) + testable_resource_producer.get_stage_list("testID", APIGatewayEnum.API_GATEWAY) + + @patch("samcli.commands.list.json_consumer.click.echo") + @patch("samcli.commands.list.json_consumer.click.get_current_context") + @patch("samcli.commands.list.cli_common.list_common_context.get_boto_client_provider_with_config") + def test_get_stage_list_not_found_exception_clienterror( + self, + mock_client_provider, + patched_click_get_current_context, + patched_click_echo, + ): + mock_client_provider.return_value.return_value.get_stages.side_effect = ClientError( + {"Error": {"Code": "NotFoundException", "Message": ""}}, + "DescribeStacks", + ) + testable_resource_producer = TestableResourcesProducer( + stack_name=None, + region="us-east-1", + profile=None, + template_file=None, + cloudformation_client=None, + iam_client=None, + cloudcontrol_client=None, + apigateway_client=mock_client_provider.return_value.return_value, + apigatewayv2_client=mock_client_provider.return_value.return_value, + mapper=None, + consumer=None, + ) + response = testable_resource_producer.get_stage_list("testID", APIGatewayEnum.API_GATEWAY) + self.assertEqual(response, []) + + @patch("samcli.commands.list.json_consumer.click.echo") + @patch("samcli.commands.list.json_consumer.click.get_current_context") + @patch("samcli.commands.list.cli_common.list_common_context.get_boto_client_provider_with_config") + def test_get_stage_list_unknown_botocore_error( + self, + mock_client_provider, + patched_click_get_current_context, + patched_click_echo, + ): + mock_client_provider.return_value.return_value.get_stages.side_effect = EndpointConnectionError( + endpoint_url="https://cloudformation.test.amazonaws.com/" + ) + with self.assertRaises(SamListUnknownBotoCoreError): + testable_resource_producer = TestableResourcesProducer( + stack_name=None, + region="us-east-1", + profile=None, + template_file=None, + cloudformation_client=None, + iam_client=None, + cloudcontrol_client=None, + apigateway_client=mock_client_provider.return_value.return_value, + apigatewayv2_client=mock_client_provider.return_value.return_value, + mapper=None, + consumer=None, + ) + testable_resource_producer.get_stage_list("testID", APIGatewayEnum.API_GATEWAY) + + +class TestBuildAPIGWEndpoints(TestCase): + @patch("samcli.commands.list.json_consumer.click.echo") + @patch("samcli.commands.list.json_consumer.click.get_current_context") + def test_build_api_gw_endpoints( + self, + patched_click_get_current_context, + patched_click_echo, + ): + testable_resource_producer = TestableResourcesProducer( + stack_name=None, + region="us-east-1", + profile=None, + template_file=None, + cloudformation_client=None, + iam_client=None, + cloudcontrol_client=None, + apigateway_client=None, + apigatewayv2_client=None, + mapper=None, + consumer=None, + ) + repsonse1 = testable_resource_producer.build_api_gw_endpoints("testID", []) + self.assertEqual(repsonse1, []) + repsonse2 = testable_resource_producer.build_api_gw_endpoints("testID", ["Prod"]) + self.assertEqual(repsonse2, ["https://testID.execute-api.us-east-1.amazonaws.com/Prod"]) + + +class TestTestableResourcesProducerProduce(TestCase): + @patch("samcli.commands.list.json_consumer.click.echo") + @patch("samcli.commands.list.json_consumer.click.get_current_context") + @patch("samcli.lib.list.testable_resources.testable_resources_producer.SamLocalStackProvider.get_stacks") + @patch("samcli.lib.list.testable_resources.testable_resources_producer.get_template_data") + @patch( + "samcli.lib.list.testable_resources.testable_resources_producer.TestableResourcesProducer.get_translated_dict" + ) + def test_produce_resources_not_found_error( + self, + mock_get_translated_dict, + mock_get_template_data, + mock_get_stacks, + patched_click_get_current_context, + patched_click_echo, + ): + mock_get_template_data.return_value = {} + mock_get_translated_dict.return_value = {} + mock_get_stacks.return_value = ([], []) + testable_resource_producer = TestableResourcesProducer( + stack_name=None, + region="us-east-1", + profile=None, + template_file=None, + cloudformation_client=None, + iam_client=None, + cloudcontrol_client=None, + apigateway_client=None, + apigatewayv2_client=None, + mapper=None, + consumer=None, + ) + with self.assertRaises(SamListLocalResourcesNotFoundError): + testable_resource_producer.produce() + + @patch("samcli.commands.list.json_consumer.click.echo") + @patch("samcli.commands.list.json_consumer.click.get_current_context") + @patch("samcli.lib.list.testable_resources.testable_resources_producer.get_template_data") + @patch( + "samcli.lib.list.testable_resources.testable_resources_producer.TestableResourcesProducer.get_translated_dict" + ) + def test_produce_no_stack_name_json( + self, + mock_get_translated_dict, + mock_get_template_data, + patched_click_get_current_context, + patched_click_echo, + ): + mock_get_template_data.return_value = {} + mock_get_translated_dict.return_value = TRANSLATED_DICT_RETURN_WITH_APIS + + stacks = SamLocalStackProvider.get_stacks( + template_file="", template_dictionary=mock_get_translated_dict.return_value + ) + testable_resource_producer = TestableResourcesProducer( + stack_name=None, + region="us-east-1", + profile=None, + template_file=None, + cloudformation_client=None, + iam_client=None, + cloudcontrol_client=None, + apigateway_client=None, + apigatewayv2_client=None, + mapper=DataToJsonMapper(), + consumer=StringConsumerJsonOutput(), + ) + testable_resource_producer.produce() + expected_output = [ + call( + '[\n {\n "LogicalResourceId": "HelloWorldFunction",\n "PhysicalResourceId": "-",\n "CloudEndpointOrFunctionURL": "-",\n "Methods": "-"\n },\n {\n "LogicalResourceId": "TestResource2",\n "PhysicalResourceId": "-",\n "CloudEndpointOrFunctionURL": "-",\n "Methods": []\n },\n {\n "LogicalResourceId": "TestResource5",\n "PhysicalResourceId": "-",\n "CloudEndpointOrFunctionURL": "-",\n "Methods": []\n },\n {\n "LogicalResourceId": "TestResource4",\n "PhysicalResourceId": "-",\n "CloudEndpointOrFunctionURL": "-",\n "Methods": []\n },\n {\n "LogicalResourceId": "ServerlessRestApi",\n "PhysicalResourceId": "-",\n "CloudEndpointOrFunctionURL": "-",\n "Methods": [\n "/hello2[\'get, put\']",\n "/hello[\'get\']"\n ]\n }\n]' + ) + ] + self.assertEqual(patched_click_echo.call_args_list, expected_output) + + @patch("samcli.commands.list.json_consumer.click.echo") + @patch("samcli.commands.list.json_consumer.click.get_current_context") + @patch("samcli.lib.list.testable_resources.testable_resources_producer.get_template_data") + @patch( + "samcli.lib.list.testable_resources.testable_resources_producer.TestableResourcesProducer.get_translated_dict" + ) + @patch( + "samcli.lib.list.testable_resources.testable_resources_producer.TestableResourcesProducer.get_resources_info" + ) + @patch("samcli.lib.list.testable_resources.testable_resources_producer.TestableResourcesProducer.get_function_url") + @patch("samcli.lib.list.testable_resources.testable_resources_producer.TestableResourcesProducer.get_stage_list") + def test_produce_has_stack_name_( + self, + mock_get_stages_list, + mock_get_function_url, + mock_get_resources_info, + mock_get_translated_dict, + mock_get_template_data, + patched_click_get_current_context, + patched_click_echo, + ): + mock_get_stages_list.return_value = ["testStage"] + mock_get_function_url.return_value = "test.function.url" + mock_get_resources_info.return_value = SAM_APP_HELLO_RETURN_RESPONSE + mock_get_template_data.return_value = {} + mock_get_translated_dict.return_value = TRANSLATED_DICT_RETURN_WITH_APIS + + stacks = SamLocalStackProvider.get_stacks( + template_file="", template_dictionary=mock_get_translated_dict.return_value + ) + testable_resource_producer = TestableResourcesProducer( + stack_name="sam-app-hello6", + region="us-east-1", + profile=None, + template_file=None, + cloudformation_client=None, + iam_client=None, + cloudcontrol_client=None, + apigateway_client=None, + apigatewayv2_client=None, + mapper=DataToJsonMapper(), + consumer=StringConsumerJsonOutput(), + ) + testable_resource_producer.produce() + expected_output = [ + call( + '[\n {\n "LogicalResourceId": "HelloWorldFunction",\n "PhysicalResourceId": "sam-app-hello6-HelloWorldFunction-testID",\n "CloudEndpointOrFunctionURL": "test.function.url",\n "Methods": "-"\n },\n {\n "LogicalResourceId": "ServerlessRestApi",\n "PhysicalResourceId": "jwompba769",\n "CloudEndpointOrFunctionURL": [\n "https://jwompba769.execute-api.us-east-1.amazonaws.com/testStage"\n ],\n "Methods": [\n "/hello2[\'get, put\']",\n "/hello[\'get\']"\n ]\n },\n {\n "LogicalResourceId": "TestResource2",\n "PhysicalResourceId": "erj31jdyw5",\n "CloudEndpointOrFunctionURL": [\n "https://erj31jdyw5.execute-api.us-east-1.amazonaws.com/testStage"\n ],\n "Methods": []\n },\n {\n "LogicalResourceId": "TestResource4",\n "PhysicalResourceId": "5u9ekr1d32",\n "CloudEndpointOrFunctionURL": [\n "https://5u9ekr1d32.execute-api.us-east-1.amazonaws.com/testStage"\n ],\n "Methods": []\n },\n {\n "LogicalResourceId": "test_apigw_restapi",\n "PhysicalResourceId": "testPID",\n "CloudEndpointOrFunctionURL": [\n "https://test.custom.bpmapping.domain"\n ],\n "Methods": []\n },\n {\n "LogicalResourceId": "TestResource5",\n "PhysicalResourceId": "-",\n "CloudEndpointOrFunctionURL": "-",\n "Methods": []\n }\n]' + ) + ] + self.assertEqual(patched_click_echo.call_args_list, expected_output)