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

40.57

/localstack-core/localstack/services/stepfunctions/asl/parse/test_state/preprocessor.py

import enum
from typing import Final

from antlr4.tree.Tree import ParseTree

from localstack.services.stepfunctions.asl.antlr.runtime.ASLParser import ASLParser
from localstack.services.stepfunctions.asl.antlt4utils.antlr4utils import (
    is_production,
)
from localstack.services.stepfunctions.asl.component.common.parargs import (
    ArgumentsJSONataTemplateValueObject,
    ArgumentsStringJSONata,
    Parameters,
)
from localstack.services.stepfunctions.asl.component.common.path.input_path import InputPath
from localstack.services.stepfunctions.asl.component.common.path.items_path import ItemsPath
from localstack.services.stepfunctions.asl.component.common.path.result_path import ResultPath
from localstack.services.stepfunctions.asl.component.common.query_language import QueryLanguage
from localstack.services.stepfunctions.asl.component.common.result_selector import ResultSelector
from localstack.services.stepfunctions.asl.component.state.state import CommonStateField
from localstack.services.stepfunctions.asl.component.state.state_choice.state_choice import (
    StateChoice,
)
from localstack.services.stepfunctions.asl.component.state.state_execution.state_map.max_concurrency import (
    MaxConcurrency,
    MaxConcurrencyJSONata,
    MaxConcurrencyPath,
)
from localstack.services.stepfunctions.asl.component.state.state_execution.state_map.state_map import (
    StateMap,
)
from localstack.services.stepfunctions.asl.component.state.state_execution.state_map.tolerated_failure import (
    ToleratedFailureCountInt,
    ToleratedFailureCountPath,
    ToleratedFailureCountStringJSONata,
    ToleratedFailurePercentage,
    ToleratedFailurePercentagePath,
    ToleratedFailurePercentageStringJSONata,
)
from localstack.services.stepfunctions.asl.component.state.state_execution.state_parallel.state_parallel import (
    StateParallel,
)
from localstack.services.stepfunctions.asl.component.state.state_execution.state_task.state_task import (
    StateTask,
)
from localstack.services.stepfunctions.asl.component.state.state_fail.state_fail import StateFail
from localstack.services.stepfunctions.asl.component.state.state_pass.result import Result
from localstack.services.stepfunctions.asl.component.state.state_pass.state_pass import StatePass
from localstack.services.stepfunctions.asl.component.state.state_succeed.state_succeed import (
    StateSucceed,
)
from localstack.services.stepfunctions.asl.component.test_state.program.test_state_program import (
    TestStateProgram,
)
from localstack.services.stepfunctions.asl.component.test_state.state.common import (
    MockedCommonState,
)
from localstack.services.stepfunctions.asl.component.test_state.state.map import (
    MockedStateMap,
)
from localstack.services.stepfunctions.asl.component.test_state.state.parallel import (
    MockedStateParallel,
)
from localstack.services.stepfunctions.asl.component.test_state.state.task import (
    MockedStateTask,
)
from localstack.services.stepfunctions.asl.component.test_state.state.test_state_state_props import (
    TestStateStateProps,
)
from localstack.services.stepfunctions.asl.eval.test_state.environment import TestStateEnvironment
from localstack.services.stepfunctions.asl.parse.preprocessor import Preprocessor
from localstack.services.stepfunctions.asl.utils.encoding import to_json_str


class InspectionDataKey(enum.Enum):
    INPUT = "input"
    AFTER_INPUT_PATH = "afterInputPath"
    AFTER_PARAMETERS = "afterParameters"
    AFTER_ARGUMENTS = "afterArguments"
    RESULT = "result"
    AFTER_RESULT_SELECTOR = "afterResultSelector"
    AFTER_RESULT_PATH = "afterResultPath"
    AFTER_ITEMS_PATH = "afterItemsPath"
    REQUEST = "request"
    RESPONSE = "response"

    MAX_CONCURRENCY = "maxConcurrency"
    TOLERATED_FAILURE_COUNT = "toleratedFailureCount"
    TOLERATED_FAILURE_PERCENTAGE = "toleratedFailurePercentage"


