22709357475

Committed 05 Mar 2026 08:35AM UTC coverage: 59.732% (-27.2%) from 86.974%

Build # 22709357475

Build Type

Pull #13880

github

Committed by

web-flow

Commit Message

Merge 28fcab93c into 710618057

Pull Request Pull Request #13880: Firehose: Replace TaggingService

Coverage Stats

12 of 12 new or added lines in 2 files covered. (100.0%)

20464 existing lines in 510 files now uncovered.

45290 of 75822 relevant lines covered (59.73%)

0.6 hits per line

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

50.59

/localstack-core/localstack/services/stepfunctions/asl/eval/callback/callback.py

import abc
from collections import OrderedDict
from threading import Event, Lock
from typing import Final

from localstack.aws.api.stepfunctions import ActivityDoesNotExist, Arn
from localstack.services.stepfunctions.backend.activity import Activity, ActivityTask
from localstack.utils.strings import long_uid

CallbackId = str


class CallbackOutcome(abc.ABC):
    callback_id: Final[CallbackId]

    def __init__(self, callback_id: str):
        self.callback_id = callback_id


class CallbackOutcomeSuccess(CallbackOutcome):
    output: Final[str]

    def __init__(self, callback_id: CallbackId, output: str):
        super().__init__(callback_id=callback_id)
        self.output = output


class CallbackOutcomeFailure(CallbackOutcome):
    error: Final[str | None]
    cause: Final[str | None]

    def __init__(self, callback_id: CallbackId, error: str | None, cause: str | None):
        super().__init__(callback_id=callback_id)
        self.error = error
        self.cause = cause


class CallbackOutcomeTimedOut(CallbackOutcome):
    pass


class CallbackTimeoutError(TimeoutError):
    pass


class CallbackConsumerError(abc.ABC): ...


class CallbackConsumerTimeout(CallbackConsumerError):
    pass


class CallbackConsumerLeft(CallbackConsumerError):
    pass


class HeartbeatEndpoint:
    _mutex: Final[Lock]
    _next_heartbeat_event: Final[Event]
    _heartbeat_seconds: Final[int]

    def __init__(self, heartbeat_seconds: int):
        self._mutex = Lock()
        self._next_heartbeat_event = Event()
        self._heartbeat_seconds = heartbeat_seconds

    def clear_and_wait(self) -> bool:
        with self._mutex:
            if self._next_heartbeat_event.is_set():
                self._next_heartbeat_event.clear()
                return True
        return self._next_heartbeat_event.wait(timeout=self._heartbeat_seconds)

    def notify(self):
        with self._mutex:
            self._next_heartbeat_event.set()


class HeartbeatTimeoutError(TimeoutError):
    pass


class HeartbeatTimedOut(CallbackConsumerError):
    pass


class ActivityTaskStartOutcome:
    worker_name: str | None

    def __init__(self, worker_name: str | None = None):
        self.worker_name = worker_name


class ActivityTaskStartEndpoint:
    _next_activity_task_start_event: Final[Event]
    _outcome: ActivityTaskStartOutcome | None

    def __init__(self):
        self._next_activity_task_start_event = Event()

    def wait(self, timeout_seconds: float) -> ActivityTaskStartOutcome | None:
        self._next_activity_task_start_event.wait(timeout=timeout_seconds)
        return self._outcome

    def notify(self, activity_task: ActivityTaskStartOutcome) -> None:
        self._outcome = activity_task
        self._next_activity_task_start_event.set()


