Coverage for / home / runner / work / bijux-cli / bijux-cli / src / bijux_cli / cli / commands / config / set.py: 99%
80 statements
« prev ^ index » next coverage.py v7.13.2, created at 2026-01-26 17:59 +0000
1# SPDX-License-Identifier: Apache-2.0
2# Copyright © 2025 Bijan Mousavi
4"""Implements the `config set` subcommand for the Bijux CLI.
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.
11Output Contract:
12 * Success: `{"status": "updated", "key": str, "value": str}`
13 * Error: `{"error": str, "code": int}`
15Exit Codes:
16 * `0`: Success.
17 * `1`: An unexpected error occurred, such as a file lock or write failure.
18 * `2`: An invalid argument was provided (e.g., malformed pair, invalid key).
19 * `3`: The key, value, or configuration path contained non-ASCII or forbidden
20 control characters.
21"""
23from __future__ import annotations
25from contextlib import suppress
26from dataclasses import dataclass
27import fcntl
28import os
29import platform
30import re
31import string
32import sys
34import typer
36from bijux_cli.cli.core.command import (
37 ascii_safe,
38 new_run_command,
39 raise_exit_intent,
40 validate_common_flags,
41)
42from bijux_cli.cli.core.constants import (
43 ENV_CONFIG,
44 OPT_FORMAT,
45 OPT_LOG_LEVEL,
46 OPT_PRETTY,
47 OPT_QUIET,
48)
49from bijux_cli.cli.core.help_text import (
50 HELP_FORMAT,
51 HELP_LOG_LEVEL,
52 HELP_NO_PRETTY,
53 HELP_QUIET,
54)
55from bijux_cli.core.di import DIContainer
56from bijux_cli.core.enums import ErrorType, LogLevel, OutputFormat
57from bijux_cli.core.precedence import current_execution_policy
58from bijux_cli.services.config.contracts import ConfigProtocol
61@dataclass(frozen=True)
62class ConfigSetIntent:
63 """Parsed intent for a config set operation."""
65 key: str
66 value: str
69def _parse_pair(
70 pair: str | None,
71 *,
72 command: str,
73 fmt: OutputFormat,
74 quiet: bool,
75 include_runtime: bool,
76 log_level: LogLevel,
77) -> ConfigSetIntent:
78 """Parse and validate a KEY=VALUE pair for config set."""
79 if pair is None:
80 if sys.stdin.isatty():
81 raise_exit_intent(
82 "Missing argument: KEY=VALUE required",
83 code=2,
84 failure="missing_argument",
85 command=command,
86 fmt=fmt,
87 quiet=quiet,
88 include_runtime=include_runtime,
89 log_level=log_level,
90 )
91 pair = sys.stdin.read().rstrip("\n")
92 if not pair or "=" not in pair:
93 raise_exit_intent(
94 "Invalid argument: KEY=VALUE required",
95 code=2,
96 failure="invalid_argument",
97 command=command,
98 fmt=fmt,
99 quiet=quiet,
100 include_runtime=include_runtime,
101 log_level=log_level,
102 error_type=ErrorType.USER_INPUT,
103 )
104 raw_key, raw_value = pair.split("=", 1)
105 key = raw_key.strip()
106 service_value_str = raw_value
107 if len(service_value_str) >= 2 and (
108 (service_value_str[0] == service_value_str[-1] == '"')
109 or (service_value_str[0] == service_value_str[-1] == "'")
110 ):
111 import codecs
113 service_value_str = codecs.decode(service_value_str[1:-1], "unicode_escape")
114 if not key:
115 raise_exit_intent(
116 "Key cannot be empty",
117 code=2,
118 failure="empty_key",
119 command=command,
120 fmt=fmt,
121 quiet=quiet,
122 include_runtime=include_runtime,
123 log_level=log_level,
124 error_type=ErrorType.USER_INPUT,
125 )
126 if not all(ord(c) < 128 for c in key + service_value_str):
127 raise_exit_intent(
128 "Non-ASCII characters are not allowed in keys or values.",
129 code=3,
130 failure="ascii_error",
131 command=command,
132 fmt=fmt,
133 quiet=quiet,
134 include_runtime=include_runtime,
135 log_level=log_level,
136 extra={"key": key},
137 )
138 if not re.match(r"^[A-Za-z0-9_]+$", key):
139 raise_exit_intent(
140 "Invalid key: only alphanumerics and underscore allowed.",
141 code=2,
142 failure="invalid_key",
143 command=command,
144 fmt=fmt,
145 quiet=quiet,
146 include_runtime=include_runtime,
147 log_level=log_level,
148 extra={"key": key},
149 error_type=ErrorType.USER_INPUT,
150 )
151 if not all(
152 c in string.printable and c not in "\r\n\t\x0b\x0c" for c in service_value_str
153 ):
154 raise_exit_intent(
155 "Control characters are not allowed in config values.",
156 code=3,
157 failure="control_char_error",
158 command=command,
159 fmt=fmt,
160 quiet=quiet,
161 include_runtime=include_runtime,
162 log_level=log_level,
163 extra={"key": key},
164 )
165 return ConfigSetIntent(key=key, value=service_value_str)
168def set_config(
169 ctx: typer.Context,
170 pair: str | None = typer.Argument(
171 None, help="KEY=VALUE to set; if omitted, read from stdin"
172 ),
173 quiet: bool = typer.Option(False, *OPT_QUIET, help=HELP_QUIET),
174 fmt: str = typer.Option("json", *OPT_FORMAT, help=HELP_FORMAT),
175 pretty: bool = typer.Option(True, OPT_PRETTY, help=HELP_NO_PRETTY),
176 log_level: str = typer.Option("info", *OPT_LOG_LEVEL, help=HELP_LOG_LEVEL),
177) -> None:
178 """Sets or updates a configuration key-value pair.
180 This function orchestrates the `set` operation. It accepts a `KEY=VALUE`
181 pair from either a command-line argument or standard input. It performs
182 extensive validation on the key and value for format and content, handles
183 file locking to prevent race conditions, and emits a structured payload
184 confirming the update.
186 Args:
187 ctx (typer.Context): The Typer context for the CLI.
188 pair (str | None): A string in "KEY=VALUE" format. If None, the pair
189 is read from stdin.
190 quiet (bool): If True, suppresses all output except for errors.
191 fmt (str): The output format, "json" or "yaml".
192 pretty (bool): If True, pretty-prints the output.
193 log_level (str): Logging level for diagnostics.
195 Returns:
196 None:
198 Raises:
199 SystemExit: Always exits with a contract-compliant status code and
200 payload, indicating success or detailing the error.
201 """
202 command = "config set"
203 policy = current_execution_policy()
204 quiet = policy.quiet
205 include_runtime = policy.include_runtime
206 pretty = policy.pretty
207 log_level_value = policy.log_level
208 fmt_lower = validate_common_flags(
209 fmt,
210 command,
211 quiet,
212 include_runtime=include_runtime,
213 log_level=log_level_value,
214 )
215 cfg_path = os.environ.get(ENV_CONFIG, "") or ""
216 if cfg_path:
217 try:
218 cfg_path.encode("ascii")
219 except UnicodeEncodeError:
220 raise_exit_intent(
221 "Non-ASCII characters in config path",
222 code=3,
223 failure="ascii",
224 command="config set",
225 fmt=OutputFormat.JSON,
226 quiet=False,
227 include_runtime=False,
228 extra={"path": "[non-ascii path provided]"},
229 log_level=log_level_value,
230 error_type=ErrorType.ASCII,
231 )
232 if cfg_path:
233 try:
234 with open(cfg_path, "a+") as fh:
235 try:
236 fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
237 except OSError:
238 raise_exit_intent(
239 "Config file is locked",
240 code=1,
241 failure="file_locked",
242 command=command,
243 fmt=fmt_lower,
244 quiet=quiet,
245 include_runtime=include_runtime,
246 log_level=log_level_value,
247 extra={"path": cfg_path},
248 )
249 finally:
250 with suppress(Exception):
251 fcntl.flock(fh, fcntl.LOCK_UN)
252 except OSError:
253 pass
254 intent = _parse_pair(
255 pair,
256 command=command,
257 fmt=fmt_lower,
258 quiet=quiet,
259 include_runtime=include_runtime,
260 log_level=log_level_value,
261 )
262 config_svc = DIContainer.current().resolve(ConfigProtocol)
263 try:
264 config_svc.set(intent.key, intent.value)
265 except Exception as exc:
266 raise_exit_intent(
267 f"Failed to set config: {exc}",
268 code=1,
269 failure="set_failed",
270 command=command,
271 fmt=fmt_lower,
272 quiet=quiet,
273 include_runtime=include_runtime,
274 log_level=log_level_value,
275 )
277 def payload_builder(include_runtime: bool) -> dict[str, object]:
278 """Builds the payload confirming a key was set or updated."""
279 payload: dict[str, object] = {
280 "status": "updated",
281 "key": intent.key,
282 "value": intent.value,
283 }
284 if include_runtime:
285 return {
286 "status": payload["status"],
287 "key": intent.key,
288 "value": intent.value,
289 "python": ascii_safe(platform.python_version(), "python_version"),
290 "platform": ascii_safe(platform.platform(), "platform"),
291 }
292 return payload
294 new_run_command(
295 command_name=command,
296 payload_builder=payload_builder,
297 quiet=quiet,
298 fmt=fmt_lower,
299 pretty=pretty,
300 log_level=log_level_value,
301 )