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

95.61

/localstack-core/localstack/aws/skeleton.py

import inspect
import logging
from collections.abc import Callable
from typing import Any, NamedTuple

from botocore import xform_name
from botocore.model import ServiceModel

from localstack.aws.api import (
    CommonServiceException,
    RequestContext,
    ServiceException,
)
from localstack.aws.api.core import ServiceRequest, ServiceRequestHandler, ServiceResponse
from localstack.aws.catalog_exceptions import get_service_availability_exception
from localstack.aws.protocol.parser import create_parser
from localstack.aws.protocol.serializer import ResponseSerializer, create_serializer
from localstack.aws.spec import load_service
from localstack.http import Response
from localstack.utils import analytics
from localstack.utils.catalog.plugins import get_aws_catalog

LOG = logging.getLogger(__name__)

DispatchTable = dict[str, ServiceRequestHandler]


def create_skeleton(service: str | ServiceModel, delegate: Any):
    if isinstance(service, str):
        service = load_service(service)

    return Skeleton(service, create_dispatch_table(delegate))


class HandlerAttributes(NamedTuple):
    """
    Holder object of the attributes added to a function by the @handler decorator.
    """

    function_name: str
    operation: str
    pass_context: bool
    expand_parameters: bool


def create_dispatch_table(delegate: object) -> DispatchTable:
    """
    Creates a dispatch table for a given object. First, the entire class tree of the object is scanned to find any
    functions that are decorated with @handler. It then resolves those functions on the delegate.
    """
    # scan class tree for @handler wrapped functions (reverse class tree so that inherited functions overwrite parent
    # functions)
    cls_tree = inspect.getmro(delegate.__class__)
    handlers: dict[str, HandlerAttributes] = {}
    cls_tree = reversed(list(cls_tree))
    for cls in cls_tree:
        if cls is object:
            continue

        for name, fn in inspect.getmembers(cls, inspect.isfunction):
            try:
                # attributes come from operation_marker in @handler wrapper
                handlers[fn.operation] = HandlerAttributes(
                    fn.__name__, fn.operation, fn.pass_context, fn.expand_parameters
                )
            except AttributeError:
                pass

    # create dispatch table from operation handlers by resolving bound functions on the delegate
    dispatch_table: DispatchTable = {}
    for handler in handlers.values():
        # resolve the bound function of the delegate
        bound_function = getattr(delegate, handler.function_name)
        # create a dispatcher
        dispatch_table[handler.operation] = ServiceRequestDispatcher(
            bound_function,
            operation=handler.operation,
            pass_context=handler.pass_context,
            expand_parameters=handler.expand_parameters,
        )

    return dispatch_table


class ServiceRequestDispatcher:
    fn: Callable
    operation: str
    expand_parameters: bool = True
    pass_context: bool = True

    def __init__(
        self,
        fn: Callable,
        operation: str,
        pass_context: bool = True,
        expand_parameters: bool = True,
    ):
        self.fn = fn
        self.operation = operation
        self.pass_context = pass_context
        self.expand_parameters = expand_parameters

    def __call__(self, context: RequestContext, request: ServiceRequest) -> ServiceResponse | None:
        args = []
        kwargs = {}

        if not self.expand_parameters:
            if self.pass_context:
                args.append(context)
            args.append(request)
        else:
            if request is None:
                kwargs = {}
            else:
                kwargs = {xform_name(k): v for k, v in request.items()}
            kwargs["context"] = context

        return self.fn(*args, **kwargs)