class CallbackEndpoint:
    callback_id: Final[CallbackId]
    _notify_event: Final[Event]
    _outcome: CallbackOutcome | None
    consumer_error: CallbackConsumerError | None
    _heartbeat_endpoint: HeartbeatEndpoint | None

    def __init__(self, callback_id: CallbackId):
        self.callback_id = callback_id
        self._notify_event = Event()
        self._outcome = None
        self.consumer_error = None
        self._heartbeat_endpoint = None

    def setup_heartbeat_endpoint(self, heartbeat_seconds: int) -> HeartbeatEndpoint:
        self._heartbeat_endpoint = HeartbeatEndpoint(heartbeat_seconds=heartbeat_seconds)
        return self._heartbeat_endpoint

    def interrupt_all(self) -> None:
        # Interrupts all waiting processes on this endpoint.
        self._notify_event.set()
        heartbeat_endpoint = self._heartbeat_endpoint
        if heartbeat_endpoint is not None:
            heartbeat_endpoint.notify()

    def notify(self, outcome: CallbackOutcome):
        self._outcome = outcome
        self._notify_event.set()
        if self._heartbeat_endpoint:
            self._heartbeat_endpoint.notify()

    def notify_heartbeat(self) -> bool:
        if not self._heartbeat_endpoint:
            return False
        self._heartbeat_endpoint.notify()
        return True

    def wait(self, timeout: float | None = None) -> CallbackOutcome | None:
        self._notify_event.wait(timeout=timeout)
        return self._outcome

    def get_outcome(self) -> CallbackOutcome | None:
        return self._outcome

    def report(self, consumer_error: CallbackConsumerError) -> None:
        self.consumer_error = consumer_error


class ActivityCallbackEndpoint(CallbackEndpoint):
    _activity_task_start_endpoint: Final[ActivityTaskStartEndpoint]
    _activity_input: Final[str]

    def __init__(self, callback_id: str, activity_input: str):
        super().__init__(callback_id=callback_id)
        self._activity_input = activity_input
        self._activity_task_start_endpoint = ActivityTaskStartEndpoint()

    def get_activity_input(self) -> str:
        return self._activity_input

    def get_activity_task_start_endpoint(self) -> ActivityTaskStartEndpoint:
        return self._activity_task_start_endpoint

    def notify_activity_task_start(self, worker_name: str | None) -> None:
        self._activity_task_start_endpoint.notify(ActivityTaskStartOutcome(worker_name=worker_name))


class CallbackNotifyConsumerError(RuntimeError):
    callback_consumer_error: CallbackConsumerError

    def __init__(self, callback_consumer_error: CallbackConsumerError):
        self.callback_consumer_error = callback_consumer_error


class CallbackOutcomeFailureError(RuntimeError):
    callback_outcome_failure: CallbackOutcomeFailure

    def __init__(self, callback_outcome_failure: CallbackOutcomeFailure):
        self.callback_outcome_failure = callback_outcome_failure


class CallbackPoolManager:
    _activity_store: Final[dict[CallbackId, Activity]]
    _pool: Final[dict[CallbackId, CallbackEndpoint]]

    def __init__(self, activity_store: dict[Arn, Activity]):
        self._activity_store = activity_store
        self._pool = OrderedDict()

    def get(self, callback_id: CallbackId) -> CallbackEndpoint | None:
        return self._pool.get(callback_id)

    def add(self, callback_id: CallbackId) -> CallbackEndpoint:
        if callback_id in self._pool:
            raise ValueError("Duplicate callback token id value.")
        callback_endpoint = CallbackEndpoint(callback_id=callback_id)
        self._pool[callback_id] = callback_endpoint
        return callback_endpoint

    def add_activity_task(
        self, callback_id: CallbackId, activity_arn: Arn, activity_input: str
    ) -> ActivityCallbackEndpoint:
        if callback_id in self._pool:
            raise ValueError("Duplicate callback token id value.")

        maybe_activity: Activity | None = self._activity_store.get(activity_arn)
        if maybe_activity is None:
            raise ActivityDoesNotExist()

        maybe_activity.add_task(ActivityTask(task_token=callback_id, task_input=activity_input))

        callback_endpoint = ActivityCallbackEndpoint(
            callback_id=callback_id, activity_input=activity_input
        )
        self._pool[callback_id] = callback_endpoint
        return callback_endpoint

    def generate(self) -> CallbackEndpoint:
        return self.add(long_uid())

    def notify(self, callback_id: CallbackId, outcome: CallbackOutcome) -> bool:
        callback_endpoint = self._pool.get(callback_id, None)
        if callback_endpoint is None:
            return False

        consumer_error: CallbackConsumerError | None = callback_endpoint.consumer_error
        if consumer_error is not None:
            raise CallbackNotifyConsumerError(callback_consumer_error=consumer_error)

        callback_endpoint.notify(outcome=outcome)
        return True

    def heartbeat(self, callback_id: CallbackId) -> bool:
        callback_endpoint = self._pool.get(callback_id, None)
        if callback_endpoint is None:
            return False

        consumer_error: CallbackConsumerError | None = callback_endpoint.consumer_error
        if consumer_error is not None:
            raise CallbackNotifyConsumerError(callback_consumer_error=consumer_error)

        return callback_endpoint.notify_heartbeat()

