a39cb164-6c44-4f53-92bc-6ed8705a075c

Committed 05 May 2025 12:45PM UTC coverage: 86.555% (+0.02%) from 86.54%

Build # a39cb164-6c44-4f53-92bc-6ed8705a075c

Build Type

push

circleci

Committed by

web-flow

Commit Message

Switch to using `kclpy-ext` (#12567)

Coverage Stats

64183 of 74153 relevant lines covered (86.55%)

0.87 hits per line

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

87.88

/localstack-core/localstack/aws/handlers/internal.py

"""Handler for routing internal localstack resources under /_localstack."""

import logging

from werkzeug.exceptions import NotFound

from localstack import constants
from localstack.http import Response
from localstack.runtime import events
from localstack.services.internal import LocalstackResources

from ..api import RequestContext
from ..chain import Handler, HandlerChain

LOG = logging.getLogger(__name__)


class LocalstackResourceHandler(Handler):
    """
    Adapter to serve LocalstackResources as a Handler.
    """

    resources: LocalstackResources

    def __init__(self, resources: LocalstackResources = None) -> None:
        from localstack.services.internal import get_internal_apis

        self.resources = resources or get_internal_apis()

    def __call__(self, chain: HandlerChain, context: RequestContext, response: Response):
        try:
            # serve
            response.update_from(self.resources.dispatch(context.request))
            chain.stop()
        except NotFound:
            path = context.request.path
            if path.startswith(constants.INTERNAL_RESOURCE_PATH + "/"):
                # only return 404 if we're accessing an internal resource, otherwise fall back to the other handlers
                LOG.warning("Unable to find resource handler for path: %s", path)
                chain.respond(404)


class RuntimeShutdownHandler(Handler):
    def __call__(self, chain: HandlerChain, context: RequestContext, response: Response):
        if events.infra_stopped.is_set():
            chain.respond(503)
        elif events.infra_stopping.is_set():
            # if we're in the process of shutting down the infrastructure, only accept internal calls, or calls to
            # internal APIs
            if context.is_internal_call:
                return
            if context.request.path.startswith("/_localstack"):
                return
            chain.respond(503)

1	"""Handler for routing internal localstack resources under /_localstack."""
2
3	import logging	1✔
4
5	from werkzeug.exceptions import NotFound	1✔
6
7	from localstack import constants	1✔
8	from localstack.http import Response	1✔
9	from localstack.runtime import events	1✔
10	from localstack.services.internal import LocalstackResources	1✔
11
12	from ..api import RequestContext	1✔
13	from ..chain import Handler, HandlerChain	1✔
14
15	LOG = logging.getLogger(__name__)	1✔
16
17
18	class LocalstackResourceHandler(Handler):	1✔
19	"""
20	Adapter to serve LocalstackResources as a Handler.
21	"""
22
23	resources: LocalstackResources	1✔
24
25	def __init__(self, resources: LocalstackResources = None) -> None:	1✔
26	from localstack.services.internal import get_internal_apis	1✔
27
28	self.resources = resources or get_internal_apis()	1✔
29
30	def __call__(self, chain: HandlerChain, context: RequestContext, response: Response):	1✔
31	try:	1✔
32	# serve
33	response.update_from(self.resources.dispatch(context.request))	1✔
34	chain.stop()	1✔
35	except NotFound:	1✔
36	path = context.request.path	1✔
37	if path.startswith(constants.INTERNAL_RESOURCE_PATH + "/"):	1✔
38	# only return 404 if we're accessing an internal resource, otherwise fall back to the other handlers
39	LOG.warning("Unable to find resource handler for path: %s", path)	1✔
40	chain.respond(404)	1✔
41
42
43	class RuntimeShutdownHandler(Handler):	1✔
44	def __call__(self, chain: HandlerChain, context: RequestContext, response: Response):	1✔
45	if events.infra_stopped.is_set():	1✔
46	chain.respond(503)	×
47	elif events.infra_stopping.is_set():	1✔
48	# if we're in the process of shutting down the infrastructure, only accept internal calls, or calls to
49	# internal APIs
50	if context.is_internal_call:	1✔
51	return	1✔
52	if context.request.path.startswith("/_localstack"):	×
53	return	×
54	chain.respond(503)	×

localstack / localstack / a39cb164-6c44-4f53-92bc-6ed8705a075c

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