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

76.92

/localstack-core/localstack/services/stepfunctions/asl/component/common/path/input_path.py

from typing import Final

from localstack.aws.api.stepfunctions import HistoryEventType, TaskFailedEventDetails
from localstack.services.stepfunctions.asl.component.common.error_name.failure_event import (
    FailureEvent,
    FailureEventException,
)
from localstack.services.stepfunctions.asl.component.common.error_name.states_error_name import (
    StatesErrorName,
)
from localstack.services.stepfunctions.asl.component.common.error_name.states_error_name_type import (
    StatesErrorNameType,
)
from localstack.services.stepfunctions.asl.component.common.string.string_expression import (
    StringJsonPath,
    StringSampler,
)
from localstack.services.stepfunctions.asl.component.eval_component import EvalComponent
from localstack.services.stepfunctions.asl.eval.environment import Environment
from localstack.services.stepfunctions.asl.eval.event.event_detail import EventDetails
from localstack.services.stepfunctions.asl.utils.json_path import NoSuchJsonPathError


class InputPath(EvalComponent):
    string_sampler: Final[StringSampler | None]

    def __init__(self, string_sampler: StringSampler | None):
        self.string_sampler = string_sampler

    def _eval_body(self, env: Environment) -> None:
        if self.string_sampler is None:
            env.stack.append({})
            return
        if isinstance(self.string_sampler, StringJsonPath):
            # JsonPaths are sampled from a given state, hence pass the state's input.
            env.stack.append(env.states.get_input())
        try:
            self.string_sampler.eval(env=env)
        except NoSuchJsonPathError as no_such_json_path_error:
            json_path = no_such_json_path_error.json_path
            cause = f"Invalid path '{json_path}' : No results for path: $['{json_path[2:]}']"
            raise FailureEventException(
                failure_event=FailureEvent(
                    env=env,
                    error_name=StatesErrorName(typ=StatesErrorNameType.StatesRuntime),
                    event_type=HistoryEventType.TaskFailed,
                    event_details=EventDetails(
                        taskFailedEventDetails=TaskFailedEventDetails(
                            error=StatesErrorNameType.StatesRuntime.to_name(), cause=cause
                        )
                    ),
                )
            )

1	from typing import Final	1✔
2
3	from localstack.aws.api.stepfunctions import HistoryEventType, TaskFailedEventDetails	1✔
4	from localstack.services.stepfunctions.asl.component.common.error_name.failure_event import (	1✔
5	FailureEvent,
6	FailureEventException,
7	)
8	from localstack.services.stepfunctions.asl.component.common.error_name.states_error_name import (	1✔
9	StatesErrorName,
10	)
11	from localstack.services.stepfunctions.asl.component.common.error_name.states_error_name_type import (	1✔
12	StatesErrorNameType,
13	)
14	from localstack.services.stepfunctions.asl.component.common.string.string_expression import (	1✔
15	StringJsonPath,
16	StringSampler,
17	)
18	from localstack.services.stepfunctions.asl.component.eval_component import EvalComponent	1✔
19	from localstack.services.stepfunctions.asl.eval.environment import Environment	1✔
20	from localstack.services.stepfunctions.asl.eval.event.event_detail import EventDetails	1✔
21	from localstack.services.stepfunctions.asl.utils.json_path import NoSuchJsonPathError	1✔
22
23
24	class InputPath(EvalComponent):	1✔
25	string_sampler: Final[StringSampler \| None]	1✔
26
27	def __init__(self, string_sampler: StringSampler \| None):	1✔
28	self.string_sampler = string_sampler	1✔
29
30	def _eval_body(self, env: Environment) -> None:	1✔
31	if self.string_sampler is None:	1✔
UNCOV 32	env.stack.append({})	×
UNCOV 33	return	×
34	if isinstance(self.string_sampler, StringJsonPath):	1✔
35	# JsonPaths are sampled from a given state, hence pass the state's input.
36	env.stack.append(env.states.get_input())	1✔
37	try:	1✔
38	self.string_sampler.eval(env=env)	1✔
UNCOV 39	except NoSuchJsonPathError as no_such_json_path_error:	×
UNCOV 40	json_path = no_such_json_path_error.json_path	×
UNCOV 41	cause = f"Invalid path '{json_path}' : No results for path: $['{json_path[2:]}']"	×
UNCOV 42	raise FailureEventException(	×
43	failure_event=FailureEvent(
44	env=env,
45	error_name=StatesErrorName(typ=StatesErrorNameType.StatesRuntime),
46	event_type=HistoryEventType.TaskFailed,
47	event_details=EventDetails(
48	taskFailedEventDetails=TaskFailedEventDetails(
49	error=StatesErrorNameType.StatesRuntime.to_name(), cause=cause
50	)
51	),
52	)
53	)

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