Coverage for  / home / runner / work / bijux-cli / bijux-cli / src / bijux_cli / cli / repl / execution.py: 99%

1# SPDX-License-Identifier: Apache-2.0 

2# Copyright © 2025 Bijan Mousavi 

3 

4"""Execution helpers for the REPL.""" 

5 

6from __future__ import annotations 

7 

8from contextlib import suppress 

9import json 

10import os 

11import shlex 

12import sys 

13 

14from bijux_cli.cli.core.constants import OPT_FORMAT, OPT_QUIET, PRETTY_FLAGS 

15from bijux_cli.cli.repl.parsing import ( 

16 _filter_control, 

17 _known_commands, 

18 _split_segments, 

19 _suggest, 

20) 

21from bijux_cli.core.enums import ExitCode, OutputFormat 

22from bijux_cli.core.exit_policy import ExitIntent, ExitIntentError 

23from bijux_cli.core.runtime import run_command 

24 

25_JSON_CMDS = { 

26 "audit", 

27 "doctor", 

28 "history", 

29 "memory", 

30 "plugins", 

31 "status", 

32 "version", 

33} 

34 

35 

36def _invoke(tokens: list[str], *, repl_quiet: bool) -> int: 

37 """Runs a single CLI command invocation within the REPL sandbox.""" 

38 from importlib import import_module 

39 

40 from typer.testing import CliRunner 

41 

42 env = {**os.environ, "PS1": ""} 

43 

44 head = tokens[0] if tokens else "" 

45 

46 if head in _JSON_CMDS and not (set(PRETTY_FLAGS) | set(OPT_FORMAT)) & set(tokens): 

47 tokens.append("--no-pretty") 

48 

49 if ( 

50 head == "config" 

51 and len(tokens) > 1 

52 and tokens[1] == "list" 

53 and not (set(PRETTY_FLAGS) & set(tokens)) 

54 ): 

55 tokens.append("--no-pretty") 

56 

57 cli_root = import_module("bijux_cli.cli.root") 

58 app = getattr(cli_root, "build_app", None) 

59 typer_app = app() if callable(app) else cli_root.app 

60 result = CliRunner().invoke(typer_app, tokens, env=env) 

61 

62 sub_quiet = any(t in OPT_QUIET for t in tokens) 

63 should_print = not repl_quiet and not sub_quiet 

64 

65 if head == "history": 

66 with suppress(Exception): 

67 data = json.loads(result.stdout or "{}") 

68 if data.get("entries", []) == []: 

69 if should_print: 

70 from bijux_cli.cli.core.command import resolve_serializer 

71 

72 pretty = ( 

73 resolve_serializer() 

74 .dumps(data, fmt=OutputFormat.JSON, pretty=True) 

75 .rstrip("\n") 

76 + "\n" 

77 ) 

78 sys.stdout.write(pretty) 

79 sys.stderr.write(result.stderr or "") 

80 return result.exit_code 

81 

82 if should_print: 

83 sys.stdout.write(result.stdout or "") 

84 sys.stderr.write(result.stderr or "") 

85 

86 return result.exit_code 

87 

88 

89def _run_piped(repl_quiet: bool) -> None: 

90 """Processes piped input commands in non-interactive mode.""" 

91 for raw_line in sys.stdin.read().splitlines(): 

92 line = raw_line.rstrip() 

93 

94 if not line or line.lstrip().startswith("#"): 

95 if not repl_quiet: 

96 from bijux_cli.cli.repl.ui import get_prompt 

97 

98 sys.stderr.write(_filter_control(str(get_prompt())) + "\n") 

99 sys.stderr.flush() 

100 continue 

101 

102 if line.lstrip().startswith(";"): 

103 bad = line.lstrip(";").strip() 

104 hint = _suggest(bad) 

105 msg = f"No such command '{bad}'." + (hint or "") 

106 if not repl_quiet: 

107 print(msg, file=sys.stderr) 

108 continue 

109 

110 for seg in _split_segments(line): 

111 seg = seg.strip() 

112 if not seg or seg.startswith("#"): 

113 continue 

114 

115 lo = seg.lower() 

116 if lo in {"exit", "quit"}: 

117 raise ExitIntentError( 

118 ExitIntent( 

119 code=ExitCode.SUCCESS, 

120 stream=None, 

121 payload=None, 

122 fmt=OutputFormat.JSON, 

123 pretty=False, 

124 show_traceback=False, 

125 ) 

126 ) 

127 

128 if seg == "docs": 

129 if not repl_quiet: 

130 print("Available topics: …") 

131 continue 

132 if seg.startswith("docs "): 

133 if not repl_quiet: 

134 print(seg.split(None, 1)[1]) 

135 continue 

136 

137 if seg.startswith("-"): 

138 bad = seg.lstrip("-") 

139 hint = _suggest(bad) 

140 msg = f"No such command '{bad}'." + (hint or "") 

141 if not repl_quiet: 

142 print(msg, file=sys.stderr) 

143 continue 

144 

145 try: 

146 tokens = shlex.split(seg) 

147 except ValueError: 

148 continue 

149 if not tokens: 

150 continue 

151 

152 head = tokens[0] 

153 

154 if head == "config": 

155 sub = tokens[1:] 

156 

157 def _emit( 

158 msg: str, 

159 failure: str, 

160 subcommand: list[str] = sub, 

161 ) -> None: 

162 """Emits a JSON error for a `config` subcommand.""" 

163 if repl_quiet: 

164 return 

165 

166 error_obj = { 

167 "error": msg, 

168 "code": 2, 

169 "failure": failure, 

170 "command": f"config {subcommand[0] if subcommand else ''}".strip(), 

171 "format": "json", 

172 } 

173 from bijux_cli.cli.core.command import resolve_serializer 

174 

175 print( 

176 resolve_serializer().dumps( 

177 error_obj, fmt=OutputFormat.JSON, pretty=False 

178 ) 

179 ) 

180 

181 if not sub: 

182 pass 

183 elif sub[0] == "set" and len(sub) == 1: 

184 _emit("Missing argument: KEY=VALUE required", "missing_argument") 

185 continue 

186 elif sub[0] in {"get", "unset"} and len(sub) == 1: 186 ↛ 190line 186 didn't jump to line 190 because the condition on line 186 was always true

187 _emit("Missing argument: key required", "missing_argument") 

188 continue 

189 

190 if head not in _known_commands(): 

191 hint = _suggest(head) 

192 msg = f"No such command '{head}'." 

193 if hint: 

194 msg += hint 

195 if not repl_quiet: 

196 print(msg, file=sys.stderr) 

197 continue 

198 else: 

199 _invoke(tokens, repl_quiet=repl_quiet) 

200 

201 raise ExitIntentError( 

202 ExitIntent( 

203 code=ExitCode.SUCCESS, 

204 stream=None, 

205 payload=None, 

206 fmt=OutputFormat.JSON, 

207 pretty=False, 

208 show_traceback=False, 

209 ) 

210 ) 

211 

212 

213def run_repl_session(*, quiet: bool, stdin_isatty: bool) -> None: 

214 """Run the REPL in piped or interactive mode.""" 

215 if quiet or not stdin_isatty: 

216 _run_piped(quiet) 

217 else: 

218 from bijux_cli.cli.repl.ui import _run_interactive 

219 

220 run_command(_run_interactive)