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

1"""Interactive shell: local line editing and history, remote execution.""" 

2 

3from __future__ import annotations 

4 

5import ast 

6import codeop 

7from pathlib import Path 

8 

9from prompt_toolkit import PromptSession 

10from prompt_toolkit.history import FileHistory 

11from prompt_toolkit.lexers import PygmentsLexer 

12from pygments.lexers.python import PythonLexer 

13 

14from .cli import display_options, print_result 

15from .client import Client 

16 

17HISTORY = Path.home() / ".dev_shell_history" 

18 

19 

20async def run_repl(client: Client, session_name: str) -> int: 

21 compiler = codeop.CommandCompiler() 

22 compiler.compiler.flags |= ast.PyCF_ALLOW_TOP_LEVEL_AWAIT 

23 prompt: PromptSession[str] = PromptSession( 

24 history=FileHistory(str(HISTORY)), lexer=PygmentsLexer(PythonLexer) 

25 ) 

26 print( 

27 f"Dev Shell connected to {client.url} (session {session_name!r}). Ctrl-D to exit." 

28 ) 

29 lines: list[str] = [] 

30 while True: 

31 try: 

32 line = await prompt.prompt_async("... " if lines else ">>> ") 

33 except KeyboardInterrupt: 

34 lines.clear() 

35 continue 

36 except EOFError: 

37 return 0 

38 lines.append(line) 

39 source = "\n".join(lines) 

40 try: 

41 # None means incomplete input: keep reading continuation lines. 

42 if compiler(source, "<dev_shell>", "single") is None: 

43 continue 

44 except SyntaxError, OverflowError, ValueError: 

45 pass # send it anyway so the server reports the error consistently 

46 lines.clear() 

47 if not source.strip(): 

48 continue 

49 if is_quit_call(source): 

50 return 0 

51 result = await client.call( 

52 "dev_shell_server/exec", 

53 code=source, 

54 session=session_name, 

55 **display_options(), 

56 ) 

57 print_result(result) 

58 

59 

60def is_quit_call(source: str) -> bool: 

61 """True for a bare `quit()` or `exit()` - the standard way to leave a REPL. 

62 

63 Sending that to the server would just raise SystemExit there, caught and 

64 reported back as a confusing traceback instead of actually exiting 

65 anything, so it's handled locally like Ctrl-D instead. 

66 """ 

67 try: 

68 (stmt,) = ast.parse(source).body 

69 except SyntaxError, ValueError: 

70 return False 

71 return ( 

72 isinstance(stmt, ast.Expr) 

73 and isinstance(stmt.value, ast.Call) 

74 and isinstance(stmt.value.func, ast.Name) 

75 and stmt.value.func.id in ("quit", "exit") 

76 and not stmt.value.args 

77 and not stmt.value.keywords 

78 )