class Skeleton:
    service: ServiceModel
    dispatch_table: DispatchTable

    def __init__(self, service: ServiceModel, implementation: Any | DispatchTable):
        self.service = service

        if isinstance(implementation, dict):
            self.dispatch_table = implementation
        else:
            self.dispatch_table = create_dispatch_table(implementation)

    def invoke(self, context: RequestContext) -> Response:
        serializer = create_serializer(context.service, context.protocol)

        if context.operation and context.service_request:
            # if the parsed request is already set in the context, re-use them
            operation, instance = context.operation, context.service_request
        else:
            # otherwise, parse the incoming HTTPRequest
            operation, instance = create_parser(context.service, context.protocol).parse(
                context.request
            )
            context.operation = operation

        try:
            # Find the operation's handler in the dispatch table
            if operation.name not in self.dispatch_table:
                LOG.warning(
                    "missing entry in dispatch table for %s.%s",
                    self.service.service_name,
                    operation.name,
                )
                raise NotImplementedError

            return self.dispatch_request(serializer, context, instance)
        except ServiceException as e:
            return self.on_service_exception(serializer, context, e)
        except NotImplementedError as e:
            return self.on_not_implemented_error(serializer, context, e)

    def dispatch_request(
        self, serializer: ResponseSerializer, context: RequestContext, instance: ServiceRequest
    ) -> Response:
        operation = context.operation

        handler = self.dispatch_table[operation.name]

        # Call the appropriate handler
        result = handler(context, instance) or {}

        # if the service handler returned an HTTP request, forego serialization and return immediately
        if isinstance(result, Response):
            return result

        context.service_response = result

        # Serialize result dict to a Response and return it
        return serializer.serialize_to_response(
            result, operation, context.request.headers, context.request_id
        )

    def on_service_exception(
        self, serializer: ResponseSerializer, context: RequestContext, exception: ServiceException
    ) -> Response:
        """
        Called by invoke if the handler of the operation raised a ServiceException.

        :param serializer: serializer which should be used to serialize the exception
        :param context: the request context
        :param exception: the exception that was raised
        :return: a Response object
        """
        context.service_exception = exception

        return serializer.serialize_error_to_response(
            exception, context.operation, context.request.headers, context.request_id
        )

    def on_not_implemented_error(
        self,
        serializer: ResponseSerializer,
        context: RequestContext,
        exception: NotImplementedError,
    ) -> Response:
        """
        Called by invoke if either the dispatch table did not contain an entry for the operation, or the service
        provider raised a NotImplementedError
        :param serializer: the serialzier which should be used to serialize the NotImplementedError
        :param context: the request context
        :param exception: the NotImplementedError that was raised
        :return: a Response object
        """
        operation = context.operation

        operation_name = operation.name
        service_name = operation.service_model.service_name
        exception_message: str | None = exception.args[0] if exception.args else None
        if exception_message is not None:
            message = exception_message
            error = CommonServiceException("InternalFailure", message, status_code=501)
            # record event
            analytics.log.event(
                "services_notimplemented", payload={"s": service_name, "a": operation_name}
            )
        else:
            service_status = get_aws_catalog().get_aws_service_status(service_name, operation_name)
            error = get_service_availability_exception(service_name, operation_name, service_status)
            message = error.message
            analytics.log.event(
                "services_notimplemented",
                payload={
                    "s": service_name,
                    "a": operation_name,
                    "c": error.error_code,
                },
            )

        LOG.info(message)
        context.service_exception = error

        return serializer.serialize_error_to_response(
            error, operation, context.request.headers, context.request_id
        )

