Coverage for /builds/svesch/cikit/src/cikit/ci/loader.py: 93%
72 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-10-09 14:17 +0000
1# SPDX-License-Identifier: GPL-3.0-or-later
2"""Command discovery for cikit CI commands.
4Author:
5 Sven Schlender
6Date:
7 2026-06-16
8"""
10from contextlib import contextmanager
11from pathlib import Path
12import sys
13import argparse
14import importlib
15import pkgutil
16from types import ModuleType
17from importlib.metadata import entry_points
19from cikit.errors import CiToolsError
22LOCAL_COMMAND_PACKAGE = "ci_commands"
23COMMAND_ENTRY_POINT_GROUP = "cikit.commands"
26def _iter_package_modules(package: ModuleType) -> list[ModuleType]:
27 """Import all public command modules from a package.
29 Args:
30 package: Python package containing command modules.
32 Returns:
33 Imported command modules.
35 Raises:
36 CiToolsError: If the supplied module is not a package.
37 """
38 package_path = getattr(package, "__path__", None)
39 if package_path is None:
40 raise CiToolsError(f"Command provider '{package.__name__}' is not a package.")
42 modules = []
44 for module_info in pkgutil.iter_modules(package_path):
45 if not module_info.name.startswith("_"): 45 ↛ 44line 45 didn't jump to line 44 because the condition on line 45 was always true
46 modules.append(importlib.import_module(f"{package.__name__}.{module_info.name}"))
48 return modules
51def _get_command_name(module: ModuleType) -> str:
52 """Return the command name declared by a module.
54 Args:
55 module: Command module to inspect.
57 Returns:
58 Validated command name.
60 Raises:
61 CiToolsError: If COMMAND_NAME is missing or invalid.
62 """
63 command_name = getattr(module, "COMMAND_NAME", None)
65 if not isinstance(command_name, str) or not command_name.strip():
66 raise CiToolsError(
67 f"Command module '{module.__name__}' must define a non-empty COMMAND_NAME."
68 )
70 return command_name
73def _iter_provider_modules() -> list[ModuleType]:
74 """Import command modules from installed command providers.
76 Returns:
77 Command modules exposed through the cikit command entry point group.
78 """
79 modules = []
81 for entry_point in entry_points(group=COMMAND_ENTRY_POINT_GROUP):
82 package = entry_point.load()
83 modules.extend(_iter_package_modules(package))
85 return modules
88def _iter_local_modules() -> list[ModuleType]:
89 """Import project-local command modules.
91 Returns:
92 Local command modules or an empty list if no command package exists.
93 """
94 package_dir = Path.cwd() / LOCAL_COMMAND_PACKAGE
96 if not package_dir.is_dir(): 96 ↛ 97line 96 didn't jump to line 97 because the condition on line 96 was never true
97 return []
99 with _project_root_import_path():
100 package = importlib.import_module(LOCAL_COMMAND_PACKAGE)
101 return _iter_package_modules(package)
104@contextmanager
105def _project_root_import_path():
106 """Temporarily add the current working directory to sys.path.
108 Yields:
109 Nothing.
110 """
111 project_root = str(Path.cwd())
112 was_present = project_root in sys.path
114 if not was_present: 114 ↛ 115line 114 didn't jump to line 115 because the condition on line 114 was never true
115 sys.path.insert(0, project_root)
117 try:
118 yield
119 finally:
120 if not was_present: 120 ↛ 121line 120 didn't jump to line 121 because the condition on line 120 was never true
121 sys.path.remove(project_root)
124def _add_provider_commands(commands: dict, modules: list[ModuleType]):
125 """Add installed provider commands to the command registry.
127 Args:
128 commands: Effective command registry.
129 modules: Provider command modules to add.
131 Raises:
132 CiToolsError: If multiple providers define the same command.
133 """
134 for module in modules:
135 command_name = _get_command_name(module)
137 if command_name in commands:
138 other = commands[command_name]
139 raise CiToolsError(
140 f"Command '{command_name}' is provided by both "
141 f"'{other.__name__}' and '{module.__name__}'."
142 )
144 commands[command_name] = module
147def _add_local_commands(commands: dict, modules: list[ModuleType]):
148 """Add project-local commands to the command registry.
150 Local commands replace installed provider commands with the same command
151 name. Duplicate command names within the local package are rejected.
153 Args:
154 commands: Effective command registry.
155 modules: Local command modules to add.
157 Raises:
158 CiToolsError: If multiple local modules define the same command.
159 """
160 local_names = set()
162 for module in modules:
163 command_name = _get_command_name(module)
165 if command_name in local_names:
166 raise CiToolsError(
167 f"Command '{command_name}' is provided by multiple local command modules."
168 )
170 local_names.add(command_name)
171 commands[command_name] = module
174def iter_command_modules() -> list[ModuleType]:
175 """Discover installed and project-local command modules.
177 Project-local commands take precedence over installed provider commands
178 with the same command name.
180 Returns:
181 Effective command modules sorted by command name.
183 Raises:
184 CiToolsError: If command metadata or command names are invalid.
185 """
186 commands = {}
188 _add_provider_commands(commands, _iter_provider_modules())
189 _add_local_commands(commands, _iter_local_modules())
191 return [commands[name] for name in sorted(commands)]
194def register_commands(subparsers: argparse._SubParsersAction):
195 """Register all detected command modules.
197 Args:
198 subparsers: Argparse subparser registry.
200 Raises:
201 AttributeError: If a command module does not provide ``register``.
202 CiToolsError: If command discovery or validation fails.
203 """
204 for module in iter_command_modules():
205 register = getattr(module, "register")
206 register(subparsers)