21017785080

Committed 14 Jan 2026 09:54PM UTC coverage: 86.96% (+0.04%) from 86.923%

Build # 21017785080

Build Type

push

github

Committed by

web-flow

Commit Message

SNS: v2 enable skipped tests (#13544)

Run Details

15 of 15 new or added lines in 3 files covered. (100.0%)

150 existing lines in 7 files now uncovered.

70327 of 80873 relevant lines covered (86.96%)

0.87 hits per line

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

93.18

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

import logging
from abc import abstractmethod
from typing import TypeAlias

from plux import Plugin

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]:
        from localstack.services.cloudformation.resource_provider import (
            plugin_manager as cfn_plugin_manager,
        )

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

localstack / localstack / 21017785080

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