Coverage for src/rhiza_hooks/check_template_bundles.py: 100%
94 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-15 06:14 +0000
1#!/usr/bin/env python3
2"""Validate template-bundles.yml structure and consistency.
4This script validates the template bundles configuration file to ensure:
51. Valid YAML syntax
62. Required fields are present
73. Bundle dependencies reference existing bundles
84. File paths follow expected patterns
95. Examples reference valid bundles
11The script reads .rhiza/template.yml to find the template repository,
12then fetches template-bundles.yml from that remote repository.
14The implementation is split across focused modules — :mod:`rhiza_hooks._bundles_fetch`
15(obtaining a document), :mod:`rhiza_hooks._bundles_validate` (structural checks),
16and :mod:`rhiza_hooks._bundles_config` (reading ``.rhiza/template.yml``). This
17module is the CLI/orchestration layer and re-exports those helpers so
18``rhiza_hooks.check_template_bundles`` remains the single public import surface.
20Those three modules are themselves private, so the package's public surface is
21unchanged by the split. Within them a leading underscore marks a helper with no
22caller outside its own file — which is why this module imports only unprefixed
23names from them.
25Exit codes:
26 0 - Validation passed
27 1 - Validation failed
28"""
30from __future__ import annotations
32import argparse
33import io
34import sys
35from pathlib import Path
36from typing import Any
38from rhiza_hooks import _bundles_validate, _repo
39from rhiza_hooks._bundles_config import get_config_data
40from rhiza_hooks._bundles_fetch import (
41 FETCH_ATTEMPTS,
42 FETCH_TIMEOUT_SECONDS,
43 Fetcher,
44 fetch_remote_bundles,
45)
48def _get_config_path(args: argparse.Namespace) -> Path:
49 """Get the configuration file path from arguments or default location."""
50 if args.filenames:
51 return Path(args.filenames[0])
52 return _repo.find_repo_root() / ".rhiza" / "template.yml"
55def _load_and_validate_config(config_path: Path) -> tuple[dict[str, Any], set[str]] | None:
56 """Load and validate configuration file.
58 Returns:
59 (config, templates_set) if validation succeeds, otherwise ``None``
60 """
61 config = get_config_data(config_path)
62 if config is None:
63 print(f"Could not load configuration from {config_path}, skipping validation")
64 return None
66 templates_to_check = config.get("templates")
67 if templates_to_check is None or not isinstance(templates_to_check, list):
68 print(f"No templates field in {config_path}, skipping bundle validation")
69 return None
71 templates_set: set[str] = {str(t) for t in templates_to_check}
72 return config, templates_set
75def _report_errors(header: str, errors: list[str]) -> None:
76 """Print a failure ``header`` followed by each error as a bullet, on stderr."""
77 print(header, file=sys.stderr)
78 for error in errors:
79 print(f" - {error}", file=sys.stderr)
82def _validate_remote_bundles(
83 template_repo: str,
84 template_branch: str,
85 templates_set: set[str],
86 *,
87 fetcher: Fetcher,
88 attempts: int = FETCH_ATTEMPTS,
89 timeout: float = FETCH_TIMEOUT_SECONDS,
90) -> tuple[dict[Any, Any] | None, list[str]]:
91 """Fetch and validate remote bundles.
93 ``fetcher`` is required rather than defaulted: this is the function that would
94 otherwise reach the network, so making the dependency explicit is what keeps a
95 test from silently doing so. The default lives once, on :func:`main`.
97 Returns:
98 Tuple of (bundles_data, errors) or (None, errors) if fetch fails
99 """
100 print(f"Fetching template bundles from {template_repo} (branch: {template_branch})")
101 print(f"Checking templates: {', '.join(sorted(templates_set))}")
103 fetched = fetcher(template_repo, template_branch, attempts=attempts, timeout=timeout)
104 if fetched.data is None:
105 _report_errors("\n✗ Failed to fetch template bundles:", fetched.errors)
106 return None, fetched.errors
108 # data is narrowed to dict[Any, Any] by the `is None` guard above.
109 data = fetched.data
111 # Validate top-level structure
112 errors = _bundles_validate.validate_top_level_fields(data)
113 if errors:
114 _report_errors("\n✗ Template bundles validation failed:", errors)
115 return None, errors
117 bundles = data.get("bundles", {})
118 if not isinstance(bundles, dict):
119 errors = ["'bundles' must be a dictionary"]
120 _report_errors("\n✗ Template bundles validation failed:", errors)
121 return None, errors
123 return data, []
126def _validate_templates_in_bundles(templates_set: set[str], bundles: dict[Any, Any], config_path: Path) -> list[str]:
127 """Validate that requested templates exist in the remote bundles and are well-formed.
129 >>> bundles = {"core": {"description": "Core files", "files": [".gitignore"]}}
130 >>> _validate_templates_in_bundles({"core"}, bundles, Path("template.yml"))
131 []
133 A template this repo asks for but the template repository does not publish is
134 the error this hook exists to catch:
136 >>> _validate_templates_in_bundles({"nope"}, bundles, Path("template.yml"))
137 ["Template 'nope' specified in template.yml not found in remote bundles"]
139 A published bundle missing its required fields is reported too:
141 >>> _validate_templates_in_bundles({"core"}, {"core": {}}, Path("template.yml"))
142 ["Bundle 'core' missing 'description'", "Bundle 'core' missing 'files'"]
143 """
144 return _bundles_validate.validate_selected_bundles(
145 templates_set,
146 bundles,
147 lambda t: f"Template '{t}' specified in {config_path} not found in remote bundles",
148 )
151def _parse_args(argv: list[str] | None) -> argparse.Namespace:
152 """Build the argument parser, then parse and validate ``argv``."""
153 parser = argparse.ArgumentParser(description="Validate template-bundles.yml from remote template repository")
154 parser.add_argument(
155 "filenames",
156 nargs="*",
157 help="Filenames to check (should be .rhiza/template.yml)",
158 )
159 parser.add_argument(
160 "--offline",
161 action="store_true",
162 help="Skip the remote bundles fetch (e.g. for offline commits) and pass",
163 )
164 parser.add_argument(
165 "--retries",
166 type=int,
167 default=FETCH_ATTEMPTS - 1,
168 help="Number of retries for transient network failures, after the initial attempt (default: %(default)s)",
169 )
170 parser.add_argument(
171 "--timeout",
172 type=float,
173 default=FETCH_TIMEOUT_SECONDS,
174 help="Per-request network timeout in seconds (default: %(default)s)",
175 )
176 args = parser.parse_args(argv)
178 if args.retries < 0:
179 parser.error("--retries must be non-negative")
180 if args.timeout <= 0:
181 parser.error("--timeout must be positive")
183 return args
186def _ensure_utf8_output() -> None:
187 """Reconfigure both output streams to UTF-8 so the ✓/✗ glyphs never crash a non-UTF-8 console.
189 Both, not just stdout: the ``✓`` goes to stdout on the success path but the ``✗``
190 headers from :func:`_report_errors` go to stderr, so guarding one stream would
191 leave the failure path — the one a user is most likely to be reading — able to
192 raise ``UnicodeEncodeError`` on a cp1252 console.
193 """
194 for stream in (sys.stdout, sys.stderr):
195 if isinstance(stream, io.TextIOWrapper):
196 stream.reconfigure(encoding="utf-8", errors="replace")
199def _run_remote_validation(
200 config: dict[str, Any],
201 templates_set: set[str],
202 config_path: Path,
203 retries: int,
204 timeout: float,
205 fetcher: Fetcher,
206) -> int:
207 """Fetch remote bundles and validate the requested templates; return an exit code."""
208 template_repo = config.get("template-repository")
209 template_branch = config.get("template-branch")
210 if not template_repo or not template_branch:
211 print(f"Missing template-repository or template-branch in {config_path}", file=sys.stderr)
212 return 1
214 data, _fetch_errors = _validate_remote_bundles(
215 template_repo,
216 template_branch,
217 templates_set,
218 fetcher=fetcher,
219 attempts=retries + 1,
220 timeout=timeout,
221 )
222 if data is None:
223 return 1
225 bundles = data.get("bundles", {})
226 errors = _validate_templates_in_bundles(templates_set, bundles, config_path)
227 if errors:
228 _report_errors("\n✗ Template bundles validation failed:", errors)
229 return 1
231 print("✓ Template bundles validation passed!")
232 return 0
235def main(argv: list[str] | None = None, fetcher: Fetcher = fetch_remote_bundles) -> int:
236 """Main entry point.
238 ``fetcher`` is the one place the real network call is named. The console script
239 takes the default; a test passes a fake document source as an argument rather than
240 rebinding this module's ``fetch_remote_bundles`` global.
241 """
242 _ensure_utf8_output()
244 args = _parse_args(argv)
246 if args.offline:
247 print("Offline mode: skipping remote template bundles validation")
248 return 0
250 config_path = _get_config_path(args)
252 # Load and validate configuration. A None result means validation was
253 # skipped (missing config or no templates field); both cases pass.
254 result = _load_and_validate_config(config_path)
255 if result is None:
256 return 0
257 config, templates_set = result
259 return _run_remote_validation(config, templates_set, config_path, args.retries, args.timeout, fetcher)
262if __name__ == "__main__": # pragma: no mutate
263 sys.exit(main())