Coverage for  / home / runner / work / bijux-cli / bijux-cli / src / bijux_cli / cli / root.py: 97%

52 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"""Build the Bijux CLI root Typer application and register plugins. 

5 

6This module assembles the root Typer app, registers core commands, and discovers 

7external plugins via entry points: 

8 

9* ``bijux.commands``: each entry must be a ``Typer`` app mounted under its 

10 entry-point name. 

11* ``bijux_cli.plugins``: flexible plugins that may: 

12 - return a ``Typer`` app, 

13 - be a callable factory/class (instantiated with no arguments), 

14 - expose ``registered_groups: dict[str, Typer]``, 

15 - expose ``register(app: Typer)`` to register commands/groups. 

16""" 

17 

18from __future__ import annotations 

19 

20from collections.abc import Iterable, Mapping 

21import logging 

22import subprocess # noqa: S603 # nosec B404 - controlled internal call 

23import sys 

24from typing import Any 

25 

26import typer 

27from typer import Context 

28 

29from bijux_cli.cli.color import resolve_click_color 

30from bijux_cli.cli.commands import register_commands, register_dynamic_plugins 

31from bijux_cli.core.intent import parse_global_config 

32from bijux_cli.core.precedence import current_execution_policy 

33from bijux_cli.core.runtime import AsyncTyper 

34 

35logger = logging.getLogger(__name__) 

36if not logger.handlers: 36 ↛ 40line 36 didn't jump to line 40 because the condition on line 36 was always true

37 logger.addHandler(logging.NullHandler()) 

38 

39 

40def _collect_names(container: Mapping[Any, Any] | Iterable[Any]) -> list[str]: 

41 """Collect command/group names from a Typer registry-like container. 

42 

43 Args: 

44 container: A list-like or dict-like container holding Typer objects. 

45 

46 Returns: 

47 A list of registered names. 

48 """ 

49 items: Iterable[Any] = ( 

50 container.values() if isinstance(container, Mapping) else container 

51 ) 

52 

53 names: list[str] = [] 

54 for obj in items: 

55 name = getattr(obj, "name", None) 

56 if isinstance(name, str) and name: 

57 names.append(name) 

58 return names 

59 

60 

61def _existing_top_level_names(app: typer.Typer) -> set[str]: 

62 """Return the set of names already registered at the top level. 

63 

64 Args: 

65 app: The root Typer application. 

66 

67 Returns: 

68 A set of names for existing groups and commands. 

69 """ 

70 groups = _collect_names(getattr(app, "registered_groups", []) or []) 

71 commands = _collect_names(getattr(app, "registered_commands", []) or []) 

72 return set(groups) | set(commands) 

73 

74 

75def register_entrypoint_plugins(app: AsyncTyper) -> None: 

76 """Discover and register plugins exposed via entry points. 

77 

78 Args: 

79 app: The root Typer application. 

80 """ 

81 register_dynamic_plugins(app) 

82 

83 

84def maybe_default_to_repl(ctx: Context) -> None: 

85 """Launch the REPL when invoked with no args; otherwise show help on error. 

86 

87 If no subcommand is chosen and no extra CLI arguments are provided, the 

88 function re-invokes the executable with the ``repl`` command. If arguments 

89 are present but no subcommand is resolved, it prints help and exits with 

90 code 2. 

91 

92 Args: 

93 ctx: The Typer context. 

94 """ 

95 if ctx.invoked_subcommand is None and len(sys.argv) == 1: 

96 subprocess.call( # noqa: S603 # nosec B603 - controlled argv 

97 [sys.argv[0], "repl"] 

98 ) 

99 elif ctx.invoked_subcommand is None: 

100 policy = current_execution_policy() 

101 typer.echo( 

102 ctx.get_help(), 

103 color=resolve_click_color(quiet=policy.quiet, fmt=None), 

104 ) 

105 raise typer.Exit(code=2) 

106 

107 

108def _log_registered(app: typer.Typer) -> None: 

109 """Log the names of registered core commands and groups at debug level. 

110 

111 Args: 

112 app: The root Typer application. 

113 """ 

114 cmds = _collect_names(getattr(app, "registered_commands", []) or []) 

115 grps = _collect_names(getattr(app, "registered_groups", []) or []) 

116 logger.debug("Core commands registered: %s", cmds) 

117 logger.debug("Core groups registered: %s", grps) 

118 

119 

120def build_app(*, load_plugins: bool = True) -> AsyncTyper: 

121 """Construct the root Typer application. 

122 

123 Returns: 

124 The fully assembled Typer application with core and plugin commands. 

125 """ 

126 app = AsyncTyper( 

127 help="Bijux CLI – Lean, plug-in-driven command-line interface.", 

128 invoke_without_command=True, 

129 ) 

130 register_commands(app) 

131 _log_registered(app) 

132 if load_plugins: 

133 register_dynamic_plugins(app) 

134 app.callback(invoke_without_command=True)(maybe_default_to_repl) 

135 return app 

136 

137 

138app = build_app() 

139 

140 

141__all__ = [ 

142 "app", 

143 "build_app", 

144 "parse_global_config", 

145 "register_entrypoint_plugins", 

146]