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 EvalContext, Frame 

6 

7if t.TYPE_CHECKING: 

8 pass 

9 

10 

11class AsyncFrame(Frame): 

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

13 

14 block_frame: "AsyncFrame | None" 

15 require_output_check: bool 

16 has_known_extends: bool 

17 toplevel: bool 

18 rootlevel: bool 

19 buffer: str | None 

20 block_buffer: list[str] 

21 extended_buffer: list[str] | None 

22 require_yield: bool 

23 buffer_count: int 

24 is_async: bool 

25 

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

27 if eval_ctx is None: 

28 from jinja2.environment import Environment 

29 from jinja2.nodes import EvalContext 

30 

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

32 

33 super().__init__(eval_ctx) 

34 self.buffer = None 

35 self.block_buffer = [] 

36 self.extended_buffer = None 

37 self.block_frame = None 

38 self.require_output_check = False 

39 self.has_known_extends = False 

40 self.toplevel = False 

41 self.rootlevel = False 

42 self.require_yield = False 

43 self.buffer_count = 0 

44 self.is_async = False 

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

46 self.block_frame_id = 0 

47 

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

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

50 rv = self.__class__(self.eval_ctx) 

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

52 rv.buffer = self.buffer 

53 rv.block_buffer = self.block_buffer 

54 rv.extended_buffer = self.extended_buffer 

55 rv.eval_ctx = self.eval_ctx 

56 rv.parent = self 

57 rv.require_output_check = self.require_output_check 

58 rv.has_known_extends = self.has_known_extends 

59 rv.toplevel = self.toplevel 

60 rv.rootlevel = self.rootlevel 

61 rv.block_frame = self.block_frame 

62 rv.require_yield = self.require_yield 

63 rv.buffer_count = self.buffer_count 

64 rv.is_async = self.is_async 

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

66 rv.block_frame_id = self.block_frame_id 

67 return rv 

68 

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

70 """Inspect nodes for symbol analysis.""" 

71 if nodes: 

72 for node in nodes: 

73 self.symbols.analyze_node(node) 

74 

75 def push_scope(self) -> None: 

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

77 pass 

78 

79 def pop_scope(self) -> None: 

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

81 pass 

82 

83 def find_break(self) -> bool: 

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

85 return False 

86 

87 def find_continue(self) -> bool: 

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

89 return False 

90 

91 def inner(self) -> "AsyncFrame": 

92 """Create an inner frame.""" 

93 return self.copy()