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

80.0

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

import abc
from typing import Final

from localstack.aws.api.stepfunctions import ExecutionFailedEventDetails, Timestamp


class ProgramState(abc.ABC): ...


class ProgramEnded(ProgramState):
    pass


class ProgramStopped(ProgramState):
    def __init__(self, stop_date: Timestamp, error: str | None, cause: str | None):
        super().__init__()
        self.stop_date: Timestamp = stop_date
        self.error: str | None = error
        self.cause: str | None = cause


class ProgramRunning(ProgramState):
    _next_state_name: str | None
    _next_field_name: str | None

    def __init__(self):
        super().__init__()
        self._next_state_name = None
        self._next_field_name = None

    @property
    def next_state_name(self) -> str:
        next_state_name = self._next_state_name
        if next_state_name is None:
            raise RuntimeError("Could not retrieve NextState from uninitialised ProgramState.")
        return next_state_name

    @next_state_name.setter
    def next_state_name(self, next_state_name) -> None:
        self._next_state_name = next_state_name
        self._next_field_name = None

    @property
    def next_field_name(self) -> str:
        return self._next_field_name

    @next_field_name.setter
    def next_field_name(self, next_field_name) -> None:
        next_state_name = self._next_state_name
        if next_state_name is None:
            raise RuntimeError("Could not set NextField from uninitialised ProgramState.")
        self._next_field_name = next_field_name


class ProgramError(ProgramState):
    error: Final[ExecutionFailedEventDetails | None]

    def __init__(self, error: ExecutionFailedEventDetails | None):
        super().__init__()
        self.error = error


class ProgramTimedOut(ProgramState):
    pass

1	import abc	1✔
2	from typing import Final	1✔
3
4	from localstack.aws.api.stepfunctions import ExecutionFailedEventDetails, Timestamp	1✔
5
6
7	class ProgramState(abc.ABC): ...	1✔
8
9
10	class ProgramEnded(ProgramState):	1✔
11	pass	1✔
12
13
14	class ProgramStopped(ProgramState):	1✔
15	def __init__(self, stop_date: Timestamp, error: str \| None, cause: str \| None):	1✔
UNCOV 16	super().__init__()	×
UNCOV 17	self.stop_date: Timestamp = stop_date	×
UNCOV 18	self.error: str \| None = error	×
UNCOV 19	self.cause: str \| None = cause	×
20
21
22	class ProgramRunning(ProgramState):	1✔
23	_next_state_name: str \| None	1✔
24	_next_field_name: str \| None	1✔
25
26	def __init__(self):	1✔
27	super().__init__()	1✔
28	self._next_state_name = None	1✔
29	self._next_field_name = None	1✔
30
31	@property	1✔
32	def next_state_name(self) -> str:	1✔
33	next_state_name = self._next_state_name	1✔
34	if next_state_name is None:	1✔
35	raise RuntimeError("Could not retrieve NextState from uninitialised ProgramState.")	×
36	return next_state_name	1✔
37
38	@next_state_name.setter	1✔
39	def next_state_name(self, next_state_name) -> None:	1✔
40	self._next_state_name = next_state_name	1✔
41	self._next_field_name = None	1✔
42
43	@property	1✔
44	def next_field_name(self) -> str:	1✔
UNCOV 45	return self._next_field_name	×
46
47	@next_field_name.setter	1✔
48	def next_field_name(self, next_field_name) -> None:	1✔
49	next_state_name = self._next_state_name	1✔
50	if next_state_name is None:	1✔
51	raise RuntimeError("Could not set NextField from uninitialised ProgramState.")	×
52	self._next_field_name = next_field_name	1✔
53
54
55	class ProgramError(ProgramState):	1✔
56	error: Final[ExecutionFailedEventDetails \| None]	1✔
57
58	def __init__(self, error: ExecutionFailedEventDetails \| None):	1✔
UNCOV 59	super().__init__()	×
UNCOV 60	self.error = error	×
61
62
63	class ProgramTimedOut(ProgramState):	1✔
64	pass	1✔

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