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

41 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 

3 

4"""Implements the `sleep` command for the Bijux CLI. 

5 

6This module provides a simple command to pause execution for a specified duration. 

7It is primarily used for scripting, testing, or rate-limiting operations within 

8automated workflows. The command returns a structured payload confirming the 

9duration slept. 

10 

11Output Contract: 

12 * Success: `{"slept": float}` 

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

14 

15Exit Codes: 

16 * `0`: Success. 

17 * `1`: Internal or configuration-related error. 

18 * `2`: Invalid argument (e.g., negative duration) or timeout exceeded. 

19""" 

20 

21from __future__ import annotations 

22 

23import platform 

24import time 

25 

26import typer 

27 

28from bijux_cli.cli.core.command import ( 

29 ascii_safe, 

30 new_run_command, 

31 validate_common_flags, 

32) 

33from bijux_cli.cli.core.constants import ( 

34 DEFAULT_COMMAND_TIMEOUT, 

35 ENV_COMMAND_TIMEOUT, 

36 OPT_FORMAT, 

37 OPT_LOG_LEVEL, 

38 OPT_PRETTY, 

39 OPT_QUIET, 

40) 

41from bijux_cli.cli.core.help_text import ( 

42 HELP_FORMAT, 

43 HELP_LOG_LEVEL, 

44 HELP_NO_PRETTY, 

45 HELP_QUIET, 

46) 

47from bijux_cli.core.di import DIContainer 

48from bijux_cli.core.enums import ErrorType 

49from bijux_cli.core.exit_policy import ExitIntentError 

50from bijux_cli.core.precedence import current_execution_policy, resolve_exit_intent 

51from bijux_cli.core.runtime import AsyncTyper 

52from bijux_cli.services.config.contracts import ConfigProtocol 

53 

54typer.core.rich = None # type: ignore[attr-defined] 

55 

56sleep_app = AsyncTyper( 

57 name="sleep", 

58 help="Pause execution for a specified duration.", 

59 rich_markup_mode=None, 

60 context_settings={"help_option_names": ["-h", "--help"]}, 

61 no_args_is_help=False, 

62) 

63 

64 

65def _build_payload(include_runtime: bool, slept: float) -> dict[str, object]: 

66 """Constructs the structured payload for the sleep command. 

67 

68 Args: 

69 include_runtime (bool): If True, includes Python version and platform 

70 information in the payload. 

71 slept (float): The number of seconds the command slept. 

72 

73 Returns: 

74 Mapping[str, object]: A dictionary containing the sleep duration and 

75 optional runtime details. 

76 """ 

77 payload: dict[str, object] = {"slept": slept} 

78 if include_runtime: 

79 payload.update( 

80 { 

81 "python": ascii_safe(platform.python_version(), "python_version"), 

82 "platform": ascii_safe(platform.platform(), "platform"), 

83 } 

84 ) 

85 return payload 

86 

87 

88@sleep_app.callback(invoke_without_command=True) 

89def sleep( 

90 ctx: typer.Context, 

91 seconds: float = typer.Option( 

92 ..., "--seconds", "-s", help="Duration in seconds (must be ≥ 0)" 

93 ), 

94 quiet: bool = typer.Option(False, *OPT_QUIET, help=HELP_QUIET), 

95 fmt: str = typer.Option("json", *OPT_FORMAT, help=HELP_FORMAT), 

96 pretty: bool = typer.Option(True, OPT_PRETTY, help=HELP_NO_PRETTY), 

97 log_level: str = typer.Option("info", *OPT_LOG_LEVEL, help=HELP_LOG_LEVEL), 

98) -> None: 

99 """Defines the entrypoint and logic for the `bijux sleep` command. 

100 

101 This function validates the requested sleep duration against configuration 

102 limits, pauses execution, and then emits a structured payload confirming 

103 the duration. 

104 

105 Args: 

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

107 seconds (float): The duration in seconds to pause execution. Must be 

108 non-negative and not exceed the configured command timeout. 

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

110 output payload. 

111 fmt (str): The output format, either "json" or "yaml". Defaults to "json". 

112 pretty (bool): If True, pretty-prints the output for human readability. log_level (str): Logging level for diagnostics. 

113 and `pretty`. 

114 

115 Returns: 

116 None: 

117 

118 Raises: 

119 SystemExit: Exits with a contract-compliant status code and payload 

120 upon any error, such as a negative sleep duration or a timeout 

121 violation. 

122 """ 

123 command = "sleep" 

124 

125 effective = current_execution_policy() 

126 fmt_lower = validate_common_flags( 

127 fmt, 

128 command, 

129 effective.quiet, 

130 include_runtime=effective.include_runtime, 

131 log_level=effective.log_level, 

132 ) 

133 quiet = effective.quiet 

134 pretty = effective.pretty 

135 

136 if seconds < 0: 

137 intent = resolve_exit_intent( 

138 message="sleep length must be non-negative", 

139 code=2, 

140 failure="negative", 

141 command=command, 

142 fmt=fmt_lower, 

143 quiet=quiet, 

144 include_runtime=effective.include_runtime, 

145 error_type=ErrorType.USER_INPUT, 

146 log_level=effective.log_level, 

147 ) 

148 raise ExitIntentError(intent) 

149 

150 cfg: ConfigProtocol = DIContainer.current().resolve(ConfigProtocol) 

151 

152 try: 

153 timeout = float(cfg.get(ENV_COMMAND_TIMEOUT, DEFAULT_COMMAND_TIMEOUT)) 

154 except Exception as exc: 

155 intent = resolve_exit_intent( 

156 message=f"Failed to read timeout: {exc}", 

157 code=1, 

158 failure="config", 

159 command=command, 

160 fmt=fmt_lower, 

161 quiet=quiet, 

162 include_runtime=effective.include_runtime, 

163 error_type=ErrorType.INTERNAL, 

164 log_level=effective.log_level, 

165 ) 

166 raise ExitIntentError(intent) from exc 

167 

168 if seconds > timeout: 

169 intent = resolve_exit_intent( 

170 message=( 

171 "Command timed out because sleep duration exceeded the configured timeout." 

172 ), 

173 code=2, 

174 failure="timeout", 

175 command=command, 

176 fmt=fmt_lower, 

177 quiet=quiet, 

178 include_runtime=effective.include_runtime, 

179 error_type=ErrorType.USER_INPUT, 

180 log_level=effective.log_level, 

181 ) 

182 raise ExitIntentError(intent) 

183 

184 time.sleep(seconds) 

185 

186 new_run_command( 

187 command_name=command, 

188 payload_builder=lambda include: _build_payload(include, seconds), 

189 quiet=quiet, 

190 fmt=fmt_lower, 

191 pretty=pretty, 

192 log_level=log_level, 

193 )