1	import inspect	1✔
2	import logging	1✔
3	from collections.abc import Callable	1✔
4	from typing import Any, NamedTuple	1✔
5
6	from botocore import xform_name	1✔
7	from botocore.model import ServiceModel	1✔
8
9	from localstack.aws.api import (	1✔
10	CommonServiceException,
11	RequestContext,
12	ServiceException,
13	)
14	from localstack.aws.api.core import ServiceRequest, ServiceRequestHandler, ServiceResponse	1✔
15	from localstack.aws.catalog_exceptions import get_service_availability_exception	1✔
16	from localstack.aws.protocol.parser import create_parser	1✔
17	from localstack.aws.protocol.serializer import ResponseSerializer, create_serializer	1✔
18	from localstack.aws.spec import load_service	1✔
19	from localstack.http import Response	1✔
20	from localstack.utils import analytics	1✔
21	from localstack.utils.catalog.plugins import get_aws_catalog	1✔
22
23	LOG = logging.getLogger(__name__)	1✔
24
25	DispatchTable = dict[str, ServiceRequestHandler]	1✔
26
27
28	def create_skeleton(service: str \| ServiceModel, delegate: Any):	1✔
29	if isinstance(service, str):	×
UNCOV 30	service = load_service(service)	×
31
UNCOV 32	return Skeleton(service, create_dispatch_table(delegate))	×
33
34
35	class HandlerAttributes(NamedTuple):	1✔
36	"""
37	Holder object of the attributes added to a function by the @handler decorator.
38	"""
39
40	function_name: str	1✔
41	operation: str	1✔
42	pass_context: bool	1✔
43	expand_parameters: bool	1✔
44
45
46	def create_dispatch_table(delegate: object) -> DispatchTable:	1✔
47	"""
48	Creates a dispatch table for a given object. First, the entire class tree of the object is scanned to find any
49	functions that are decorated with @handler. It then resolves those functions on the delegate.
50	"""
51	# scan class tree for @handler wrapped functions (reverse class tree so that inherited functions overwrite parent
52	# functions)
53	cls_tree = inspect.getmro(delegate.__class__)	1✔
54	handlers: dict[str, HandlerAttributes] = {}	1✔
55	cls_tree = reversed(list(cls_tree))	1✔
56	for cls in cls_tree:	1✔
57	if cls is object:	1✔
58	continue	1✔
59
60	for name, fn in inspect.getmembers(cls, inspect.isfunction):	1✔
61	try:	1✔
62	# attributes come from operation_marker in @handler wrapper
63	handlers[fn.operation] = HandlerAttributes(	1✔
64	fn.__name__, fn.operation, fn.pass_context, fn.expand_parameters
65	)
66	except AttributeError:	1✔
67	pass	1✔
68
69	# create dispatch table from operation handlers by resolving bound functions on the delegate
70	dispatch_table: DispatchTable = {}	1✔
71	for handler in handlers.values():	1✔
72	# resolve the bound function of the delegate
73	bound_function = getattr(delegate, handler.function_name)	1✔
74	# create a dispatcher
75	dispatch_table[handler.operation] = ServiceRequestDispatcher(	1✔
76	bound_function,
77	operation=handler.operation,
78	pass_context=handler.pass_context,
79	expand_parameters=handler.expand_parameters,
80	)
81
82	return dispatch_table	1✔
83
84
85	class ServiceRequestDispatcher:	1✔
86	fn: Callable	1✔
87	operation: str	1✔
88	expand_parameters: bool = True	1✔
89	pass_context: bool = True	1✔
90
91	def __init__(	1✔
92	self,
93	fn: Callable,
94	operation: str,
95	pass_context: bool = True,
96	expand_parameters: bool = True,
97	):
98	self.fn = fn	1✔
99	self.operation = operation	1✔
100	self.pass_context = pass_context	1✔
101	self.expand_parameters = expand_parameters	1✔
102
103	def __call__(self, context: RequestContext, request: ServiceRequest) -> ServiceResponse \| None:	1✔
104	args = []	1✔
105	kwargs = {}	1✔
106
107	if not self.expand_parameters:	1✔
108	if self.pass_context:	1✔
109	args.append(context)	1✔
110	args.append(request)	1✔
111	else:
112	if request is None:	1✔
UNCOV 113	kwargs = {}	×
114	else:
115	kwargs = {xform_name(k): v for k, v in request.items()}	1✔
116	kwargs["context"] = context	1✔
117
118	return self.fn(args, *kwargs)	1✔
119
120
121	class Skeleton:	1✔
122	service: ServiceModel	1✔
123	dispatch_table: DispatchTable	1✔
124
125	def __init__(self, service: ServiceModel, implementation: Any \| DispatchTable):	1✔
126	self.service = service	1✔
127
128	if isinstance(implementation, dict):	1✔
129	self.dispatch_table = implementation	1✔
130	else:
131	self.dispatch_table = create_dispatch_table(implementation)	1✔
132
133	def invoke(self, context: RequestContext) -> Response:	1✔
134	serializer = create_serializer(context.service, context.protocol)	1✔
135
136	if context.operation and context.service_request:	1✔
137	# if the parsed request is already set in the context, re-use them
138	operation, instance = context.operation, context.service_request	1✔
139	else:
140	# otherwise, parse the incoming HTTPRequest
141	operation, instance = create_parser(context.service, context.protocol).parse(	1✔
142	context.request
143	)
144	context.operation = operation	1✔
145
146	try:	1✔
147	# Find the operation's handler in the dispatch table
148	if operation.name not in self.dispatch_table:	1✔
149	LOG.warning(	1✔
150	"missing entry in dispatch table for %s.%s",
151	self.service.service_name,
152	operation.name,
153	)
154	raise NotImplementedError
155
156	return self.dispatch_request(serializer, context, instance)	1✔
157	except ServiceException as e:	1✔
158	return self.on_service_exception(serializer, context, e)	1✔
159	except NotImplementedError as e:	1✔
160	return self.on_not_implemented_error(serializer, context, e)	1✔
161
162	def dispatch_request(	1✔
163	self, serializer: ResponseSerializer, context: RequestContext, instance: ServiceRequest
164	) -> Response:
165	operation = context.operation	1✔
166
167	handler = self.dispatch_table[operation.name]	1✔
168
169	# Call the appropriate handler
170	result = handler(context, instance) or {}	1✔
171
172	# if the service handler returned an HTTP request, forego serialization and return immediately
173	if isinstance(result, Response):	1✔
UNCOV 174	return result	×
175
176	context.service_response = result	1✔
177
178	# Serialize result dict to a Response and return it
179	return serializer.serialize_to_response(	1✔
180	result, operation, context.request.headers, context.request_id
181	)
182
183	def on_service_exception(	1✔
184	self, serializer: ResponseSerializer, context: RequestContext, exception: ServiceException
185	) -> Response:
186	"""
187	Called by invoke if the handler of the operation raised a ServiceException.
188
189	:param serializer: serializer which should be used to serialize the exception
190	:param context: the request context
191	:param exception: the exception that was raised
192	:return: a Response object
193	"""
194	context.service_exception = exception	1✔
195
196	return serializer.serialize_error_to_response(	1✔
197	exception, context.operation, context.request.headers, context.request_id
198	)
199
200	def on_not_implemented_error(	1✔
201	self,
202	serializer: ResponseSerializer,
203	context: RequestContext,
204	exception: NotImplementedError,
205	) -> Response:
206	"""
207	Called by invoke if either the dispatch table did not contain an entry for the operation, or the service
208	provider raised a NotImplementedError
209	:param serializer: the serialzier which should be used to serialize the NotImplementedError
210	:param context: the request context
211	:param exception: the NotImplementedError that was raised
212	:return: a Response object
213	"""
214	operation = context.operation	1✔
215
216	operation_name = operation.name	1✔
217	service_name = operation.service_model.service_name	1✔
218	exception_message: str \| None = exception.args[0] if exception.args else None	1✔
219	if exception_message is not None:	1✔
220	message = exception_message	1✔
221	error = CommonServiceException("InternalFailure", message, status_code=501)	1✔
222	# record event
223	analytics.log.event(	1✔
224	"services_notimplemented", payload={"s": service_name, "a": operation_name}
225	)
226	else:
227	service_status = get_aws_catalog().get_aws_service_status(service_name, operation_name)	1✔
228	error = get_service_availability_exception(service_name, operation_name, service_status)	1✔
229	message = error.message	1✔
230	analytics.log.event(	1✔
231	"services_notimplemented",
232	payload={
233	"s": service_name,
234	"a": operation_name,
235	"c": error.error_code,
236	},
237	)
238
239	LOG.info(message)	1✔
240	context.service_exception = error	1✔
241
242	return serializer.serialize_error_to_response(	1✔
243	error, operation, context.request.headers, context.request_id
244	)

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