Coverage for /home/runner/work/bijux-cli/bijux-cli/src/bijux_cli/commands/config/set.py: 100%

1# SPDX-License-Identifier: MIT 

2# Copyright © 2025 Bijan Mousavi 

3 

4"""Implements the `config set` subcommand for the Bijux CLI. 

5 

6This module contains the logic for creating or updating a key-value pair in 

7the active configuration store. It accepts input either as a direct argument 

8or from stdin, performs strict validation on keys and values, and provides a 

9structured, machine-readable response. 

10 

11Output Contract: 

12 * Success: `{"status": "updated", "key": str, "value": str}` 

13 * Verbose: Adds `{"python": str, "platform": str}` to the payload. 

14 * Error: `{"error": str, "code": int}` 

15 

16Exit Codes: 

17 * `0`: Success. 

18 * `1`: An unexpected error occurred, such as a file lock or write failure. 

19 * `2`: An invalid argument was provided (e.g., malformed pair, invalid key). 

20 * `3`: The key, value, or configuration path contained non-ASCII or forbidden 

21 control characters. 

22""" 

23 

24from __future__ import annotations 

25 

26from contextlib import suppress 

27import os 

28import platform 

29import re 

30import string 

31import sys 

32 

33import typer 

34 

35from bijux_cli.commands.utilities import ( 

36 ascii_safe, 

37 emit_error_and_exit, 

38 new_run_command, 

39 parse_global_flags, 

40) 

41from bijux_cli.contracts import ConfigProtocol 

42from bijux_cli.core.constants import ( 

43 HELP_DEBUG, 

44 HELP_FORMAT, 

45 HELP_NO_PRETTY, 

46 HELP_QUIET, 

47 HELP_VERBOSE, 

48) 

49from bijux_cli.core.di import DIContainer 

50 

51 

52def set_config( 

53 ctx: typer.Context, 

54 pair: str | None = typer.Argument( 

55 None, help="KEY=VALUE to set; if omitted, read from stdin" 

56 ), 

57 quiet: bool = typer.Option(False, "-q", "--quiet", help=HELP_QUIET), 

58 verbose: bool = typer.Option(False, "-v", "--verbose", help=HELP_VERBOSE), 

59 fmt: str = typer.Option("json", "-f", "--format", help=HELP_FORMAT), 

60 pretty: bool = typer.Option(True, "--pretty/--no-pretty", help=HELP_NO_PRETTY), 

61 debug: bool = typer.Option(False, "-d", "--debug", help=HELP_DEBUG), 

62) -> None: 

63 """Sets or updates a configuration key-value pair. 

64 

65 This function orchestrates the `set` operation. It accepts a `KEY=VALUE` 

66 pair from either a command-line argument or standard input. It performs 

67 extensive validation on the key and value for format and content, handles 

68 file locking to prevent race conditions, and emits a structured payload 

69 confirming the update. 

70 

71 Args: 

72 ctx (typer.Context): The Typer context for the CLI. 

73 pair (str | None): A string in "KEY=VALUE" format. If None, the pair 

74 is read from stdin. 

75 quiet (bool): If True, suppresses all output except for errors. 

76 verbose (bool): If True, includes Python/platform details in the output. 

77 fmt (str): The output format, "json" or "yaml". 

78 pretty (bool): If True, pretty-prints the output. 

79 debug (bool): If True, enables debug diagnostics. 

80 

81 Returns: 

82 None: 

83 

84 Raises: 

85 SystemExit: Always exits with a contract-compliant status code and 

86 payload, indicating success or detailing the error. 

87 """ 

88 cfg_path = os.environ.get("BIJUXCLI_CONFIG", "") or "" 

89 if cfg_path: 

90 try: 

91 cfg_path.encode("ascii") 

92 except UnicodeEncodeError: 

93 emit_error_and_exit( 

94 "Non-ASCII characters in config path", 

95 code=3, 

96 failure="ascii", 

97 command="config set", 

98 fmt="json", 

99 quiet=False, 

100 include_runtime=False, 

101 debug=False, 

102 extra={"path": "[non-ascii path provided]"}, 

103 ) 

104 flags = parse_global_flags() 

105 quiet = flags["quiet"] 

106 verbose = flags["verbose"] 

107 fmt = flags["format"] 

108 pretty = flags["pretty"] 

109 debug = flags["debug"] 

110 include_runtime = verbose 

111 fmt_lower = fmt.lower() 

112 command = "config set" 

113 if os.name == "posix": 

114 with suppress(Exception): 

115 import fcntl 

116 

117 with open(cfg_path, "a+") as fh: 

118 try: 

119 fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB) 

120 except OSError: 