def _decorated_updates_inspection_data(method, inspection_data_key: InspectionDataKey):
    def wrapper(env: TestStateEnvironment, *args, **kwargs):
        method(env, *args, **kwargs)
        result = env.stack[-1]
        if not isinstance(result, (int, float)):
            result = to_json_str(result)
        # We know that the enum value used here corresponds to a supported inspection data field by design.
        env.inspection_data[inspection_data_key.value] = result  # noqa

    return wrapper


def _decorate_state_field(state_field: CommonStateField, is_single_state: bool = False) -> None:
    if isinstance(state_field, StateMap):
        MockedStateMap.wrap(state_field, is_single_state)
    elif isinstance(state_field, StateParallel):
        MockedStateParallel.wrap(state_field, is_single_state)
    elif isinstance(state_field, StateTask):
        MockedStateTask.wrap(state_field, is_single_state)
    elif isinstance(state_field, (StateChoice, StatePass, StateFail, StateSucceed)):
        MockedCommonState.wrap(state_field, is_single_state)


def find_state(state_name: str, states: dict[str, CommonStateField]) -> CommonStateField | None:
    if state_name in states:
        return states[state_name]

    for state in states.values():
        if isinstance(state, StateMap):
            found_state = find_state(state_name, state.iteration_component._states.states)
            if found_state:
                return found_state
        elif isinstance(state, StateParallel):
            for program in state.branches.programs:
                found_state = find_state(state_name, program.states.states)
                if found_state:
                    return found_state


