Coverage for jinja2_async_environment / compiler_modules / frame.py: 100%

1"""Async frame implementation for template compilation.""" 

2 

3import typing as t 

4 

5from jinja2.compiler import Frame 

6from jinja2.nodes import EvalContext 

7 

8if t.TYPE_CHECKING: 

9 pass 

10 

11 

12class AsyncFrame(Frame): 

13 """Async-aware frame for template compilation context.""" 

14 

15 block_frame: bool # Keep consistent with base class 

16 block_frame_ref: "AsyncFrame | None" # New attribute for async frame reference 

17 require_output_check: bool 

18 has_known_extends: bool 

19 toplevel: bool 

20 rootlevel: bool 

21 buffer: str | None 

22 block_buffer: list[str] 

23 extended_buffer: list[str] | None 

24 require_yield: bool 

25 buffer_count: int 

26 is_async: bool 

27 

28 def __init__(self, eval_ctx: EvalContext | None = None) -> None: 

29 if eval_ctx is None: 

30 from jinja2.environment import Environment 

31 from jinja2.nodes import EvalContext 

32 

33 eval_ctx = EvalContext(Environment(autoescape=True), "template") 

34 

35 super().__init__(eval_ctx) 

36 self.buffer = None 

37 self.block_buffer = [] 

38 self.extended_buffer = None 

39 self.block_frame = False 

40 self.block_frame_ref = None 

41 self.require_output_check = False 

42 self.has_known_extends = False 

43 self.toplevel = False 

44 self.rootlevel = False 

45 self.require_yield = False 

46 self.buffer_count = 0 

47 self.is_async = False 

48 self.block_counters: dict[str, int] = {} 

49 self.block_frame_id = 0 

50 

51 def copy(self) -> t.Self: 

52 """Create a copy of this frame.""" 

53 rv = self.__class__(self.eval_ctx) 

54 rv.symbols = self.symbols.copy() # noqa: FURB145 

55 rv.buffer = self.buffer 

56 rv.block_buffer = self.block_buffer 

57 rv.extended_buffer = self.extended_buffer 

58 rv.eval_ctx = self.eval_ctx 

59 rv.parent = self 

60 rv.require_output_check = self.require_output_check 

61 rv.has_known_extends = self.has_known_extends 

62 rv.toplevel = self.toplevel 

63 rv.rootlevel = self.rootlevel 

64 rv.block_frame = self.block_frame 

65 rv.block_frame_ref = self.block_frame_ref 

66 rv.require_yield = self.require_yield 

67 rv.buffer_count = self.buffer_count 

68 rv.is_async = self.is_async 

69 rv.block_counters = self.block_counters.copy() # noqa: FURB145 

70 rv.block_frame_id = self.block_frame_id 

71 return rv 

72 

73 def inspect(self, nodes: t.Any | None = None) -> None: 

74 """Inspect nodes for symbol analysis.""" 

75 if nodes: 

76 for node in nodes: 

77 self.symbols.analyze_node(node) 

78 

79 def push_scope(self) -> None: 

80 """Push a new scope (no-op for async frame).""" 

81 pass 

82 

83 def pop_scope(self) -> None: 

84 """Pop the current scope (no-op for async frame).""" 

85 pass 

86 

87 def find_break(self) -> bool: 

88 """Check if break statement is valid (always False).""" 

89 return False 

90 

91 def find_continue(self) -> bool: 

92 """Check if continue statement is valid (always False).""" 

93 return False 

94 

95 def inner(self, isolated: bool = False) -> "AsyncFrame": 

96 """Create an inner frame.""" 

97 return self.copy()