fb37c2a2-4bd7-49b4-891b-ede18134b4b1

Committed 20 Jan 2025 04:22PM UTC coverage: 86.826% (-0.02%) from 86.844%

Build # fb37c2a2-4bd7-49b4-891b-ede18134b4b1

Build Type

push

circleci

Committed by

web-flow

Commit Message

Add tests for KMS resource tagging and untagging (#12121)

Coverage Stats

61076 of 70343 relevant lines covered (86.83%)

0.87 hits per line

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

44.9

/localstack-core/localstack/utils/lambda_debug_mode/lambda_debug_mode_session.py

from __future__ import annotations

import logging
import os
import time
from threading import Event, Thread
from typing import Optional

from localstack.aws.api.lambda_ import Arn
from localstack.config import LAMBDA_DEBUG_MODE, LAMBDA_DEBUG_MODE_CONFIG_PATH
from localstack.utils.lambda_debug_mode.lambda_debug_mode_config import (
    LambdaDebugConfig,
    LambdaDebugModeConfig,
    load_lambda_debug_mode_config,
)
from localstack.utils.objects import singleton_factory

LOG = logging.getLogger(__name__)


class LambdaDebugModeSession:
    _is_lambda_debug_mode: bool

    _configuration_file_path: Optional[str]
    _watch_thread: Optional[Thread]
    _initialised_event: Optional[Event]
    _stop_event: Optional[Event]
    _config: Optional[LambdaDebugModeConfig]

    def __init__(self):
        self._is_lambda_debug_mode = bool(LAMBDA_DEBUG_MODE)

        # Disabled Lambda Debug Mode state initialisation.
        self._configuration_file_path = None
        self._watch_thread = None
        self._initialised_event = None
        self._stop_event = None
        self._config = None

        # Lambda Debug Mode is not enabled: leave as disabled state and return.
        if not self._is_lambda_debug_mode:
            return

        # Lambda Debug Mode is enabled.
        # Instantiate the configuration requirements if a configuration file is given.
        self._configuration_file_path = LAMBDA_DEBUG_MODE_CONFIG_PATH
        if not self._configuration_file_path:
            return

        # A configuration file path is given: initialised the resources to load and watch the file.

        # Signal and block on first loading to ensure this is enforced from the very first
        # invocation, as this module is not loaded at startup. The LambdaDebugModeConfigWatch
        # thread will then take care of updating the configuration periodically and asynchronously.
        # This may somewhat slow down the first upstream thread loading this module, but not
        # future calls. On the other hand, avoiding this mechanism means that first Lambda calls
        # occur with no Debug configuration.
        self._initialised_event = Event()

        # Signals when a shutdown signal from the application is registered.
        self._stop_event = Event()

        self._watch_thread = Thread(
            target=self._watch_logic, args=(), daemon=True, name="LambdaDebugModeConfigWatch"
        )
        self._watch_thread.start()

    @staticmethod
    @singleton_factory
    def get() -> LambdaDebugModeSession:
        """Returns a singleton instance of the Lambda Debug Mode session."""
        return LambdaDebugModeSession()

    def ensure_running(self) -> None:
        # Nothing to start.
        if self._watch_thread is None or self._watch_thread.is_alive():
            return
        try:
            self._watch_thread.start()
        except Exception as exception:
            exception_str = str(exception)
            # The thread was already restarted by another process.
            if (
                isinstance(exception, RuntimeError)
                and exception_str
                and "threads can only be started once" in exception_str
            ):
                return
            LOG.error(
                "Lambda Debug Mode could not restart the "
                "hot reloading of the configuration file, '%s'",
                exception_str,
            )

    def signal_stop(self) -> None:
        stop_event = self._stop_event
        if stop_event is not None:
            stop_event.set()

    def _load_lambda_debug_mode_config(self):
        yaml_configuration_string = None
        try:
            with open(self._configuration_file_path, "r") as df:
                yaml_configuration_string = df.read()
        except FileNotFoundError:
            LOG.error(
                "Error: The file lambda debug config file '%s' was not found.",
                self._configuration_file_path,
            )
        except IsADirectoryError:
            LOG.error(
                "Error: Expected a lambda debug config file but found a directory at '%s'.",
                self._configuration_file_path,
            )
        except PermissionError:
            LOG.error(
                "Error: Permission denied while trying to read the lambda debug config file '%s'.",
                self._configuration_file_path,
            )
        except Exception as ex:
            LOG.error(
                "Error: An unexpected error occurred while reading lambda debug config '%s': '%s'",
                self._configuration_file_path,
                ex,
            )
        if not yaml_configuration_string:
            return None

        self._config = load_lambda_debug_mode_config(yaml_configuration_string)
        if self._config is not None:
            LOG.info("Lambda Debug Mode is now enforcing the latest configuration.")
        else:
            LOG.warning(
                "Lambda Debug Mode could not load the latest configuration due to an error, "
                "check logs for more details."
            )

    def _config_file_epoch_last_modified_or_now(self) -> int:
        try:
            modified_time = os.path.getmtime(self._configuration_file_path)
            return int(modified_time)
        except Exception as e:
            LOG.warning("Lambda Debug Mode could not access the configuration file: %s", e)
            epoch_now = int(time.time())
            return epoch_now

    def _watch_logic(self) -> None:
        # TODO: consider relying on system calls (watchdog lib for cross-platform support)
        #  instead of monitoring last modified dates.
        # Run the first load and signal as initialised.
        epoch_last_loaded: int = self._config_file_epoch_last_modified_or_now()
        self._load_lambda_debug_mode_config()
        self._initialised_event.set()

        # Monitor for file changes whilst the application is running.
        while not self._stop_event.is_set():
            time.sleep(1)
            epoch_last_modified = self._config_file_epoch_last_modified_or_now()
            if epoch_last_modified > epoch_last_loaded:
                epoch_last_loaded = epoch_last_modified
                self._load_lambda_debug_mode_config()

    def _get_initialised_config(self) -> Optional[LambdaDebugModeConfig]:
        # Check the session is not initialising, and if so then wait for initialisation to finish.
        # Note: the initialisation event is otherwise left set since after first initialisation has terminated.
        if self._initialised_event is not None:
            self._initialised_event.wait()
        return self._config

    def is_lambda_debug_mode(self) -> bool:
        return self._is_lambda_debug_mode

    def debug_config_for(self, lambda_arn: Arn) -> Optional[LambdaDebugConfig]:
        config = self._get_initialised_config()
        return config.functions.get(lambda_arn) if config else None

1	from __future__ import annotations	1✔
2
3	import logging	1✔
4	import os	1✔
5	import time	1✔
6	from threading import Event, Thread	1✔
7	from typing import Optional	1✔
8
9	from localstack.aws.api.lambda_ import Arn	1✔
10	from localstack.config import LAMBDA_DEBUG_MODE, LAMBDA_DEBUG_MODE_CONFIG_PATH	1✔
11	from localstack.utils.lambda_debug_mode.lambda_debug_mode_config import (	1✔
12	LambdaDebugConfig,
13	LambdaDebugModeConfig,
14	load_lambda_debug_mode_config,
15	)
16	from localstack.utils.objects import singleton_factory	1✔
17
18	LOG = logging.getLogger(__name__)	1✔
19
20
21	class LambdaDebugModeSession:	1✔
22	_is_lambda_debug_mode: bool	1✔
23
24	_configuration_file_path: Optional[str]	1✔
25	_watch_thread: Optional[Thread]	1✔
26	_initialised_event: Optional[Event]	1✔
27	_stop_event: Optional[Event]	1✔
28	_config: Optional[LambdaDebugModeConfig]	1✔
29
30	def __init__(self):	1✔
31	self._is_lambda_debug_mode = bool(LAMBDA_DEBUG_MODE)	1✔
32
33	# Disabled Lambda Debug Mode state initialisation.
34	self._configuration_file_path = None	1✔
35	self._watch_thread = None	1✔
36	self._initialised_event = None	1✔
37	self._stop_event = None	1✔
38	self._config = None	1✔
39
40	# Lambda Debug Mode is not enabled: leave as disabled state and return.
41	if not self._is_lambda_debug_mode:	1✔
42	return	1✔
43
44	# Lambda Debug Mode is enabled.
45	# Instantiate the configuration requirements if a configuration file is given.
46	self._configuration_file_path = LAMBDA_DEBUG_MODE_CONFIG_PATH	×
47	if not self._configuration_file_path:	×
48	return	×
49
50	# A configuration file path is given: initialised the resources to load and watch the file.
51
52	# Signal and block on first loading to ensure this is enforced from the very first
53	# invocation, as this module is not loaded at startup. The LambdaDebugModeConfigWatch
54	# thread will then take care of updating the configuration periodically and asynchronously.
55	# This may somewhat slow down the first upstream thread loading this module, but not
56	# future calls. On the other hand, avoiding this mechanism means that first Lambda calls
57	# occur with no Debug configuration.
58	self._initialised_event = Event()	×
59
60	# Signals when a shutdown signal from the application is registered.
61	self._stop_event = Event()	×
62
63	self._watch_thread = Thread(	×
64	target=self._watch_logic, args=(), daemon=True, name="LambdaDebugModeConfigWatch"
65	)
66	self._watch_thread.start()	×
67
68	@staticmethod	1✔
69	@singleton_factory	1✔
70	def get() -> LambdaDebugModeSession:	1✔
71	"""Returns a singleton instance of the Lambda Debug Mode session."""
72	return LambdaDebugModeSession()	1✔
73
74	def ensure_running(self) -> None:	1✔
75	# Nothing to start.
76	if self._watch_thread is None or self._watch_thread.is_alive():	1✔
77	return	1✔
78	try:	×
79	self._watch_thread.start()	×
80	except Exception as exception:	×
81	exception_str = str(exception)	×
82	# The thread was already restarted by another process.
83	if (	×
84	isinstance(exception, RuntimeError)
85	and exception_str
86	and "threads can only be started once" in exception_str
87	):
88	return	×
89	LOG.error(	×
90	"Lambda Debug Mode could not restart the "
91	"hot reloading of the configuration file, '%s'",
92	exception_str,
93	)
94
95	def signal_stop(self) -> None:	1✔
96	stop_event = self._stop_event	1✔
97	if stop_event is not None:	1✔
98	stop_event.set()	×
99
100	def _load_lambda_debug_mode_config(self):	1✔
101	yaml_configuration_string = None	×
102	try:	×
103	with open(self._configuration_file_path, "r") as df:	×
104	yaml_configuration_string = df.read()	×
105	except FileNotFoundError:	×
106	LOG.error(	×
107	"Error: The file lambda debug config file '%s' was not found.",
108	self._configuration_file_path,
109	)
110	except IsADirectoryError:	×
111	LOG.error(	×
112	"Error: Expected a lambda debug config file but found a directory at '%s'.",
113	self._configuration_file_path,
114	)
115	except PermissionError:	×
116	LOG.error(	×
117	"Error: Permission denied while trying to read the lambda debug config file '%s'.",
118	self._configuration_file_path,
119	)
120	except Exception as ex:	×
121	LOG.error(	×
122	"Error: An unexpected error occurred while reading lambda debug config '%s': '%s'",
123	self._configuration_file_path,
124	ex,
125	)
126	if not yaml_configuration_string:	×
127	return None	×
128
129	self._config = load_lambda_debug_mode_config(yaml_configuration_string)	×
130	if self._config is not None:	×
131	LOG.info("Lambda Debug Mode is now enforcing the latest configuration.")	×
132	else:
133	LOG.warning(	×
134	"Lambda Debug Mode could not load the latest configuration due to an error, "
135	"check logs for more details."
136	)
137
138	def _config_file_epoch_last_modified_or_now(self) -> int:	1✔
139	try:	×
140	modified_time = os.path.getmtime(self._configuration_file_path)	×
141	return int(modified_time)	×
142	except Exception as e:	×
143	LOG.warning("Lambda Debug Mode could not access the configuration file: %s", e)	×
144	epoch_now = int(time.time())	×
145	return epoch_now	×
146
147	def _watch_logic(self) -> None:	1✔
148	# TODO: consider relying on system calls (watchdog lib for cross-platform support)
149	# instead of monitoring last modified dates.
150	# Run the first load and signal as initialised.
151	epoch_last_loaded: int = self._config_file_epoch_last_modified_or_now()	×
152	self._load_lambda_debug_mode_config()	×
153	self._initialised_event.set()	×
154
155	# Monitor for file changes whilst the application is running.
156	while not self._stop_event.is_set():	×
157	time.sleep(1)	×
158	epoch_last_modified = self._config_file_epoch_last_modified_or_now()	×
159	if epoch_last_modified > epoch_last_loaded:	×
160	epoch_last_loaded = epoch_last_modified	×
161	self._load_lambda_debug_mode_config()	×
162
163	def _get_initialised_config(self) -> Optional[LambdaDebugModeConfig]:	1✔
164	# Check the session is not initialising, and if so then wait for initialisation to finish.
165	# Note: the initialisation event is otherwise left set since after first initialisation has terminated.
166	if self._initialised_event is not None:	×
167	self._initialised_event.wait()	×
168	return self._config	×
169
170	def is_lambda_debug_mode(self) -> bool:	1✔
171	return self._is_lambda_debug_mode	1✔
172
173	def debug_config_for(self, lambda_arn: Arn) -> Optional[LambdaDebugConfig]:	1✔
174	config = self._get_initialised_config()	×
175	return config.functions.get(lambda_arn) if config else None	×

localstack / localstack / fb37c2a2-4bd7-49b4-891b-ede18134b4b1

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