class TestStatePreprocessor(Preprocessor):
    STATE_NAME: Final[str] = "StateName"
    _state_name_stack: list[str] = []

    def to_test_state_program(
        self, tree: ParseTree, state_name: str | None = None
    ) -> TestStateProgram:
        if is_production(tree, ASLParser.RULE_state_machine):
            # full definition passed in
            program = self.visitState_machine(ctx=tree)
            state_field = find_state(state_name, program.states.states)
            _decorate_state_field(state_field, False)
            return TestStateProgram(state_field)

        if is_production(tree, ASLParser.RULE_state_decl_body):
            # single state case
            state_props = self.visitState_decl_body(ctx=tree)
            state_field = self._common_state_field_of(state_props=state_props)
            _decorate_state_field(state_field, True)
            return TestStateProgram(state_field)

        return super().visit(tree)

    def visitState_decl(self, ctx: ASLParser.State_declContext) -> CommonStateField:
        # if we are parsing a full state machine, we need to record the state_name prior to stepping
        # into the state body definition.
        state_name = self._inner_string_of(parser_rule_context=ctx.string_literal())
        self._state_name_stack.append(state_name)
        state_props: TestStateStateProps = self.visit(ctx.state_decl_body())
        state_field = self._common_state_field_of(state_props=state_props)
        return state_field

    def visitState_decl_body(self, ctx: ASLParser.State_decl_bodyContext) -> TestStateStateProps:
        self._open_query_language_scope(ctx)
        state_props = TestStateStateProps()
        state_props.name = (
            self._state_name_stack.pop(-1) if self._state_name_stack else self.STATE_NAME
        )
        for child in ctx.children:
            cmp = self.visit(child)
            state_props.add(cmp)
        if state_props.get(QueryLanguage) is None:
            state_props.add(self._get_current_query_language())
        self._close_query_language_scope()
        return state_props

    def visitInput_path_decl(self, ctx: ASLParser.Input_path_declContext) -> InputPath:
        input_path: InputPath = super().visitInput_path_decl(ctx=ctx)
        input_path._eval_body = _decorated_updates_inspection_data(
            method=input_path._eval_body,  # noqa
            inspection_data_key=InspectionDataKey.AFTER_INPUT_PATH,
        )
        return input_path

    def visitParameters_decl(self, ctx: ASLParser.Parameters_declContext) -> Parameters:
        parameters: Parameters = super().visitParameters_decl(ctx=ctx)
        parameters._eval_body = _decorated_updates_inspection_data(
            method=parameters._eval_body,  # noqa
            inspection_data_key=InspectionDataKey.AFTER_PARAMETERS,
        )
        return parameters

    def visitResult_selector_decl(
        self, ctx: ASLParser.Result_selector_declContext
    ) -> ResultSelector:
        result_selector: ResultSelector = super().visitResult_selector_decl(ctx=ctx)
        result_selector._eval_body = _decorated_updates_inspection_data(
            method=result_selector._eval_body,  # noqa
            inspection_data_key=InspectionDataKey.AFTER_RESULT_SELECTOR,
        )
        return result_selector

    def visitResult_path_decl(self, ctx: ASLParser.Result_path_declContext) -> ResultPath:
        result_path: ResultPath = super().visitResult_path_decl(ctx=ctx)
        result_path._eval_body = _decorated_updates_inspection_data(
            method=result_path._eval_body,  # noqa
            inspection_data_key=InspectionDataKey.AFTER_RESULT_PATH,
        )
        return result_path

    def visitResult_decl(self, ctx: ASLParser.Result_declContext) -> Result:
        result: Result = super().visitResult_decl(ctx=ctx)
        result._eval_body = _decorated_updates_inspection_data(
            method=result._eval_body,
            inspection_data_key=InspectionDataKey.RESULT,  # noqa
        )
        return result

    def visitMax_concurrency_int(self, ctx: ASLParser.Max_concurrency_intContext) -> MaxConcurrency:
        max_concurrency: MaxConcurrency = super().visitMax_concurrency_int(ctx)
        max_concurrency._eval_body = _decorated_updates_inspection_data(
            method=max_concurrency._eval_body,
            inspection_data_key=InspectionDataKey.MAX_CONCURRENCY,  # noqa
        )
        return max_concurrency

    def visitMax_concurrency_jsonata(
        self, ctx: ASLParser.Max_concurrency_jsonataContext
    ) -> MaxConcurrencyJSONata:
        max_concurrency_jsonata: MaxConcurrencyJSONata = super().visitMax_concurrency_jsonata(ctx)
        max_concurrency_jsonata._eval_body = _decorated_updates_inspection_data(
            method=max_concurrency_jsonata._eval_body,
            inspection_data_key=InspectionDataKey.MAX_CONCURRENCY,  # noqa
        )
        return max_concurrency_jsonata

    def visitMax_concurrency_path(
        self, ctx: ASLParser.Max_concurrency_declContext
    ) -> MaxConcurrencyPath:
        max_concurrency_path: MaxConcurrencyPath = super().visitMax_concurrency_path(ctx)
        max_concurrency_path._eval_body = _decorated_updates_inspection_data(
            method=max_concurrency_path._eval_body,
            inspection_data_key=InspectionDataKey.MAX_CONCURRENCY,  # noqa
        )
        return max_concurrency_path

    def visitTolerated_failure_count_int(self, ctx) -> ToleratedFailureCountInt:
        tolerated_failure_count: ToleratedFailureCountInt = (
            super().visitTolerated_failure_count_int(ctx)
        )
        tolerated_failure_count._eval_body = _decorated_updates_inspection_data(
            method=tolerated_failure_count._eval_body,
            inspection_data_key=InspectionDataKey.TOLERATED_FAILURE_COUNT,
        )
        return tolerated_failure_count

    def visitTolerated_failure_count_path(self, ctx) -> ToleratedFailureCountPath:
        tolerated_failure_count_path: ToleratedFailureCountPath = (
            super().visitTolerated_failure_count_path(ctx)
        )
        tolerated_failure_count_path._eval_body = _decorated_updates_inspection_data(
            method=tolerated_failure_count_path._eval_body,
            inspection_data_key=InspectionDataKey.TOLERATED_FAILURE_COUNT,
        )
        return tolerated_failure_count_path

    def visitTolerated_failure_count_string_jsonata(
        self, ctx
    ) -> ToleratedFailureCountStringJSONata:
        tolerated_failure_count_jsonata: ToleratedFailureCountStringJSONata = (
            super().visitTolerated_failure_count_string_jsonata(ctx)
        )
        tolerated_failure_count_jsonata._eval_body = _decorated_updates_inspection_data(
            method=tolerated_failure_count_jsonata._eval_body,
            inspection_data_key=InspectionDataKey.TOLERATED_FAILURE_COUNT,
        )
        return tolerated_failure_count_jsonata

    def visitTolerated_failure_percentage_number(self, ctx) -> ToleratedFailurePercentage:
        tolerated_failure_percentage: ToleratedFailurePercentage = (
            super().visitTolerated_failure_percentage_number(ctx)
        )
        tolerated_failure_percentage._eval_body = _decorated_updates_inspection_data(
            method=tolerated_failure_percentage._eval_body,
            inspection_data_key=InspectionDataKey.TOLERATED_FAILURE_PERCENTAGE,
        )
        return tolerated_failure_percentage

    def visitTolerated_failure_percentage_path(self, ctx) -> ToleratedFailurePercentagePath:
        tolerated_failure_percentage_path: ToleratedFailurePercentagePath = (
            super().visitTolerated_failure_percentage_path(ctx)
        )
        tolerated_failure_percentage_path._eval_body = _decorated_updates_inspection_data(
            method=tolerated_failure_percentage_path._eval_body,
            inspection_data_key=InspectionDataKey.TOLERATED_FAILURE_PERCENTAGE,
        )
        return tolerated_failure_percentage_path

    def visitTolerated_failure_percentage_string_jsonata(
        self, ctx
    ) -> ToleratedFailurePercentageStringJSONata:
        tolerated_failure_percentage_jsonata: ToleratedFailurePercentageStringJSONata = (
            super().visitTolerated_failure_percentage_string_jsonata(ctx)
        )
        tolerated_failure_percentage_jsonata._eval_body = _decorated_updates_inspection_data(
            method=tolerated_failure_percentage_jsonata._eval_body,
            inspection_data_key=InspectionDataKey.TOLERATED_FAILURE_PERCENTAGE,
        )
        return tolerated_failure_percentage_jsonata

    def visitItems_path_decl(self, ctx) -> ItemsPath:
        items_path: ItemsPath = super().visitItems_path_decl(ctx)
        items_path._eval_body = _decorated_updates_inspection_data(
            method=items_path._eval_body,
            inspection_data_key=InspectionDataKey.AFTER_ITEMS_PATH,
        )
        return items_path

    def visitArguments_string_jsonata(self, ctx):
        arguments: ArgumentsStringJSONata = super().visitArguments_string_jsonata(ctx)
        arguments._eval_body = _decorated_updates_inspection_data(
            method=arguments._eval_body,
            inspection_data_key=InspectionDataKey.AFTER_ARGUMENTS,
        )
        return arguments

    def visitArguments_jsonata_template_value_object(self, ctx):
        arguments: ArgumentsJSONataTemplateValueObject = (
            super().visitArguments_jsonata_template_value_object(ctx)
        )
        arguments._eval_body = _decorated_updates_inspection_data(
            method=arguments._eval_body,
            inspection_data_key=InspectionDataKey.AFTER_ARGUMENTS,
        )
        return arguments

