20476669022

Committed 23 Dec 2025 06:23PM UTC coverage: 86.921% (+0.009%) from 86.912%

Build # 20476669022

Build Type

push

github

Committed by

web-flow

Commit Message

Fix KMS model annotations (#13563)

Run Details

2 of 2 new or added lines in 1 file covered. (100.0%)

266 existing lines in 7 files now uncovered.

70055 of 80596 relevant lines covered (86.92%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

80.95

/localstack-core/localstack/services/cloudformation/engine/v2/change_set_resource_support_checker.py

from localstack.aws.api.cloudformation import ChangeSetType
from localstack.services.cloudformation.engine.v2.change_set_model import NodeResource
from localstack.services.cloudformation.engine.v2.change_set_model_visitor import (
    ChangeSetModelVisitor,
)
from localstack.services.cloudformation.engine.v2.unsupported_resource import (
    should_ignore_unsupported_resource_type,
)
from localstack.services.cloudformation.resources import AWS_AVAILABLE_CFN_RESOURCES
from localstack.utils.catalog.catalog import (
    AwsServicesSupportStatus,
    CatalogPlugin,
    CfnResourceSupportStatus,
)
from localstack.utils.catalog.common import (
    AwsServicesSupportInLatest,
    AwsServiceSupportAtRuntime,
    CloudFormationResourcesSupportAtRuntime,
    CloudFormationResourcesSupportInLatest,
)
from localstack.utils.catalog.plugins import get_aws_catalog


# TODO handle all available resource types
def _get_service_name(resource_type: str) -> str | None:
    parts = resource_type.split("::")
    if len(parts) == 1:
        return None

    match parts:
        case _ if "Cognito::IdentityPool" in resource_type:
            return "cognito-identity"
        case [*_, "Cognito", "UserPool"]:
            return "cognito-idp"
        case [*_, "Cognito", _]:
            return "cognito-idp"
        case [*_, "Elasticsearch", _]:
            return "es"
        case [*_, "OpenSearchService", _]:
            return "opensearch"
        case [*_, "KinesisFirehose", _]:
            return "firehose"
        case [*_, "ResourceGroups", _]:
            return "resource-groups"
        case [*_, "CertificateManager", _]:
            return "acm"
        case _ if "ElasticLoadBalancing::" in resource_type:
            return "elb"
        case _ if "ElasticLoadBalancingV2::" in resource_type:
            return "elbv2"
        case _ if "ApplicationAutoScaling::" in resource_type:
            return "application-autoscaling"
        case _ if "MSK::" in resource_type:
            return "kafka"
        case _ if "Timestream::" in resource_type:
            return "timestream-write"
        case [_, service, *_]:
            return service.lower()


def _build_resource_failure_message(
    resource_type: str, status: AwsServicesSupportStatus | CfnResourceSupportStatus
) -> str:
    service_name = _get_service_name(resource_type) or "malformed"
    template = "Sorry, the {resource} resource in the {service} service is not supported."
    match status:
        case CloudFormationResourcesSupportAtRuntime.NOT_IMPLEMENTED:
            template = "Sorry, the {resource} resource (from the {service} service) is not supported by this version of LocalStack, but is available in the latest version."
        case CloudFormationResourcesSupportInLatest.NOT_SUPPORTED:
            template = "Sorry, the {resource} resource (from the {service} service) is not currently supported by LocalStack."
        case AwsServiceSupportAtRuntime.AVAILABLE_WITH_LICENSE_UPGRADE:
            template = "Sorry, the {service} service (for the {resource} resource) is not included within your LocalStack license, but is available in an upgraded license."
        case AwsServiceSupportAtRuntime.NOT_IMPLEMENTED:
            template = "The API for service {service} (for the {resource} resource) is either not included in your current license plan or has not yet been emulated by LocalStack."
        case AwsServicesSupportInLatest.NOT_SUPPORTED:
            template = "Sorry, the {service} (for the {resource} resource) service is not currently supported by LocalStack."
        case AwsServicesSupportInLatest.SUPPORTED_WITH_LICENSE_UPGRADE:
            template = "Sorry, the {service} service (for the {resource} resource) is not supported by this version of LocalStack, but is available in the latest version if you upgrade to the latest stable version."
    return template.format(
        resource=resource_type,
        service=service_name,
    )


class ChangeSetResourceSupportChecker(ChangeSetModelVisitor):
    change_set_type: ChangeSetType
    catalog: CatalogPlugin

    TITLE_MESSAGE = "Unsupported resources detected:"

    def __init__(self, change_set_type: ChangeSetType):
        self._resource_failure_messages: dict[str, str] = {}
        self.change_set_type = change_set_type
        self.catalog = get_aws_catalog()

    def visit_node_resource(self, node_resource: NodeResource):
        resource_type = node_resource.type_.value
        ignore_unsupported = should_ignore_unsupported_resource_type(
            resource_type=resource_type, change_set_type=self.change_set_type
        )

        if resource_type not in self._resource_failure_messages and not ignore_unsupported:
            if resource_type not in AWS_AVAILABLE_CFN_RESOURCES:
                # Ignore non-AWS resources
                pass
            support_status = self._resource_support_status(resource_type)
            if support_status == CloudFormationResourcesSupportAtRuntime.AVAILABLE:
                pass
            else:
                failure_message = _build_resource_failure_message(resource_type, support_status)
                self._resource_failure_messages[resource_type] = failure_message
        super().visit_node_resource(node_resource)

    def _resource_support_status(
        self, resource_type: str
    ) -> AwsServicesSupportStatus | CfnResourceSupportStatus:
        service_name = _get_service_name(resource_type)
        return self.catalog.get_cloudformation_resource_status(resource_type, service_name, True)

    @property
    def failure_messages(self) -> list[str]:
        return list(self._resource_failure_messages.values())

1	from localstack.aws.api.cloudformation import ChangeSetType	1✔
2	from localstack.services.cloudformation.engine.v2.change_set_model import NodeResource	1✔
3	from localstack.services.cloudformation.engine.v2.change_set_model_visitor import (	1✔
4	ChangeSetModelVisitor,
5	)
6	from localstack.services.cloudformation.engine.v2.unsupported_resource import (	1✔
7	should_ignore_unsupported_resource_type,
8	)
9	from localstack.services.cloudformation.resources import AWS_AVAILABLE_CFN_RESOURCES	1✔
10	from localstack.utils.catalog.catalog import (	1✔
11	AwsServicesSupportStatus,
12	CatalogPlugin,
13	CfnResourceSupportStatus,
14	)
15	from localstack.utils.catalog.common import (	1✔
16	AwsServicesSupportInLatest,
17	AwsServiceSupportAtRuntime,
18	CloudFormationResourcesSupportAtRuntime,
19	CloudFormationResourcesSupportInLatest,
20	)
21	from localstack.utils.catalog.plugins import get_aws_catalog	1✔
22
23
24	# TODO handle all available resource types
25	def _get_service_name(resource_type: str) -> str \| None:	1✔
26	parts = resource_type.split("::")	1✔
27	if len(parts) == 1:	1✔
UNCOV 28	return None	×
29
30	match parts:	1✔
31	case _ if "Cognito::IdentityPool" in resource_type:	1✔
32	return "cognito-identity"	×
33	case [*_, "Cognito", "UserPool"]:	1✔
34	return "cognito-idp"	×
35	case [*_, "Cognito", _]:	1✔
36	return "cognito-idp"	×
37	case [*_, "Elasticsearch", _]:	1✔
38	return "es"	×
39	case [*_, "OpenSearchService", _]:	1✔
40	return "opensearch"	×
41	case [*_, "KinesisFirehose", _]:	1✔
42	return "firehose"	×
43	case [*_, "ResourceGroups", _]:	1✔
44	return "resource-groups"	×
45	case [*_, "CertificateManager", _]:	1✔
46	return "acm"	×
47	case _ if "ElasticLoadBalancing::" in resource_type:	1✔
48	return "elb"	×
49	case _ if "ElasticLoadBalancingV2::" in resource_type:	1✔
50	return "elbv2"	×
51	case _ if "ApplicationAutoScaling::" in resource_type:	1✔
52	return "application-autoscaling"	×
53	case _ if "MSK::" in resource_type:	1✔
54	return "kafka"	×
55	case _ if "Timestream::" in resource_type:	1✔
UNCOV 56	return "timestream-write"	×
57	case [_, service, *_]:	1✔
58	return service.lower()	1✔
59
60
61	def _build_resource_failure_message(	1✔
62	resource_type: str, status: AwsServicesSupportStatus \| CfnResourceSupportStatus
63	) -> str:
64	service_name = _get_service_name(resource_type) or "malformed"	1✔
65	template = "Sorry, the {resource} resource in the {service} service is not supported."	1✔
66	match status:	1✔
67	case CloudFormationResourcesSupportAtRuntime.NOT_IMPLEMENTED:	1✔
68	template = "Sorry, the {resource} resource (from the {service} service) is not supported by this version of LocalStack, but is available in the latest version."	1✔
69	case CloudFormationResourcesSupportInLatest.NOT_SUPPORTED:	1✔
70	template = "Sorry, the {resource} resource (from the {service} service) is not currently supported by LocalStack."	1✔
71	case AwsServiceSupportAtRuntime.AVAILABLE_WITH_LICENSE_UPGRADE:	1✔
72	template = "Sorry, the {service} service (for the {resource} resource) is not included within your LocalStack license, but is available in an upgraded license."	1✔
73	case AwsServiceSupportAtRuntime.NOT_IMPLEMENTED:	1✔
74	template = "The API for service {service} (for the {resource} resource) is either not included in your current license plan or has not yet been emulated by LocalStack."	×
75	case AwsServicesSupportInLatest.NOT_SUPPORTED:	1✔
UNCOV 76	template = "Sorry, the {service} (for the {resource} resource) service is not currently supported by LocalStack."	×
77	case AwsServicesSupportInLatest.SUPPORTED_WITH_LICENSE_UPGRADE:	1✔
78	template = "Sorry, the {service} service (for the {resource} resource) is not supported by this version of LocalStack, but is available in the latest version if you upgrade to the latest stable version."	1✔
79	return template.format(	1✔
80	resource=resource_type,
81	service=service_name,
82	)
83
84
85	class ChangeSetResourceSupportChecker(ChangeSetModelVisitor):	1✔
86	change_set_type: ChangeSetType	1✔
87	catalog: CatalogPlugin	1✔
88
89	TITLE_MESSAGE = "Unsupported resources detected:"	1✔
90
91	def __init__(self, change_set_type: ChangeSetType):	1✔
92	self._resource_failure_messages: dict[str, str] = {}	1✔
93	self.change_set_type = change_set_type	1✔
94	self.catalog = get_aws_catalog()	1✔
95
96	def visit_node_resource(self, node_resource: NodeResource):	1✔
97	resource_type = node_resource.type_.value	1✔
98	ignore_unsupported = should_ignore_unsupported_resource_type(	1✔
99	resource_type=resource_type, change_set_type=self.change_set_type
100	)
101
102	if resource_type not in self._resource_failure_messages and not ignore_unsupported:	1✔
103	if resource_type not in AWS_AVAILABLE_CFN_RESOURCES:	1✔
104	# Ignore non-AWS resources
105	pass	1✔
106	support_status = self._resource_support_status(resource_type)	1✔
107	if support_status == CloudFormationResourcesSupportAtRuntime.AVAILABLE:	1✔
108	pass	1✔
109	else:
110	failure_message = _build_resource_failure_message(resource_type, support_status)	1✔
111	self._resource_failure_messages[resource_type] = failure_message	1✔
112	super().visit_node_resource(node_resource)	1✔
113
114	def _resource_support_status(	1✔
115	self, resource_type: str
116	) -> AwsServicesSupportStatus \| CfnResourceSupportStatus:
117	service_name = _get_service_name(resource_type)	1✔
118	return self.catalog.get_cloudformation_resource_status(resource_type, service_name, True)	1✔
119
120	@property	1✔
121	def failure_messages(self) -> list[str]:	1✔
122	return list(self._resource_failure_messages.values())	1✔

localstack / localstack / 20476669022

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous