Coverage for src/zapy/templating/eval.py: 100%

1import ast 

2from typing import Any, cast 

3 

4from .traceback import annotate_traceback 

5 

6_python_code = """ 

7async def __exec_wrapper(_globs): 

8 globals().update(**_globs) 

9 __placeholder 

10 return locals() 

11""" 

12 

13 

14async def exec_async(custom_code: str, _globals: dict) -> None: 

15 parsed_ast = ast.parse(_python_code) 

16 

17 def is_placeholder(node: ast.AST) -> bool: 

18 return isinstance(node, ast.Expr) and isinstance(node.value, ast.Name) and node.value.id == "__placeholder" 

19 

20 for node in ast.walk(parsed_ast): 

21 if is_placeholder(node): 

22 custom_ast = ast.parse(custom_code) 

23 node = cast(ast.Expr, node) 

24 new_node = cast(ast.expr, custom_ast) 

25 node.value = new_node 

26 

27 unparsed = ast.unparse(parsed_ast) 

28 exec(unparsed) # noqa S102 

29 

30 try: 

31 func = locals()["__exec_wrapper"] 

32 new_locals = await func(_globals) 

33 except BaseException as e: 

34 annotate_traceback(e, unparsed, location="exec_async") 

35 raise 

36 

37 _globals.update(new_locals) 

38 

39 

40def exec_sync(custom_code: str, _globals: dict) -> None: 

41 try: 

42 return exec(custom_code, _globals) # noqa S102 

43 except BaseException as e: 

44 annotate_traceback(e, custom_code, location="exec_sync") 

45 raise 

46 

47 

48def eval_sync(custom_code: str, _globals: dict) -> Any: 

49 try: 

50 return eval(custom_code, _globals) # noqa S307 

51 except BaseException as e: 

52 annotate_traceback(e, custom_code, location="eval_sync") 

53 raise