121 emit_error_and_exit( 

122 "Config file is locked", 

123 code=1, 

124 failure="file_locked", 

125 command=command, 

126 fmt=fmt_lower, 

127 quiet=quiet, 

128 include_runtime=include_runtime, 

129 debug=debug, 

130 extra={"path": cfg_path}, 

131 ) 

132 finally: 

133 with suppress(Exception): 

134 fcntl.flock(fh, fcntl.LOCK_UN) 

135 if pair is None: 

136 if sys.stdin.isatty(): 

137 emit_error_and_exit( 

138 "Missing argument: KEY=VALUE required", 

139 code=2, 

140 failure="missing_argument", 

141 command=command, 

142 fmt=fmt_lower, 

143 quiet=quiet, 

144 include_runtime=include_runtime, 

145 debug=debug, 

146 ) 

147 pair = sys.stdin.read().rstrip("\n") 

148 if not pair or "=" not in pair: 

149 emit_error_and_exit( 

150 "Invalid argument: KEY=VALUE required", 

151 code=2, 

152 failure="invalid_argument", 

153 command=command, 

154 fmt=fmt_lower, 

155 quiet=quiet, 

156 include_runtime=include_runtime, 

157 debug=debug, 

158 ) 

159 raw_key, raw_value = pair.split("=", 1) 

160 key = raw_key.strip() 

161 service_value_str = raw_value 

162 if len(service_value_str) >= 2 and ( 

163 (service_value_str[0] == service_value_str[-1] == '"') 

164 or (service_value_str[0] == service_value_str[-1] == "'") 

165 ): 

166 import codecs 

167 

168 service_value_str = codecs.decode(service_value_str[1:-1], "unicode_escape") 

169 if not key: 

170 emit_error_and_exit( 

171 "Key cannot be empty", 

172 code=2, 

173 failure="empty_key", 

174 command=command, 

175 fmt=fmt_lower, 

176 quiet=quiet, 

177 include_runtime=include_runtime, 

178 debug=debug, 

179 ) 

180 if not all(ord(c) < 128 for c in key + service_value_str): 

181 emit_error_and_exit( 

182 "Non-ASCII characters are not allowed in keys or values.", 

183 code=3, 

184 failure="ascii_error", 

185 command=command, 

186 fmt=fmt_lower, 

187 quiet=quiet, 

188 include_runtime=include_runtime, 

189 debug=debug, 

190 extra={"key": key}, 

191 ) 

192 if not re.match(r"^[A-Za-z0-9_]+$", key): 

193 emit_error_and_exit( 

194 "Invalid key: only alphanumerics and underscore allowed.", 

195 code=2, 

196 failure="invalid_key", 

197 command=command, 

198 fmt=fmt_lower, 

199 quiet=quiet, 

200 include_runtime=include_runtime, 

201 debug=debug, 

202 extra={"key": key}, 

203 ) 

204 if not all( 

205 c in string.printable and c not in "\r\n\t\x0b\x0c" for c in service_value_str 

206 ): 

207 emit_error_and_exit( 

208 "Control characters are not allowed in config values.", 

209 code=3, 

210 failure="control_char_error", 

211 command=command, 

212 fmt=fmt_lower, 

213 quiet=quiet, 

214 include_runtime=include_runtime, 

215 debug=debug, 

216 extra={"key": key}, 

217 ) 

218 config_svc = DIContainer.current().resolve(ConfigProtocol) 

219 try: 

220 config_svc.set(key, service_value_str) 

221 except Exception as exc: 

222 emit_error_and_exit( 

223 f"Failed to set config: {exc}", 

224 code=1, 

225 failure="set_failed", 

226 command=command, 

227 fmt=fmt_lower, 

228 quiet=quiet, 

229 include_runtime=include_runtime, 

230 debug=debug, 

231 ) 

232 

233 def payload_builder(include_runtime: bool) -> dict[str, object]: 

234 """Builds the payload confirming a key was set or updated. 

235 

236 Args: 

237 include_runtime (bool): If True, includes Python and platform info. 

238 

239 Returns: 

240 dict[str, object]: The structured payload. 

241 """ 

242 payload: dict[str, object] = { 

243 "status": "updated", 

244 "key": key, 

245 "value": service_value_str, 

246 } 

247 if include_runtime: 

248 payload["python"] = ascii_safe(platform.python_version(), "python_version") 

249 payload["platform"] = ascii_safe(platform.platform(), "platform") 

250 return payload 

251 

252 new_run_command( 

253 command_name=command, 

254 payload_builder=payload_builder, 

255 quiet=quiet, 

256 verbose=verbose, 

257 fmt=fmt_lower, 

258 pretty=pretty, 

259 debug=debug, 

260 )