Coverage for src/rhiza_hooks/check_template_bundles.py: 100%

1#!/usr/bin/env python3 

2"""Validate template-bundles.yml structure and consistency. 

3 

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 

10 

11The script reads .rhiza/template.yml to find the template repository, 

12then fetches template-bundles.yml from that remote repository. 

13 

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. 

19 

20Exit codes: 

21 0 - Validation passed 

22 1 - Validation failed 

23""" 

24 

25from __future__ import annotations 

26 

27import argparse 

28import io 

29import sys 

30from pathlib import Path 

31from typing import Any 

32 

33from rhiza_hooks import _bundles_validate, _repo 

34from rhiza_hooks._bundles_config import _get_config_data 

35from rhiza_hooks._bundles_fetch import ( 

36 _FETCH_ATTEMPTS, 

37 _FETCH_TIMEOUT_SECONDS, 

38 _fetch_remote_bundles, 

39) 

40 

41 

42def _get_config_path(args: argparse.Namespace) -> Path: 

43 """Get the configuration file path from arguments or default location.""" 

44 if args.filenames: 

45 return Path(args.filenames[0]) 

46 return _repo.find_repo_root() / ".rhiza" / "template.yml" 

47 

48 

49def _load_and_validate_config(config_path: Path) -> tuple[dict[str, Any], set[str]] | None: 

50 """Load and validate configuration file. 

51 

52 Returns: 

53 (config, templates_set) if validation succeeds, otherwise ``None`` 

54 """ 

55 config = _get_config_data(config_path) 

56 if config is None: 

57 print(f"Could not load configuration from {config_path}, skipping validation") 

58 return None 

59 

60 templates_to_check = config.get("templates") 

61 if templates_to_check is None or not isinstance(templates_to_check, list): 

62 print(f"No templates field in {config_path}, skipping bundle validation") 

63 return None 

64 

65 templates_set: set[str] = {str(t) for t in templates_to_check} 

66 return config, templates_set 

67 

68 

69def _report_errors(header: str, errors: list[str]) -> None: 

70 """Print a failure ``header`` followed by each error as a bullet.""" 

71 print(header) 

72 for error in errors: 

73 print(f" - {error}") 

74 

75 

76def _validate_remote_bundles( 

77 template_repo: str, 

78 template_branch: str, 

79 templates_set: set[str], 

80 attempts: int = _FETCH_ATTEMPTS, 

81 timeout: float = _FETCH_TIMEOUT_SECONDS, 

82) -> tuple[dict[Any, Any] | None, list[str]]: 

83 """Fetch and validate remote bundles. 

84 

85 Returns: 

86 Tuple of (bundles_data, errors) or (None, errors) if fetch fails 

87 """ 

88 print(f"Fetching template bundles from {template_repo} (branch: {template_branch})") 

89 print(f"Checking templates: {', '.join(sorted(templates_set))}") 

90 

91 fetched = _fetch_remote_bundles(template_repo, template_branch, attempts=attempts, timeout=timeout) 

92 if fetched.data is None: 

93 _report_errors("\n✗ Failed to fetch template bundles:", fetched.errors) 

94 return None, fetched.errors 

95 

96 # data is narrowed to dict[Any, Any] by the `is None` guard above. 

97 data = fetched.data 

98 

99 # Validate top-level structure 

100 errors = _bundles_validate._validate_top_level_fields(data) 

101 if errors: 

102 _report_errors("\n✗ Template bundles validation failed:", errors) 

103 return None, errors 

104 

105 bundles = data.get("bundles", {}) 

106 if not isinstance(bundles, dict): 

107 errors = ["'bundles' must be a dictionary"] 

108 _report_errors("\n✗ Template bundles validation failed:", errors) 

109 return None, errors 

110 

111 return data, [] 

112 

113 

114def _validate_templates_in_bundles(templates_set: set[str], bundles: dict[Any, Any], config_path: Path) -> list[str]: 

115 """Validate that requested templates exist in the remote bundles and are well-formed.""" 

116 return _bundles_validate._validate_selected_bundles( 

117 templates_set, 

118 bundles, 

119 lambda t: f"Template '{t}' specified in {config_path} not found in remote bundles", 

120 ) 

121 

122 

123def _parse_args(argv: list[str] | None) -> argparse.Namespace: 

124 """Build the argument parser, then parse and validate ``argv``.""" 

125 parser = argparse.ArgumentParser(description="Validate template-bundles.yml from remote template repository") 

126 parser.add_argument( 

127 "filenames", 

128 nargs="*", 

129 help="Filenames to check (should be .rhiza/template.yml)", 

130 ) 

131 parser.add_argument( 

132 "--offline", 

133 action="store_true", 

134 help="Skip the remote bundles fetch (e.g. for offline commits) and pass", 

135 ) 

136 parser.add_argument( 

137 "--retries", 

138 type=int, 

139 default=_FETCH_ATTEMPTS - 1, 

140 help="Number of retries for transient network failures, after the initial attempt (default: %(default)s)", 

141 ) 

142 parser.add_argument( 

143 "--timeout", 

144 type=float, 

145 default=_FETCH_TIMEOUT_SECONDS, 

146 help="Per-request network timeout in seconds (default: %(default)s)", 

147 ) 

148 args = parser.parse_args(argv) 

149 

150 if args.retries < 0: 

151 parser.error("--retries must be non-negative") 

152 if args.timeout <= 0: 

153 parser.error("--timeout must be positive") 

154 

155 return args 

156 

157 

158def _ensure_utf8_stdout() -> None: 

159 """Reconfigure stdout to UTF-8 so the ✓/✗ status glyphs never crash a non-UTF-8 console.""" 

160 if isinstance(sys.stdout, io.TextIOWrapper): 

161 sys.stdout.reconfigure(encoding="utf-8", errors="replace") 

162 

163 

164def _run_remote_validation( 

165 config: dict[str, Any], 

166 templates_set: set[str], 

167 config_path: Path, 

168 retries: int, 

169 timeout: float, 

170) -> int: 

171 """Fetch remote bundles and validate the requested templates; return an exit code.""" 

172 template_repo = config.get("template-repository") 

173 template_branch = config.get("template-branch") 

174 if not template_repo or not template_branch: 

175 print(f"Missing template-repository or template-branch in {config_path}") 

176 return 1 

177 

178 data, _fetch_errors = _validate_remote_bundles( 

179 template_repo, 

180 template_branch, 

181 templates_set, 

182 attempts=retries + 1, 

183 timeout=timeout, 

184 ) 

185 if data is None: 

186 return 1 

187 

188 bundles = data.get("bundles", {}) 

189 errors = _validate_templates_in_bundles(templates_set, bundles, config_path) 

190 if errors: 

191 _report_errors("\n✗ Template bundles validation failed:", errors) 

192 return 1 

193 

194 print("✓ Template bundles validation passed!") 

195 return 0 

196 

197 

198def main(argv: list[str] | None = None) -> int: 

199 """Main entry point.""" 

200 _ensure_utf8_stdout() 

201 

202 args = _parse_args(argv) 

203 

204 if args.offline: 

205 print("Offline mode: skipping remote template bundles validation") 

206 return 0 

207 

208 config_path = _get_config_path(args) 

209 

210 # Load and validate configuration. A None result means validation was 

211 # skipped (missing config or no templates field); both cases pass. 

212 result = _load_and_validate_config(config_path) 

213 if result is None: 

214 return 0 

215 config, templates_set = result 

216 

217 return _run_remote_validation(config, templates_set, config_path, args.retries, args.timeout) 

218 

219 

220if __name__ == "__main__": # pragma: no mutate 

221 sys.exit(main())