1	import enum	1✔
2	from typing import Final	1✔
3
4	from antlr4.tree.Tree import ParseTree	1✔
5
6	from localstack.services.stepfunctions.asl.antlr.runtime.ASLParser import ASLParser	1✔
7	from localstack.services.stepfunctions.asl.antlt4utils.antlr4utils import (	1✔
8	is_production,
9	)
10	from localstack.services.stepfunctions.asl.component.common.parargs import (	1✔
11	ArgumentsJSONataTemplateValueObject,
12	ArgumentsStringJSONata,
13	Parameters,
14	)
15	from localstack.services.stepfunctions.asl.component.common.path.input_path import InputPath	1✔
16	from localstack.services.stepfunctions.asl.component.common.path.items_path import ItemsPath	1✔
17	from localstack.services.stepfunctions.asl.component.common.path.result_path import ResultPath	1✔
18	from localstack.services.stepfunctions.asl.component.common.query_language import QueryLanguage	1✔
19	from localstack.services.stepfunctions.asl.component.common.result_selector import ResultSelector	1✔
20	from localstack.services.stepfunctions.asl.component.state.state import CommonStateField	1✔
21	from localstack.services.stepfunctions.asl.component.state.state_choice.state_choice import (	1✔
22	StateChoice,
23	)
24	from localstack.services.stepfunctions.asl.component.state.state_execution.state_map.max_concurrency import (	1✔
25	MaxConcurrency,
26	MaxConcurrencyJSONata,
27	MaxConcurrencyPath,
28	)
29	from localstack.services.stepfunctions.asl.component.state.state_execution.state_map.state_map import (	1✔
30	StateMap,
31	)
32	from localstack.services.stepfunctions.asl.component.state.state_execution.state_map.tolerated_failure import (	1✔
33	ToleratedFailureCountInt,
34	ToleratedFailureCountPath,
35	ToleratedFailureCountStringJSONata,
36	ToleratedFailurePercentage,
37	ToleratedFailurePercentagePath,
38	ToleratedFailurePercentageStringJSONata,
39	)
40	from localstack.services.stepfunctions.asl.component.state.state_execution.state_parallel.state_parallel import (	1✔
41	StateParallel,
42	)
43	from localstack.services.stepfunctions.asl.component.state.state_execution.state_task.state_task import (	1✔
44	StateTask,
45	)
46	from localstack.services.stepfunctions.asl.component.state.state_fail.state_fail import StateFail	1✔
47	from localstack.services.stepfunctions.asl.component.state.state_pass.result import Result	1✔
48	from localstack.services.stepfunctions.asl.component.state.state_pass.state_pass import StatePass	1✔
49	from localstack.services.stepfunctions.asl.component.state.state_succeed.state_succeed import (	1✔
50	StateSucceed,
51	)
52	from localstack.services.stepfunctions.asl.component.test_state.program.test_state_program import (	1✔
53	TestStateProgram,
54	)
55	from localstack.services.stepfunctions.asl.component.test_state.state.common import (	1✔
56	MockedCommonState,
57	)
58	from localstack.services.stepfunctions.asl.component.test_state.state.map import (	1✔
59	MockedStateMap,
60	)
61	from localstack.services.stepfunctions.asl.component.test_state.state.parallel import (	1✔
62	MockedStateParallel,
63	)
64	from localstack.services.stepfunctions.asl.component.test_state.state.task import (	1✔
65	MockedStateTask,
66	)
67	from localstack.services.stepfunctions.asl.component.test_state.state.test_state_state_props import (	1✔
68	TestStateStateProps,
69	)
70	from localstack.services.stepfunctions.asl.eval.test_state.environment import TestStateEnvironment	1✔
71	from localstack.services.stepfunctions.asl.parse.preprocessor import Preprocessor	1✔
72	from localstack.services.stepfunctions.asl.utils.encoding import to_json_str	1✔
73
74
75	class InspectionDataKey(enum.Enum):	1✔
76	INPUT = "input"	1✔
77	AFTER_INPUT_PATH = "afterInputPath"	1✔
78	AFTER_PARAMETERS = "afterParameters"	1✔
79	AFTER_ARGUMENTS = "afterArguments"	1✔
80	RESULT = "result"	1✔
81	AFTER_RESULT_SELECTOR = "afterResultSelector"	1✔
82	AFTER_RESULT_PATH = "afterResultPath"	1✔
83	AFTER_ITEMS_PATH = "afterItemsPath"	1✔
84	REQUEST = "request"	1✔
85	RESPONSE = "response"	1✔
86
87	MAX_CONCURRENCY = "maxConcurrency"	1✔
88	TOLERATED_FAILURE_COUNT = "toleratedFailureCount"	1✔
89	TOLERATED_FAILURE_PERCENTAGE = "toleratedFailurePercentage"	1✔
90
91
92	def _decorated_updates_inspection_data(method, inspection_data_key: InspectionDataKey):	1✔
UNCOV 93	def wrapper(env: TestStateEnvironment, args, *kwargs):	×
UNCOV 94	method(env, args, *kwargs)	×
UNCOV 95	result = env.stack[-1]	×
UNCOV 96	if not isinstance(result, (int, float)):	×
UNCOV 97	result = to_json_str(result)	×
98	# We know that the enum value used here corresponds to a supported inspection data field by design.
UNCOV 99	env.inspection_data[inspection_data_key.value] = result # noqa	×
100
UNCOV 101	return wrapper	×
102
103
104	def _decorate_state_field(state_field: CommonStateField, is_single_state: bool = False) -> None:	1✔
UNCOV 105	if isinstance(state_field, StateMap):	×
UNCOV 106	MockedStateMap.wrap(state_field, is_single_state)	×
UNCOV 107	elif isinstance(state_field, StateParallel):	×
UNCOV 108	MockedStateParallel.wrap(state_field, is_single_state)	×
UNCOV 109	elif isinstance(state_field, StateTask):	×
UNCOV 110	MockedStateTask.wrap(state_field, is_single_state)	×
UNCOV 111	elif isinstance(state_field, (StateChoice, StatePass, StateFail, StateSucceed)):	×
UNCOV 112	MockedCommonState.wrap(state_field, is_single_state)	×
113
114
115	def find_state(state_name: str, states: dict[str, CommonStateField]) -> CommonStateField \| None:	1✔
UNCOV 116	if state_name in states:	×
UNCOV 117	return states[state_name]	×
118
UNCOV 119	for state in states.values():	×
UNCOV 120	if isinstance(state, StateMap):	×
UNCOV 121	found_state = find_state(state_name, state.iteration_component._states.states)	×
UNCOV 122	if found_state:	×
UNCOV 123	return found_state	×
UNCOV 124	elif isinstance(state, StateParallel):	×
UNCOV 125	for program in state.branches.programs:	×
UNCOV 126	found_state = find_state(state_name, program.states.states)	×
UNCOV 127	if found_state:	×
UNCOV 128	return found_state	×
129
130
131	class TestStatePreprocessor(Preprocessor):	1✔
132	STATE_NAME: Final[str] = "StateName"	1✔
133	_state_name_stack: list[str] = []	1✔
134
135	def to_test_state_program(	1✔
136	self, tree: ParseTree, state_name: str \| None = None
137	) -> TestStateProgram:
UNCOV 138	if is_production(tree, ASLParser.RULE_state_machine):	×
139	# full definition passed in
UNCOV 140	program = self.visitState_machine(ctx=tree)	×
UNCOV 141	state_field = find_state(state_name, program.states.states)	×
UNCOV 142	_decorate_state_field(state_field, False)	×
UNCOV 143	return TestStateProgram(state_field)	×
144
UNCOV 145	if is_production(tree, ASLParser.RULE_state_decl_body):	×
146	# single state case
UNCOV 147	state_props = self.visitState_decl_body(ctx=tree)	×
UNCOV 148	state_field = self._common_state_field_of(state_props=state_props)	×
UNCOV 149	_decorate_state_field(state_field, True)	×
UNCOV 150	return TestStateProgram(state_field)	×
151
UNCOV 152	return super().visit(tree)	×
153
154	def visitState_decl(self, ctx: ASLParser.State_declContext) -> CommonStateField:	1✔
155	# if we are parsing a full state machine, we need to record the state_name prior to stepping
156	# into the state body definition.
UNCOV 157	state_name = self._inner_string_of(parser_rule_context=ctx.string_literal())	×
UNCOV 158	self._state_name_stack.append(state_name)	×
UNCOV 159	state_props: TestStateStateProps = self.visit(ctx.state_decl_body())	×
UNCOV 160	state_field = self._common_state_field_of(state_props=state_props)	×
UNCOV 161	return state_field	×
162
163	def visitState_decl_body(self, ctx: ASLParser.State_decl_bodyContext) -> TestStateStateProps:	1✔
UNCOV 164	self._open_query_language_scope(ctx)	×
UNCOV 165	state_props = TestStateStateProps()	×
UNCOV 166	state_props.name = (	×
167	self._state_name_stack.pop(-1) if self._state_name_stack else self.STATE_NAME
168	)
UNCOV 169	for child in ctx.children:	×
UNCOV 170	cmp = self.visit(child)	×
UNCOV 171	state_props.add(cmp)	×
UNCOV 172	if state_props.get(QueryLanguage) is None:	×
UNCOV 173	state_props.add(self._get_current_query_language())	×
UNCOV 174	self._close_query_language_scope()	×
UNCOV 175	return state_props	×
176
177	def visitInput_path_decl(self, ctx: ASLParser.Input_path_declContext) -> InputPath:	1✔
UNCOV 178	input_path: InputPath = super().visitInput_path_decl(ctx=ctx)	×
UNCOV 179	input_path._eval_body = _decorated_updates_inspection_data(	×
180	method=input_path._eval_body, # noqa
181	inspection_data_key=InspectionDataKey.AFTER_INPUT_PATH,
182	)
UNCOV 183	return input_path	×
184
185	def visitParameters_decl(self, ctx: ASLParser.Parameters_declContext) -> Parameters:	1✔
UNCOV 186	parameters: Parameters = super().visitParameters_decl(ctx=ctx)	×
UNCOV 187	parameters._eval_body = _decorated_updates_inspection_data(	×
188	method=parameters._eval_body, # noqa
189	inspection_data_key=InspectionDataKey.AFTER_PARAMETERS,
190	)
UNCOV 191	return parameters	×
192
193	def visitResult_selector_decl(	1✔
194	self, ctx: ASLParser.Result_selector_declContext
195	) -> ResultSelector:
UNCOV 196	result_selector: ResultSelector = super().visitResult_selector_decl(ctx=ctx)	×
UNCOV 197	result_selector._eval_body = _decorated_updates_inspection_data(	×
198	method=result_selector._eval_body, # noqa
199	inspection_data_key=InspectionDataKey.AFTER_RESULT_SELECTOR,
200	)
UNCOV 201	return result_selector	×
202
203	def visitResult_path_decl(self, ctx: ASLParser.Result_path_declContext) -> ResultPath:	1✔
UNCOV 204	result_path: ResultPath = super().visitResult_path_decl(ctx=ctx)	×
UNCOV 205	result_path._eval_body = _decorated_updates_inspection_data(	×
206	method=result_path._eval_body, # noqa
207	inspection_data_key=InspectionDataKey.AFTER_RESULT_PATH,
208	)
UNCOV 209	return result_path	×
210
211	def visitResult_decl(self, ctx: ASLParser.Result_declContext) -> Result:	1✔
UNCOV 212	result: Result = super().visitResult_decl(ctx=ctx)	×
UNCOV 213	result._eval_body = _decorated_updates_inspection_data(	×
214	method=result._eval_body,
215	inspection_data_key=InspectionDataKey.RESULT, # noqa
216	)
UNCOV 217	return result	×
218
219	def visitMax_concurrency_int(self, ctx: ASLParser.Max_concurrency_intContext) -> MaxConcurrency:	1✔
UNCOV 220	max_concurrency: MaxConcurrency = super().visitMax_concurrency_int(ctx)	×
UNCOV 221	max_concurrency._eval_body = _decorated_updates_inspection_data(	×
222	method=max_concurrency._eval_body,
223	inspection_data_key=InspectionDataKey.MAX_CONCURRENCY, # noqa
224	)
UNCOV 225	return max_concurrency	×
226
227	def visitMax_concurrency_jsonata(	1✔
228	self, ctx: ASLParser.Max_concurrency_jsonataContext
229	) -> MaxConcurrencyJSONata:
UNCOV 230	max_concurrency_jsonata: MaxConcurrencyJSONata = super().visitMax_concurrency_jsonata(ctx)	×
UNCOV 231	max_concurrency_jsonata._eval_body = _decorated_updates_inspection_data(	×
232	method=max_concurrency_jsonata._eval_body,
233	inspection_data_key=InspectionDataKey.MAX_CONCURRENCY, # noqa
234	)
UNCOV 235	return max_concurrency_jsonata	×
236
237	def visitMax_concurrency_path(	1✔
238	self, ctx: ASLParser.Max_concurrency_declContext
239	) -> MaxConcurrencyPath:
UNCOV 240	max_concurrency_path: MaxConcurrencyPath = super().visitMax_concurrency_path(ctx)	×
UNCOV 241	max_concurrency_path._eval_body = _decorated_updates_inspection_data(	×
242	method=max_concurrency_path._eval_body,
243	inspection_data_key=InspectionDataKey.MAX_CONCURRENCY, # noqa
244	)
245	return max_concurrency_path	×
246
247	def visitTolerated_failure_count_int(self, ctx) -> ToleratedFailureCountInt:	1✔
248	tolerated_failure_count: ToleratedFailureCountInt = (	×
249	super().visitTolerated_failure_count_int(ctx)
250	)
UNCOV 251	tolerated_failure_count._eval_body = _decorated_updates_inspection_data(	×
252	method=tolerated_failure_count._eval_body,
253	inspection_data_key=InspectionDataKey.TOLERATED_FAILURE_COUNT,
254	)
UNCOV 255	return tolerated_failure_count	×
256
257	def visitTolerated_failure_count_path(self, ctx) -> ToleratedFailureCountPath:	1✔
UNCOV 258	tolerated_failure_count_path: ToleratedFailureCountPath = (	×
259	super().visitTolerated_failure_count_path(ctx)
260	)
UNCOV 261	tolerated_failure_count_path._eval_body = _decorated_updates_inspection_data(	×
262	method=tolerated_failure_count_path._eval_body,
263	inspection_data_key=InspectionDataKey.TOLERATED_FAILURE_COUNT,
264	)
UNCOV 265	return tolerated_failure_count_path	×
266
267	def visitTolerated_failure_count_string_jsonata(	1✔
268	self, ctx
269	) -> ToleratedFailureCountStringJSONata:
270	tolerated_failure_count_jsonata: ToleratedFailureCountStringJSONata = (	×
271	super().visitTolerated_failure_count_string_jsonata(ctx)
272	)
UNCOV 273	tolerated_failure_count_jsonata._eval_body = _decorated_updates_inspection_data(	×
274	method=tolerated_failure_count_jsonata._eval_body,
275	inspection_data_key=InspectionDataKey.TOLERATED_FAILURE_COUNT,
276	)
277	return tolerated_failure_count_jsonata	×
278
279	def visitTolerated_failure_percentage_number(self, ctx) -> ToleratedFailurePercentage:	1✔
280	tolerated_failure_percentage: ToleratedFailurePercentage = (	×
281	super().visitTolerated_failure_percentage_number(ctx)
282	)
UNCOV 283	tolerated_failure_percentage._eval_body = _decorated_updates_inspection_data(	×
284	method=tolerated_failure_percentage._eval_body,
285	inspection_data_key=InspectionDataKey.TOLERATED_FAILURE_PERCENTAGE,
286	)
UNCOV 287	return tolerated_failure_percentage	×
288
289	def visitTolerated_failure_percentage_path(self, ctx) -> ToleratedFailurePercentagePath:	1✔
UNCOV 290	tolerated_failure_percentage_path: ToleratedFailurePercentagePath = (	×
291	super().visitTolerated_failure_percentage_path(ctx)
292	)
UNCOV 293	tolerated_failure_percentage_path._eval_body = _decorated_updates_inspection_data(	×
294	method=tolerated_failure_percentage_path._eval_body,
295	inspection_data_key=InspectionDataKey.TOLERATED_FAILURE_PERCENTAGE,
296	)
UNCOV 297	return tolerated_failure_percentage_path	×
298
299	def visitTolerated_failure_percentage_string_jsonata(	1✔
300	self, ctx
301	) -> ToleratedFailurePercentageStringJSONata:
UNCOV 302	tolerated_failure_percentage_jsonata: ToleratedFailurePercentageStringJSONata = (	×
303	super().visitTolerated_failure_percentage_string_jsonata(ctx)
304	)
UNCOV 305	tolerated_failure_percentage_jsonata._eval_body = _decorated_updates_inspection_data(	×
306	method=tolerated_failure_percentage_jsonata._eval_body,
307	inspection_data_key=InspectionDataKey.TOLERATED_FAILURE_PERCENTAGE,
308	)
UNCOV 309	return tolerated_failure_percentage_jsonata	×
310
311	def visitItems_path_decl(self, ctx) -> ItemsPath:	1✔
312	items_path: ItemsPath = super().visitItems_path_decl(ctx)	×
UNCOV 313	items_path._eval_body = _decorated_updates_inspection_data(	×
314	method=items_path._eval_body,
315	inspection_data_key=InspectionDataKey.AFTER_ITEMS_PATH,
316	)
UNCOV 317	return items_path	×
318
319	def visitArguments_string_jsonata(self, ctx):	1✔
UNCOV 320	arguments: ArgumentsStringJSONata = super().visitArguments_string_jsonata(ctx)	×
UNCOV 321	arguments._eval_body = _decorated_updates_inspection_data(	×
322	method=arguments._eval_body,
323	inspection_data_key=InspectionDataKey.AFTER_ARGUMENTS,
324	)
UNCOV 325	return arguments	×
326
327	def visitArguments_jsonata_template_value_object(self, ctx):	1✔
UNCOV 328	arguments: ArgumentsJSONataTemplateValueObject = (	×
329	super().visitArguments_jsonata_template_value_object(ctx)
330	)
UNCOV 331	arguments._eval_body = _decorated_updates_inspection_data(	×
332	method=arguments._eval_body,
333	inspection_data_key=InspectionDataKey.AFTER_ARGUMENTS,
334	)
UNCOV 335	return arguments	×

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