20014894797

Committed 06 Dec 2025 01:32PM UTC coverage: 86.869% (-0.04%) from 86.904%

Build # 20014894797

Build Type

push

github

Committed by

web-flow

Commit Message

CFn: handle updates with empty resource properties (#13471)

Run Details

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

419 existing lines in 20 files now uncovered.

69891 of 80456 relevant lines covered (86.87%)

0.87 hits per line

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

92.05

/localstack-core/localstack/utils/catalog/catalog.py

import logging
from abc import abstractmethod
from typing import TypeAlias

from plux import Plugin

from localstack.services.cloudformation.resource_provider import (
    plugin_manager as cfn_plugin_manager,
)
from localstack.utils.catalog.catalog_loader import RemoteCatalogLoader
from localstack.utils.catalog.common import (
    AwsServiceOperationsSupportInLatest,
    AwsServicesSupportInLatest,
    AwsServiceSupportAtRuntime,
    CloudFormationResourcesSupportAtRuntime,
    CloudFormationResourcesSupportInLatest,
    LocalstackEmulatorType,
)

ServiceName = str
ServiceOperations = set[str]
ProviderName = str
CfnResourceName = str
CfnResourceMethodName = str
AwsServicesSupportStatus: TypeAlias = (
    AwsServiceSupportAtRuntime | AwsServicesSupportInLatest | AwsServiceOperationsSupportInLatest
)
CfnResourceSupportStatus: TypeAlias = (
    CloudFormationResourcesSupportInLatest | CloudFormationResourcesSupportAtRuntime
)
CfnResourceCatalog = dict[LocalstackEmulatorType, dict[CfnResourceName, set[CfnResourceMethodName]]]

LOG = logging.getLogger(__name__)


class CatalogPlugin(Plugin):
    namespace = "localstack.utils.catalog"

    @staticmethod
    def _get_cfn_resources_catalog(cloudformation_resources: dict) -> CfnResourceCatalog:
        cfn_resources_catalog = {}
        for emulator_type, resources in cloudformation_resources.items():
            cfn_resources_catalog[emulator_type] = {}
            for resource_name, resource in resources.items():
                cfn_resources_catalog[emulator_type][resource_name] = set(resource.methods)
        return cfn_resources_catalog

    @staticmethod
    def _get_services_at_runtime() -> set[ServiceName]:
        from localstack.services.plugins import SERVICE_PLUGINS

        return set(SERVICE_PLUGINS.list_available())

    @staticmethod
    def _get_cfn_resources_available_at_runtime() -> set[CfnResourceName]:
        return set(cfn_plugin_manager.list_names())

    @abstractmethod
    def get_aws_service_status(
        self, service_name: str, operation_name: str | None = None
    ) -> AwsServicesSupportStatus | None:
        pass

    @abstractmethod
    def get_cloudformation_resource_status(
        self, resource_name: str, service_name: str, is_pro_resource: bool = False
    ) -> CfnResourceSupportStatus | AwsServicesSupportInLatest | None:
        pass


class AwsCatalogRuntimePlugin(CatalogPlugin):
    name = "aws-catalog-runtime-only"

    def get_aws_service_status(
        self, service_name: str, operation_name: str | None = None
    ) -> AwsServicesSupportStatus | None:
        return None

    def get_cloudformation_resource_status(
        self, resource_name: str, service_name: str, is_pro_resource: bool = False
    ) -> CfnResourceSupportStatus | AwsServicesSupportInLatest | None:
        return None


class AwsCatalogRemoteStatePlugin(CatalogPlugin):
    name = "aws-catalog-remote-state"
    current_emulator_type: LocalstackEmulatorType = LocalstackEmulatorType.COMMUNITY
    services_in_latest: dict[ServiceName, dict[LocalstackEmulatorType, ServiceOperations]] = {}
    services_at_runtime: set[ServiceName] = set()
    cfn_resources_in_latest: CfnResourceCatalog = {}
    cfn_resources_at_runtime: set[CfnResourceName] = set()

    def __init__(self, remote_catalog_loader: RemoteCatalogLoader | None = None) -> None:
        catalog_loader = remote_catalog_loader or RemoteCatalogLoader()
        remote_catalog = catalog_loader.get_remote_catalog()
        for service_name, emulators in remote_catalog.services.items():
            for emulator_type, service_provider in emulators.items():
                self.services_in_latest.setdefault(service_name, {})[emulator_type] = set(
                    service_provider.operations
                )

        self.cfn_resources_in_latest = self._get_cfn_resources_catalog(
            remote_catalog.cloudformation_resources
        )
        self.cfn_resources_at_runtime = self._get_cfn_resources_available_at_runtime()
        self.services_at_runtime = self._get_services_at_runtime()

    def get_aws_service_status(
        self, service_name: str, operation_name: str | None = None
    ) -> AwsServicesSupportStatus | None:
        if not self.services_in_latest:
            return None
        if service_name not in self.services_in_latest:
            return AwsServicesSupportInLatest.NOT_SUPPORTED
        if self.current_emulator_type not in self.services_in_latest[service_name]:
            return AwsServicesSupportInLatest.SUPPORTED_WITH_LICENSE_UPGRADE
        if not operation_name:
            return AwsServicesSupportInLatest.SUPPORTED
        if operation_name in self.services_in_latest[service_name][self.current_emulator_type]:
            return AwsServiceOperationsSupportInLatest.SUPPORTED
        for emulator_type in self.services_in_latest[service_name]:
            if emulator_type is self.current_emulator_type:
                continue
            if operation_name in self.services_in_latest[service_name][emulator_type]:
                return AwsServiceOperationsSupportInLatest.SUPPORTED_WITH_LICENSE_UPGRADE
        return AwsServiceOperationsSupportInLatest.NOT_SUPPORTED

    def get_cloudformation_resource_status(
        self, resource_name: str, service_name: str, is_pro_resource: bool = False
    ) -> CfnResourceSupportStatus | AwsServicesSupportInLatest | None:
        if resource_name in self.cfn_resources_at_runtime:
            return CloudFormationResourcesSupportAtRuntime.AVAILABLE
        if service_name in self.services_at_runtime:
            if resource_name in self.cfn_resources_in_latest[self.current_emulator_type]:
                return CloudFormationResourcesSupportInLatest.SUPPORTED
            else:
                return CloudFormationResourcesSupportInLatest.NOT_SUPPORTED
        if service_name in self.services_in_latest:
            return self.get_aws_service_status(service_name, operation_name=None)
        return AwsServicesSupportInLatest.NOT_SUPPORTED

1	import logging	1✔
2	from abc import abstractmethod	1✔
3	from typing import TypeAlias	1✔
4
5	from plux import Plugin	1✔
6
7	from localstack.services.cloudformation.resource_provider import (	1✔
8	plugin_manager as cfn_plugin_manager,
9	)
10	from localstack.utils.catalog.catalog_loader import RemoteCatalogLoader	1✔
11	from localstack.utils.catalog.common import (	1✔
12	AwsServiceOperationsSupportInLatest,
13	AwsServicesSupportInLatest,
14	AwsServiceSupportAtRuntime,
15	CloudFormationResourcesSupportAtRuntime,
16	CloudFormationResourcesSupportInLatest,
17	LocalstackEmulatorType,
18	)
19
20	ServiceName = str	1✔
21	ServiceOperations = set[str]	1✔
22	ProviderName = str	1✔
23	CfnResourceName = str	1✔
24	CfnResourceMethodName = str	1✔
25	AwsServicesSupportStatus: TypeAlias = (	1✔
26	AwsServiceSupportAtRuntime \| AwsServicesSupportInLatest \| AwsServiceOperationsSupportInLatest
27	)
28	CfnResourceSupportStatus: TypeAlias = (	1✔
29	CloudFormationResourcesSupportInLatest \| CloudFormationResourcesSupportAtRuntime
30	)
31	CfnResourceCatalog = dict[LocalstackEmulatorType, dict[CfnResourceName, set[CfnResourceMethodName]]]	1✔
32
33	LOG = logging.getLogger(__name__)	1✔
34
35
36	class CatalogPlugin(Plugin):	1✔
37	namespace = "localstack.utils.catalog"	1✔
38
39	@staticmethod	1✔
40	def _get_cfn_resources_catalog(cloudformation_resources: dict) -> CfnResourceCatalog:	1✔
41	cfn_resources_catalog = {}	1✔
42	for emulator_type, resources in cloudformation_resources.items():	1✔
43	cfn_resources_catalog[emulator_type] = {}	1✔
44	for resource_name, resource in resources.items():	1✔
45	cfn_resources_catalog[emulator_type][resource_name] = set(resource.methods)	1✔
46	return cfn_resources_catalog	1✔
47
48	@staticmethod	1✔
49	def _get_services_at_runtime() -> set[ServiceName]:	1✔
50	from localstack.services.plugins import SERVICE_PLUGINS	1✔
51
52	return set(SERVICE_PLUGINS.list_available())	1✔
53
54	@staticmethod	1✔
55	def _get_cfn_resources_available_at_runtime() -> set[CfnResourceName]:	1✔
56	return set(cfn_plugin_manager.list_names())	1✔
57
58	@abstractmethod	1✔
59	def get_aws_service_status(	1✔
60	self, service_name: str, operation_name: str \| None = None
61	) -> AwsServicesSupportStatus \| None:
UNCOV 62	pass	×
63
64	@abstractmethod	1✔
65	def get_cloudformation_resource_status(	1✔
66	self, resource_name: str, service_name: str, is_pro_resource: bool = False
67	) -> CfnResourceSupportStatus \| AwsServicesSupportInLatest \| None:
UNCOV 68	pass	×
69
70
71	class AwsCatalogRuntimePlugin(CatalogPlugin):	1✔
72	name = "aws-catalog-runtime-only"	1✔
73
74	def get_aws_service_status(	1✔
75	self, service_name: str, operation_name: str \| None = None
76	) -> AwsServicesSupportStatus \| None:
UNCOV 77	return None	×
78
79	def get_cloudformation_resource_status(	1✔
80	self, resource_name: str, service_name: str, is_pro_resource: bool = False
81	) -> CfnResourceSupportStatus \| AwsServicesSupportInLatest \| None:
UNCOV 82	return None	×
83
84
85	class AwsCatalogRemoteStatePlugin(CatalogPlugin):	1✔
86	name = "aws-catalog-remote-state"	1✔
87	current_emulator_type: LocalstackEmulatorType = LocalstackEmulatorType.COMMUNITY	1✔
88	services_in_latest: dict[ServiceName, dict[LocalstackEmulatorType, ServiceOperations]] = {}	1✔
89	services_at_runtime: set[ServiceName] = set()	1✔
90	cfn_resources_in_latest: CfnResourceCatalog = {}	1✔
91	cfn_resources_at_runtime: set[CfnResourceName] = set()	1✔
92
93	def __init__(self, remote_catalog_loader: RemoteCatalogLoader \| None = None) -> None:	1✔
94	catalog_loader = remote_catalog_loader or RemoteCatalogLoader()	1✔
95	remote_catalog = catalog_loader.get_remote_catalog()	1✔
96	for service_name, emulators in remote_catalog.services.items():	1✔
97	for emulator_type, service_provider in emulators.items():	1✔
98	self.services_in_latest.setdefault(service_name, {})[emulator_type] = set(	1✔
99	service_provider.operations
100	)
101
102	self.cfn_resources_in_latest = self._get_cfn_resources_catalog(	1✔
103	remote_catalog.cloudformation_resources
104	)
105	self.cfn_resources_at_runtime = self._get_cfn_resources_available_at_runtime()	1✔
106	self.services_at_runtime = self._get_services_at_runtime()	1✔
107
108	def get_aws_service_status(	1✔
109	self, service_name: str, operation_name: str \| None = None
110	) -> AwsServicesSupportStatus \| None:
111	if not self.services_in_latest:	1✔
UNCOV 112	return None	×
113	if service_name not in self.services_in_latest:	1✔
114	return AwsServicesSupportInLatest.NOT_SUPPORTED	1✔
115	if self.current_emulator_type not in self.services_in_latest[service_name]:	1✔
116	return AwsServicesSupportInLatest.SUPPORTED_WITH_LICENSE_UPGRADE	1✔
117	if not operation_name:	1✔
118	return AwsServicesSupportInLatest.SUPPORTED	1✔
119	if operation_name in self.services_in_latest[service_name][self.current_emulator_type]:	1✔
120	return AwsServiceOperationsSupportInLatest.SUPPORTED	1✔
121	for emulator_type in self.services_in_latest[service_name]:	1✔
122	if emulator_type is self.current_emulator_type:	1✔
UNCOV 123	continue	×
124	if operation_name in self.services_in_latest[service_name][emulator_type]:	1✔
125	return AwsServiceOperationsSupportInLatest.SUPPORTED_WITH_LICENSE_UPGRADE	1✔
126	return AwsServiceOperationsSupportInLatest.NOT_SUPPORTED	1✔
127
128	def get_cloudformation_resource_status(	1✔
129	self, resource_name: str, service_name: str, is_pro_resource: bool = False
130	) -> CfnResourceSupportStatus \| AwsServicesSupportInLatest \| None:
131	if resource_name in self.cfn_resources_at_runtime:	1✔
132	return CloudFormationResourcesSupportAtRuntime.AVAILABLE	1✔
133	if service_name in self.services_at_runtime:	1✔
134	if resource_name in self.cfn_resources_in_latest[self.current_emulator_type]:	1✔
UNCOV 135	return CloudFormationResourcesSupportInLatest.SUPPORTED	×
136	else:
137	return CloudFormationResourcesSupportInLatest.NOT_SUPPORTED	1✔
138	if service_name in self.services_in_latest:	1✔
139	return self.get_aws_service_status(service_name, operation_name=None)	1✔
140	return AwsServicesSupportInLatest.NOT_SUPPORTED	1✔

localstack / localstack / 20014894797

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