1	import abc	1✔
2	from collections import OrderedDict	1✔
3	from threading import Event, Lock	1✔
4	from typing import Final	1✔
5
6	from localstack.aws.api.stepfunctions import ActivityDoesNotExist, Arn	1✔
7	from localstack.services.stepfunctions.backend.activity import Activity, ActivityTask	1✔
8	from localstack.utils.strings import long_uid	1✔
9
10	CallbackId = str	1✔
11
12
13	class CallbackOutcome(abc.ABC):	1✔
14	callback_id: Final[CallbackId]	1✔
15
16	def __init__(self, callback_id: str):	1✔
UNCOV 17	self.callback_id = callback_id	×
18
19
20	class CallbackOutcomeSuccess(CallbackOutcome):	1✔
21	output: Final[str]	1✔
22
23	def __init__(self, callback_id: CallbackId, output: str):	1✔
UNCOV 24	super().__init__(callback_id=callback_id)	×
UNCOV 25	self.output = output	×
26
27
28	class CallbackOutcomeFailure(CallbackOutcome):	1✔
29	error: Final[str \| None]	1✔
30	cause: Final[str \| None]	1✔
31
32	def __init__(self, callback_id: CallbackId, error: str \| None, cause: str \| None):	1✔
UNCOV 33	super().__init__(callback_id=callback_id)	×
UNCOV 34	self.error = error	×
UNCOV 35	self.cause = cause	×
36
37
38	class CallbackOutcomeTimedOut(CallbackOutcome):	1✔
39	pass	1✔
40
41
42	class CallbackTimeoutError(TimeoutError):	1✔
43	pass	1✔
44
45
46	class CallbackConsumerError(abc.ABC): ...	1✔
47
48
49	class CallbackConsumerTimeout(CallbackConsumerError):	1✔
50	pass	1✔
51
52
53	class CallbackConsumerLeft(CallbackConsumerError):	1✔
54	pass	1✔
55
56
57	class HeartbeatEndpoint:	1✔
58	_mutex: Final[Lock]	1✔
59	_next_heartbeat_event: Final[Event]	1✔
60	_heartbeat_seconds: Final[int]	1✔
61
62	def __init__(self, heartbeat_seconds: int):	1✔
UNCOV 63	self._mutex = Lock()	×
UNCOV 64	self._next_heartbeat_event = Event()	×
UNCOV 65	self._heartbeat_seconds = heartbeat_seconds	×
66
67	def clear_and_wait(self) -> bool:	1✔
UNCOV 68	with self._mutex:	×
UNCOV 69	if self._next_heartbeat_event.is_set():	×
UNCOV 70	self._next_heartbeat_event.clear()	×
UNCOV 71	return True	×
UNCOV 72	return self._next_heartbeat_event.wait(timeout=self._heartbeat_seconds)	×
73
74	def notify(self):	1✔
UNCOV 75	with self._mutex:	×
UNCOV 76	self._next_heartbeat_event.set()	×
77
78
79	class HeartbeatTimeoutError(TimeoutError):	1✔
80	pass	1✔
81
82
83	class HeartbeatTimedOut(CallbackConsumerError):	1✔
84	pass	1✔
85
86
87	class ActivityTaskStartOutcome:	1✔
88	worker_name: str \| None	1✔
89
90	def __init__(self, worker_name: str \| None = None):	1✔
UNCOV 91	self.worker_name = worker_name	×
92
93
94	class ActivityTaskStartEndpoint:	1✔
95	_next_activity_task_start_event: Final[Event]	1✔
96	_outcome: ActivityTaskStartOutcome \| None	1✔
97
98	def __init__(self):	1✔
UNCOV 99	self._next_activity_task_start_event = Event()	×
100
101	def wait(self, timeout_seconds: float) -> ActivityTaskStartOutcome \| None:	1✔
UNCOV 102	self._next_activity_task_start_event.wait(timeout=timeout_seconds)	×
UNCOV 103	return self._outcome	×
104
105	def notify(self, activity_task: ActivityTaskStartOutcome) -> None:	1✔
UNCOV 106	self._outcome = activity_task	×
UNCOV 107	self._next_activity_task_start_event.set()	×
108
109
110	class CallbackEndpoint:	1✔
111	callback_id: Final[CallbackId]	1✔
112	_notify_event: Final[Event]	1✔
113	_outcome: CallbackOutcome \| None	1✔
114	consumer_error: CallbackConsumerError \| None	1✔
115	_heartbeat_endpoint: HeartbeatEndpoint \| None	1✔
116
117	def __init__(self, callback_id: CallbackId):	1✔
UNCOV 118	self.callback_id = callback_id	×
UNCOV 119	self._notify_event = Event()	×
UNCOV 120	self._outcome = None	×
UNCOV 121	self.consumer_error = None	×
UNCOV 122	self._heartbeat_endpoint = None	×
123
124	def setup_heartbeat_endpoint(self, heartbeat_seconds: int) -> HeartbeatEndpoint:	1✔
UNCOV 125	self._heartbeat_endpoint = HeartbeatEndpoint(heartbeat_seconds=heartbeat_seconds)	×
UNCOV 126	return self._heartbeat_endpoint	×
127
128	def interrupt_all(self) -> None:	1✔
129	# Interrupts all waiting processes on this endpoint.
UNCOV 130	self._notify_event.set()	×
UNCOV 131	heartbeat_endpoint = self._heartbeat_endpoint	×
UNCOV 132	if heartbeat_endpoint is not None:	×
UNCOV 133	heartbeat_endpoint.notify()	×
134
135	def notify(self, outcome: CallbackOutcome):	1✔
UNCOV 136	self._outcome = outcome	×
UNCOV 137	self._notify_event.set()	×
UNCOV 138	if self._heartbeat_endpoint:	×
UNCOV 139	self._heartbeat_endpoint.notify()	×
140
141	def notify_heartbeat(self) -> bool:	1✔
UNCOV 142	if not self._heartbeat_endpoint:	×
143	return False	×
UNCOV 144	self._heartbeat_endpoint.notify()	×
UNCOV 145	return True	×
146
147	def wait(self, timeout: float \| None = None) -> CallbackOutcome \| None:	1✔
UNCOV 148	self._notify_event.wait(timeout=timeout)	×
UNCOV 149	return self._outcome	×
150
151	def get_outcome(self) -> CallbackOutcome \| None:	1✔
UNCOV 152	return self._outcome	×
153
154	def report(self, consumer_error: CallbackConsumerError) -> None:	1✔
155	self.consumer_error = consumer_error	×
156
157
158	class ActivityCallbackEndpoint(CallbackEndpoint):	1✔
159	_activity_task_start_endpoint: Final[ActivityTaskStartEndpoint]	1✔
160	_activity_input: Final[str]	1✔
161
162	def __init__(self, callback_id: str, activity_input: str):	1✔
UNCOV 163	super().__init__(callback_id=callback_id)	×
UNCOV 164	self._activity_input = activity_input	×
UNCOV 165	self._activity_task_start_endpoint = ActivityTaskStartEndpoint()	×
166
167	def get_activity_input(self) -> str:	1✔
168	return self._activity_input	×
169
170	def get_activity_task_start_endpoint(self) -> ActivityTaskStartEndpoint:	1✔
UNCOV 171	return self._activity_task_start_endpoint	×
172
173	def notify_activity_task_start(self, worker_name: str \| None) -> None:	1✔
UNCOV 174	self._activity_task_start_endpoint.notify(ActivityTaskStartOutcome(worker_name=worker_name))	×
175
176
177	class CallbackNotifyConsumerError(RuntimeError):	1✔
178	callback_consumer_error: CallbackConsumerError	1✔
179
180	def __init__(self, callback_consumer_error: CallbackConsumerError):	1✔
181	self.callback_consumer_error = callback_consumer_error	×
182
183
184	class CallbackOutcomeFailureError(RuntimeError):	1✔
185	callback_outcome_failure: CallbackOutcomeFailure	1✔
186
187	def __init__(self, callback_outcome_failure: CallbackOutcomeFailure):	1✔
UNCOV 188	self.callback_outcome_failure = callback_outcome_failure	×
189
190
191	class CallbackPoolManager:	1✔
192	_activity_store: Final[dict[CallbackId, Activity]]	1✔
193	_pool: Final[dict[CallbackId, CallbackEndpoint]]	1✔
194
195	def __init__(self, activity_store: dict[Arn, Activity]):	1✔
196	self._activity_store = activity_store	1✔
197	self._pool = OrderedDict()	1✔
198
199	def get(self, callback_id: CallbackId) -> CallbackEndpoint \| None:	1✔
UNCOV 200	return self._pool.get(callback_id)	×
201
202	def add(self, callback_id: CallbackId) -> CallbackEndpoint:	1✔
UNCOV 203	if callback_id in self._pool:	×
204	raise ValueError("Duplicate callback token id value.")	×
UNCOV 205	callback_endpoint = CallbackEndpoint(callback_id=callback_id)	×
UNCOV 206	self._pool[callback_id] = callback_endpoint	×
UNCOV 207	return callback_endpoint	×
208
209	def add_activity_task(	1✔
210	self, callback_id: CallbackId, activity_arn: Arn, activity_input: str
211	) -> ActivityCallbackEndpoint:
UNCOV 212	if callback_id in self._pool:	×
213	raise ValueError("Duplicate callback token id value.")	×
214
UNCOV 215	maybe_activity: Activity \| None = self._activity_store.get(activity_arn)	×
UNCOV 216	if maybe_activity is None:	×
UNCOV 217	raise ActivityDoesNotExist()	×
218
UNCOV 219	maybe_activity.add_task(ActivityTask(task_token=callback_id, task_input=activity_input))	×
220
UNCOV 221	callback_endpoint = ActivityCallbackEndpoint(	×
222	callback_id=callback_id, activity_input=activity_input
223	)
UNCOV 224	self._pool[callback_id] = callback_endpoint	×
UNCOV 225	return callback_endpoint	×
226
227	def generate(self) -> CallbackEndpoint:	1✔
228	return self.add(long_uid())	×
229
230	def notify(self, callback_id: CallbackId, outcome: CallbackOutcome) -> bool:	1✔
UNCOV 231	callback_endpoint = self._pool.get(callback_id, None)	×
UNCOV 232	if callback_endpoint is None:	×
UNCOV 233	return False	×
234
UNCOV 235	consumer_error: CallbackConsumerError \| None = callback_endpoint.consumer_error	×
UNCOV 236	if consumer_error is not None:	×
237	raise CallbackNotifyConsumerError(callback_consumer_error=consumer_error)	×
238
UNCOV 239	callback_endpoint.notify(outcome=outcome)	×
UNCOV 240	return True	×
241
242	def heartbeat(self, callback_id: CallbackId) -> bool:	1✔
UNCOV 243	callback_endpoint = self._pool.get(callback_id, None)	×
UNCOV 244	if callback_endpoint is None:	×
UNCOV 245	return False	×
246
UNCOV 247	consumer_error: CallbackConsumerError \| None = callback_endpoint.consumer_error	×
UNCOV 248	if consumer_error is not None:	×
249	raise CallbackNotifyConsumerError(callback_consumer_error=consumer_error)	×
250
UNCOV 251	return callback_endpoint.notify_heartbeat()	×

localstack / localstack / 22709357475

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