Coverage for src/dev_shell/local_session.py: 0%

1"""API client mode's exec engine: runs snippets right here in the CLI 

2process, not on a remote Home Assistant session - there's no server to send 

3code to, since API client mode talks to Home Assistant only through its 

4standard API (see api_objtree.py). Deliberately mirrors 

5custom_components/dev_shell_server/session.py's exec model (top-level await, 

6trailing-expression echo via rich, rich tracebacks) rather than importing it: 

7that module lives in a separate HACS-deployed package with its own packaging 

8boundary, and duplicating ~100 lines here is simpler than bridging it. 

9 

10Simpler than session.py in one respect: output goes straight to the real 

11terminal, so there's no print()/help() capturing-and-shipping-back machinery, 

12and rich's Console can auto-detect color/width instead of being told. 

13""" 

14 

15from __future__ import annotations 

16 

17import ast 

18import builtins 

19import inspect 

20import itertools 

21import linecache 

22import sys 

23import traceback 

24from dataclasses import dataclass 

25from typing import Any 

26 

27from rich.console import Console 

28from rich.pretty import Pretty 

29from rich.traceback import Traceback 

30 

31_cell_counter = itertools.count(1) 

32 

33_console = Console() 

34_error_console = Console(stderr=True) 

35 

36 

37@dataclass 

38class LocalSession: 

39 """A namespace that persists between `run()` calls until the process exits.""" 

40 

41 globals_: dict[str, Any] 

42 

43 def __post_init__(self) -> None: 

44 self.globals_.setdefault("__name__", "__dev_shell_api__") 

45 self.globals_.setdefault("__builtins__", builtins) 

46 

47 async def run(self, source: str) -> None: 

48 """Execute source, printing the trailing expression's value (if any) 

49 or a traceback directly to the real terminal - there's no result to 

50 ship back over a wire, so this doesn't return one.""" 

51 try: 

52 value = await self._execute(source) 

53 except BaseException as err: # noqa: BLE001 - report everything, incl. SystemExit 

54 _print_error(err) 

55 return 

56 if value is not None: 

57 self.globals_["_"] = value 

58 _console.print(Pretty(value)) 

59 

60 async def _execute(self, source: str) -> Any: 

61 # Not "<dev_shell-N>": rich.traceback refuses to show source for any 

62 # filename starting with "<" (treats it like "<stdin>"), no matter what 

63 # linecache holds. An absolute-looking path sidesteps that - rich joins a 

64 # relative one onto the cwd before the linecache lookup, which would miss. 

65 filename = f"/dev_shell/api_cell_{next(_cell_counter)}" 

66 linecache.cache[filename] = ( 

67 len(source), 

68 None, 

69 source.splitlines(keepends=True), 

70 filename, 

71 ) 

72 tree = ast.parse(source, filename, "exec") 

73 

74 last_expr = None 

75 last_stmt = tree.body[-1] if tree.body else None 

76 if isinstance(last_stmt, ast.Expr): 

77 tree.body.pop() 

78 last_expr = ast.Expression(last_stmt.value) 

79 

80 flags = ast.PyCF_ALLOW_TOP_LEVEL_AWAIT 

81 # dont_inherit=True: compile() otherwise inherits this module's own 

82 # `from __future__ import annotations`, which would make every 

83 # annotation in the user's code a plain string instead of a real value. 

84 if tree.body: 

85 await _run_code( 

86 compile(tree, filename, "exec", flags=flags, dont_inherit=True), 

87 self.globals_, 

88 ) 

89 if last_expr is None: 

90 return None 

91 return await _run_code( 

92 compile(last_expr, filename, "eval", flags=flags, dont_inherit=True), 

93 self.globals_, 

94 ) 

95 

96 

97async def _run_code(code: Any, globals_: dict[str, Any]) -> Any: 

98 result = eval(code, globals_) # nosec B307 - the whole point of a dev shell 

99 if code.co_flags & inspect.CO_COROUTINE: 

100 result = await result 

101 return result 

102 

103 

104def _print_error(err: BaseException) -> None: 

105 tb = err.__traceback__ 

106 # Drop the frames belonging to this module so the traceback starts at user code. 

107 while tb is not None and tb.tb_frame.f_code.co_filename == __file__: 

108 tb = tb.tb_next 

109 if isinstance(err, SyntaxError): 

110 # No frames worth showing for this one - plain is fine. 

111 print( 

112 "".join(traceback.format_exception_only(type(err), err)), 

113 end="", 

114 file=sys.stderr, 

115 ) 

116 return 

117 _error_console.print(Traceback.from_exception(type(err), err, tb))