16820655284

Committed 07 Aug 2025 05:03PM UTC coverage: 86.841% (-0.05%) from 86.892%

Build # 16820655284

Build Type

push

github

Committed by

web-flow

Commit Message

CFNV2: support CDK bootstrap and deployment (#12967)

Coverage Stats

32 of 38 new or added lines in 5 files covered. (84.21%)

2013 existing lines in 125 files now uncovered.

66606 of 76699 relevant lines covered (86.84%)

0.87 hits per line

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

94.44

/localstack-core/localstack/runtime/shutdown.py

import logging
from collections.abc import Callable
from typing import Any

from localstack.runtime import hooks
from localstack.utils.functions import call_safe

LOG = logging.getLogger(__name__)

SERVICE_SHUTDOWN_PRIORITY = -10
"""Shutdown hook priority for shutting down service plugins."""


class ShutdownHandlers:
    """
    Register / unregister shutdown handlers. All registered shutdown handlers should execute as fast as possible.
    Blocking shutdown handlers will block infra shutdown.
    """

    def __init__(self):
        self._callbacks = []

    def register(self, shutdown_handler: Callable[[], Any]) -> None:
        """
        Register shutdown handler. Handler should not block or take more than a couple seconds.

        :param shutdown_handler: Callable without parameters
        """
        self._callbacks.append(shutdown_handler)

    def unregister(self, shutdown_handler: Callable[[], Any]) -> None:
        """
        Unregister a handler. Idempotent operation.

        :param shutdown_handler: Shutdown handler which was previously registered
        """
        try:
            self._callbacks.remove(shutdown_handler)
        except ValueError:
            pass

    def run(self) -> None:
        """
        Execute shutdown handlers in reverse order of registration.
        Should only be called once, on shutdown.
        """
        for callback in reversed(list(self._callbacks)):
            call_safe(callback)


SHUTDOWN_HANDLERS = ShutdownHandlers()
"""Shutdown handlers run with default priority in an on_infra_shutdown hook."""

ON_AFTER_SERVICE_SHUTDOWN_HANDLERS = ShutdownHandlers()
"""Shutdown handlers that are executed after all services have been shut down."""


@hooks.on_infra_shutdown()
def run_shutdown_handlers():
    SHUTDOWN_HANDLERS.run()


@hooks.on_infra_shutdown(priority=SERVICE_SHUTDOWN_PRIORITY)
def shutdown_services():
    # TODO: this belongs into the shutdown procedure of a `Platform` or `RuntimeContainer` class.
    from localstack.services.plugins import SERVICE_PLUGINS

    LOG.info("[shutdown] Stopping all services")
    SERVICE_PLUGINS.stop_all_services()


@hooks.on_infra_shutdown(priority=SERVICE_SHUTDOWN_PRIORITY - 10)
def run_on_after_service_shutdown_handlers():
    ON_AFTER_SERVICE_SHUTDOWN_HANDLERS.run()

1	import logging	1✔
2	from collections.abc import Callable	1✔
3	from typing import Any	1✔
4
5	from localstack.runtime import hooks	1✔
6	from localstack.utils.functions import call_safe	1✔
7
8	LOG = logging.getLogger(__name__)	1✔
9
10	SERVICE_SHUTDOWN_PRIORITY = -10	1✔
11	"""Shutdown hook priority for shutting down service plugins."""	1✔
12
13
14	class ShutdownHandlers:	1✔
15	"""
16	Register / unregister shutdown handlers. All registered shutdown handlers should execute as fast as possible.
17	Blocking shutdown handlers will block infra shutdown.
18	"""
19
20	def __init__(self):	1✔
21	self._callbacks = []	1✔
22
23	def register(self, shutdown_handler: Callable[[], Any]) -> None:	1✔
24	"""
25	Register shutdown handler. Handler should not block or take more than a couple seconds.
26
27	:param shutdown_handler: Callable without parameters
28	"""
29	self._callbacks.append(shutdown_handler)	1✔
30
31	def unregister(self, shutdown_handler: Callable[[], Any]) -> None:	1✔
32	"""
33	Unregister a handler. Idempotent operation.
34
35	:param shutdown_handler: Shutdown handler which was previously registered
36	"""
37	try:	1✔
38	self._callbacks.remove(shutdown_handler)	1✔
39	except ValueError:	×
UNCOV 40	pass	×
41
42	def run(self) -> None:	1✔
43	"""
44	Execute shutdown handlers in reverse order of registration.
45	Should only be called once, on shutdown.
46	"""
47	for callback in reversed(list(self._callbacks)):	1✔
48	call_safe(callback)	1✔
49
50
51	SHUTDOWN_HANDLERS = ShutdownHandlers()	1✔
52	"""Shutdown handlers run with default priority in an on_infra_shutdown hook."""	1✔
53
54	ON_AFTER_SERVICE_SHUTDOWN_HANDLERS = ShutdownHandlers()	1✔
55	"""Shutdown handlers that are executed after all services have been shut down."""	1✔
56
57
58	@hooks.on_infra_shutdown()	1✔
59	def run_shutdown_handlers():	1✔
60	SHUTDOWN_HANDLERS.run()	1✔
61
62
63	@hooks.on_infra_shutdown(priority=SERVICE_SHUTDOWN_PRIORITY)	1✔
64	def shutdown_services():	1✔
65	# TODO: this belongs into the shutdown procedure of a `Platform` or `RuntimeContainer` class.
66	from localstack.services.plugins import SERVICE_PLUGINS	1✔
67
68	LOG.info("[shutdown] Stopping all services")	1✔
69	SERVICE_PLUGINS.stop_all_services()	1✔
70
71
72	@hooks.on_infra_shutdown(priority=SERVICE_SHUTDOWN_PRIORITY - 10)	1✔
73	def run_on_after_service_shutdown_handlers():	1✔
74	ON_AFTER_SERVICE_SHUTDOWN_HANDLERS.run()	1✔

localstack / localstack / 16820655284

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