16820655284

Committed 07 Aug 2025 05:03PM UTC coverage: 86.841% (-0.05%) from 86.892%

Build # 16820655284

Build Type

push

github

Committed by

web-flow

Commit Message

CFNV2: support CDK bootstrap and deployment (#12967)

Coverage Stats

32 of 38 new or added lines in 5 files covered. (84.21%)

2013 existing lines in 125 files now uncovered.

66606 of 76699 relevant lines covered (86.84%)

0.87 hits per line

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

96.1

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

import re

from antlr4.tree.Tree import ParseTree, TerminalNodeImpl

from localstack.services.stepfunctions.asl.antlr.runtime.ASLIntrinsicLexer import ASLIntrinsicLexer
from localstack.services.stepfunctions.asl.antlr.runtime.ASLIntrinsicParser import (
    ASLIntrinsicParser,
)
from localstack.services.stepfunctions.asl.antlr.runtime.ASLIntrinsicParserVisitor import (
    ASLIntrinsicParserVisitor,
)
from localstack.services.stepfunctions.asl.antlt4utils.antlr4utils import (
    is_production,
    is_terminal,
)
from localstack.services.stepfunctions.asl.component.common.query_language import (
    QueryLanguageMode,
)
from localstack.services.stepfunctions.asl.component.common.string.string_expression import (
    StringVariableSample,
)
from localstack.services.stepfunctions.asl.component.component import Component
from localstack.services.stepfunctions.asl.component.intrinsic.argument.argument import (
    Argument,
    ArgumentContextPath,
    ArgumentFunction,
    ArgumentJsonPath,
    ArgumentList,
    ArgumentLiteral,
    ArgumentVar,
)
from localstack.services.stepfunctions.asl.component.intrinsic.function.function import Function
from localstack.services.stepfunctions.asl.component.intrinsic.function.statesfunction.factory import (
    StatesFunctionFactory,
)
from localstack.services.stepfunctions.asl.component.intrinsic.function.statesfunction.states_function import (
    StatesFunction,
)
from localstack.services.stepfunctions.asl.component.intrinsic.functionname.state_function_name_types import (
    StatesFunctionNameType,
)
from localstack.services.stepfunctions.asl.component.intrinsic.functionname.states_function_name import (
    StatesFunctionName,
)


class Preprocessor(ASLIntrinsicParserVisitor):
    @staticmethod
    def _replace_escaped_characters(match):
        escaped_char = match.group(1)
        if escaped_char.isalpha():
            replacements = {"n": "\n", "t": "\t", "r": "\r"}
            return replacements.get(escaped_char, escaped_char)
        elif escaped_char == '"':
            return '"'
        else:
            return match.group(0)

    @staticmethod
    def _text_of_str(parse_tree: ParseTree) -> str:
        pt = is_production(parse_tree) or is_terminal(parse_tree)
        inner_str = pt.getText()
        inner_str = inner_str[1:-1]
        inner_str = re.sub(r"\\(.)", Preprocessor._replace_escaped_characters, inner_str)
        return inner_str

    def visitFunc_arg_int(self, ctx: ASLIntrinsicParser.Func_arg_intContext) -> ArgumentLiteral:
        integer = int(ctx.INT().getText())
        return ArgumentLiteral(definition_value=integer)

    def visitFunc_arg_float(self, ctx: ASLIntrinsicParser.Func_arg_floatContext) -> ArgumentLiteral:
        number = float(ctx.INT().getText())
        return ArgumentLiteral(definition_value=number)

    def visitFunc_arg_string(
        self, ctx: ASLIntrinsicParser.Func_arg_stringContext
    ) -> ArgumentLiteral:
        text: str = self._text_of_str(ctx.STRING())
        return ArgumentLiteral(definition_value=text)

    def visitFunc_arg_bool(self, ctx: ASLIntrinsicParser.Func_arg_boolContext) -> ArgumentLiteral:
        bool_term: TerminalNodeImpl = ctx.children[0]
        bool_term_rule: int = bool_term.getSymbol().type
        bool_val: bool = bool_term_rule == ASLIntrinsicLexer.TRUE
        return ArgumentLiteral(definition_value=bool_val)

    def visitFunc_arg_list(self, ctx: ASLIntrinsicParser.Func_arg_listContext) -> ArgumentList:
        arguments: list[Argument] = list()
        for child in ctx.children:
            cmp: Component | None = self.visit(child)
            if isinstance(cmp, Argument):
                arguments.append(cmp)
        return ArgumentList(arguments=arguments)

    def visitFunc_arg_context_path(
        self, ctx: ASLIntrinsicParser.Func_arg_context_pathContext
    ) -> ArgumentContextPath:
        context_path: str = ctx.CONTEXT_PATH_STRING().getText()
        return ArgumentContextPath(context_path=context_path)

    def visitFunc_arg_json_path(
        self, ctx: ASLIntrinsicParser.Func_arg_json_pathContext
    ) -> ArgumentJsonPath:
        json_path: str = ctx.JSON_PATH_STRING().getText()
        return ArgumentJsonPath(json_path=json_path)

    def visitFunc_arg_var(self, ctx: ASLIntrinsicParser.Func_arg_varContext) -> ArgumentVar:
        expression: str = ctx.STRING_VARIABLE().getText()
        string_variable_sample = StringVariableSample(
            query_language_mode=QueryLanguageMode.JSONPath, expression=expression
        )
        return ArgumentVar(string_variable_sample=string_variable_sample)

    def visitFunc_arg_func_decl(
        self, ctx: ASLIntrinsicParser.Func_arg_func_declContext
    ) -> ArgumentFunction:
        function: Function = self.visit(ctx.states_func_decl())
        return ArgumentFunction(function=function)

    def visitState_fun_name(
        self, ctx: ASLIntrinsicParser.State_fun_nameContext
    ) -> StatesFunctionName:
        tok_typ: int = ctx.children[0].symbol.type
        name_typ = StatesFunctionNameType(tok_typ)
        return StatesFunctionName(function_type=name_typ)

    def visitStates_func_decl(
        self, ctx: ASLIntrinsicParser.States_func_declContext
    ) -> StatesFunction:
        func_name: StatesFunctionName = self.visit(ctx.state_fun_name())
        argument_list: ArgumentList = self.visit(ctx.func_arg_list())
        func: StatesFunction = StatesFunctionFactory.from_name(
            func_name=func_name, argument_list=argument_list
        )
        return func

    def visitFunc_decl(self, ctx: ASLIntrinsicParser.Func_declContext) -> Function:
        return self.visit(ctx.children[0])

1	import re	1✔
2
3	from antlr4.tree.Tree import ParseTree, TerminalNodeImpl	1✔
4
5	from localstack.services.stepfunctions.asl.antlr.runtime.ASLIntrinsicLexer import ASLIntrinsicLexer	1✔
6	from localstack.services.stepfunctions.asl.antlr.runtime.ASLIntrinsicParser import (	1✔
7	ASLIntrinsicParser,
8	)
9	from localstack.services.stepfunctions.asl.antlr.runtime.ASLIntrinsicParserVisitor import (	1✔
10	ASLIntrinsicParserVisitor,
11	)
12	from localstack.services.stepfunctions.asl.antlt4utils.antlr4utils import (	1✔
13	is_production,
14	is_terminal,
15	)
16	from localstack.services.stepfunctions.asl.component.common.query_language import (	1✔
17	QueryLanguageMode,
18	)
19	from localstack.services.stepfunctions.asl.component.common.string.string_expression import (	1✔
20	StringVariableSample,
21	)
22	from localstack.services.stepfunctions.asl.component.component import Component	1✔
23	from localstack.services.stepfunctions.asl.component.intrinsic.argument.argument import (	1✔
24	Argument,
25	ArgumentContextPath,
26	ArgumentFunction,
27	ArgumentJsonPath,
28	ArgumentList,
29	ArgumentLiteral,
30	ArgumentVar,
31	)
32	from localstack.services.stepfunctions.asl.component.intrinsic.function.function import Function	1✔
33	from localstack.services.stepfunctions.asl.component.intrinsic.function.statesfunction.factory import (	1✔
34	StatesFunctionFactory,
35	)
36	from localstack.services.stepfunctions.asl.component.intrinsic.function.statesfunction.states_function import (	1✔
37	StatesFunction,
38	)
39	from localstack.services.stepfunctions.asl.component.intrinsic.functionname.state_function_name_types import (	1✔
40	StatesFunctionNameType,
41	)
42	from localstack.services.stepfunctions.asl.component.intrinsic.functionname.states_function_name import (	1✔
43	StatesFunctionName,
44	)
45
46
47	class Preprocessor(ASLIntrinsicParserVisitor):	1✔
48	@staticmethod	1✔
49	def _replace_escaped_characters(match):	1✔
50	escaped_char = match.group(1)	1✔
51	if escaped_char.isalpha():	1✔
52	replacements = {"n": "\n", "t": "\t", "r": "\r"}	1✔
53	return replacements.get(escaped_char, escaped_char)	1✔
54	elif escaped_char == '"':	1✔
55	return '"'	1✔
56	else:
UNCOV 57	return match.group(0)	×
58
59	@staticmethod	1✔
60	def _text_of_str(parse_tree: ParseTree) -> str:	1✔
61	pt = is_production(parse_tree) or is_terminal(parse_tree)	1✔
62	inner_str = pt.getText()	1✔
63	inner_str = inner_str[1:-1]	1✔
64	inner_str = re.sub(r"\\(.)", Preprocessor._replace_escaped_characters, inner_str)	1✔
65	return inner_str	1✔
66
67	def visitFunc_arg_int(self, ctx: ASLIntrinsicParser.Func_arg_intContext) -> ArgumentLiteral:	1✔
68	integer = int(ctx.INT().getText())	1✔
69	return ArgumentLiteral(definition_value=integer)	1✔
70
71	def visitFunc_arg_float(self, ctx: ASLIntrinsicParser.Func_arg_floatContext) -> ArgumentLiteral:	1✔
UNCOV 72	number = float(ctx.INT().getText())	×
73	return ArgumentLiteral(definition_value=number)	×
74
75	def visitFunc_arg_string(	1✔
76	self, ctx: ASLIntrinsicParser.Func_arg_stringContext
77	) -> ArgumentLiteral:
78	text: str = self._text_of_str(ctx.STRING())	1✔
79	return ArgumentLiteral(definition_value=text)	1✔
80
81	def visitFunc_arg_bool(self, ctx: ASLIntrinsicParser.Func_arg_boolContext) -> ArgumentLiteral:	1✔
82	bool_term: TerminalNodeImpl = ctx.children[0]	1✔
83	bool_term_rule: int = bool_term.getSymbol().type	1✔
84	bool_val: bool = bool_term_rule == ASLIntrinsicLexer.TRUE	1✔
85	return ArgumentLiteral(definition_value=bool_val)	1✔
86
87	def visitFunc_arg_list(self, ctx: ASLIntrinsicParser.Func_arg_listContext) -> ArgumentList:	1✔
88	arguments: list[Argument] = list()	1✔
89	for child in ctx.children:	1✔
90	cmp: Component \| None = self.visit(child)	1✔
91	if isinstance(cmp, Argument):	1✔
92	arguments.append(cmp)	1✔
93	return ArgumentList(arguments=arguments)	1✔
94
95	def visitFunc_arg_context_path(	1✔
96	self, ctx: ASLIntrinsicParser.Func_arg_context_pathContext
97	) -> ArgumentContextPath:
98	context_path: str = ctx.CONTEXT_PATH_STRING().getText()	1✔
99	return ArgumentContextPath(context_path=context_path)	1✔
100
101	def visitFunc_arg_json_path(	1✔
102	self, ctx: ASLIntrinsicParser.Func_arg_json_pathContext
103	) -> ArgumentJsonPath:
104	json_path: str = ctx.JSON_PATH_STRING().getText()	1✔
105	return ArgumentJsonPath(json_path=json_path)	1✔
106
107	def visitFunc_arg_var(self, ctx: ASLIntrinsicParser.Func_arg_varContext) -> ArgumentVar:	1✔
108	expression: str = ctx.STRING_VARIABLE().getText()	1✔
109	string_variable_sample = StringVariableSample(	1✔
110	query_language_mode=QueryLanguageMode.JSONPath, expression=expression
111	)
112	return ArgumentVar(string_variable_sample=string_variable_sample)	1✔
113
114	def visitFunc_arg_func_decl(	1✔
115	self, ctx: ASLIntrinsicParser.Func_arg_func_declContext
116	) -> ArgumentFunction:
117	function: Function = self.visit(ctx.states_func_decl())	1✔
118	return ArgumentFunction(function=function)	1✔
119
120	def visitState_fun_name(	1✔
121	self, ctx: ASLIntrinsicParser.State_fun_nameContext
122	) -> StatesFunctionName:
123	tok_typ: int = ctx.children[0].symbol.type	1✔
124	name_typ = StatesFunctionNameType(tok_typ)	1✔
125	return StatesFunctionName(function_type=name_typ)	1✔
126
127	def visitStates_func_decl(	1✔
128	self, ctx: ASLIntrinsicParser.States_func_declContext
129	) -> StatesFunction:
130	func_name: StatesFunctionName = self.visit(ctx.state_fun_name())	1✔
131	argument_list: ArgumentList = self.visit(ctx.func_arg_list())	1✔
132	func: StatesFunction = StatesFunctionFactory.from_name(	1✔
133	func_name=func_name, argument_list=argument_list
134	)
135	return func	1✔
136
137	def visitFunc_decl(self, ctx: ASLIntrinsicParser.Func_declContext) -> Function:	1✔
138	return self.visit(ctx.children[0])	1✔

localstack / localstack / 16820655284

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