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

1#!/usr/bin/env python3 

2"""Load and fetch template-bundles documents into a typed result. 

3 

4This module is responsible solely for *obtaining* a template-bundles document — 

5from a local file, from already-fetched bytes, or from a remote GitHub 

6repository — and returning it as a :class:`BundlesDoc`. Structural validation of 

7the returned mapping lives in :mod:`rhiza_hooks._bundles_validate`. 

8""" 

9 

10from __future__ import annotations 

11 

12import time 

13from dataclasses import dataclass 

14from pathlib import Path 

15from typing import Any 

16from urllib.error import HTTPError, URLError 

17from urllib.parse import urlparse 

18from urllib.request import urlopen 

19 

20import yaml 

21 

22from rhiza_hooks._yaml import YamlError, YamlFailure, load_yaml_mapping 

23 

24# Remote fetch is retried on transient network errors before giving up. Two 

25# attempts = one initial try plus one retry, with a short linear backoff. 

26# These are defaults; the CLI exposes `--retries` and `--timeout` to override. 

27_FETCH_ATTEMPTS = 2 

28_FETCH_BACKOFF_SECONDS = 1.0 

29_FETCH_TIMEOUT_SECONDS = 10.0 

30 

31 

32@dataclass(frozen=True) 

33class BundlesDoc: 

34 """Outcome of loading/parsing a template-bundles document. 

35 

36 ``data`` holds the parsed mapping on success and is ``None`` on failure; 

37 ``errors`` carries the failure messages (empty on success). The two are 

38 mutually exclusive, so callers branch on ``data is None`` — which also lets 

39 the type checker narrow ``data`` to ``dict`` on the success path without a 

40 cast. 

41 """ 

42 

43 data: dict[Any, Any] | None 

44 errors: list[str] 

45 

46 

47def load_local_bundles(bundles_path: Path) -> BundlesDoc: 

48 """Load and parse a local template-bundles file into a :class:`BundlesDoc`. 

49 

50 This is the local-file counterpart to :func:`_fetch_remote_bundles` and part 

51 of this module's cross-module surface — :mod:`rhiza_hooks._bundles_validate` 

52 calls it to load a document before validating it. 

53 """ 

54 result = load_yaml_mapping(bundles_path) 

55 if not isinstance(result, YamlFailure): 

56 return BundlesDoc(result, []) 

57 

58 messages = { 

59 YamlError.NOT_FOUND: f"Template bundles file not found: {bundles_path}", 

60 YamlError.INVALID: f"Invalid YAML: {result.detail}", 

61 YamlError.EMPTY: "Template bundles file is empty", 

62 YamlError.NOT_MAPPING: "Template bundles file must be a dictionary", 

63 } 

64 return BundlesDoc(None, [messages[result.kind]]) 

65 

66 

67def _parse_remote_bundles(content: bytes) -> BundlesDoc: 

68 """Parse fetched template-bundles.yml content into a :class:`BundlesDoc`.""" 

69 try: 

70 data = yaml.safe_load(content) 

71 except yaml.YAMLError as e: 

72 return BundlesDoc(None, [f"Invalid YAML in remote template bundles: {e}"]) 

73 

74 if data is None: 

75 return BundlesDoc(None, ["Remote template bundles file is empty"]) 

76 

77 if not isinstance(data, dict): 

78 return BundlesDoc(None, ["Remote template bundles must be a dictionary"]) 

79 

80 return BundlesDoc(data, []) 

81 

82 

83def _fetch_once(url: str, timeout: float, repo: str, branch: str) -> bytes | BundlesDoc | str: 

84 """Perform a single fetch attempt. 

85 

86 Returns the raw response ``bytes`` on success; a :class:`BundlesDoc` carrying 

87 a permanent HTTP error (e.g. 404) that must not be retried; or an error 

88 ``str`` describing a transient network/timeout failure that may be retried. 

89 """ 

90 try: 

91 with urlopen(url, timeout=timeout) as response: # noqa: S310 # nosec B310 

92 content: bytes = response.read() 

93 except HTTPError as e: 

94 if e.code == 404: 

95 return BundlesDoc(None, [f"Template bundles file not found in repository {repo} (branch: {branch})"]) 

96 return BundlesDoc(None, [f"HTTP error fetching template bundles: {e.code} {e.reason}"]) 

97 except URLError as e: 

98 return f"Error fetching template bundles from {url}: {e.reason}" 

99 except TimeoutError: 

100 return f"Timeout fetching template bundles from {url}" 

101 return content 

102 

103 

104def _log_failed_attempt(attempt: int, attempts: int, error: str, backoff: float) -> None: 

105 """Log a failed fetch attempt, sleeping with linear backoff if retries remain. 

106 

107 Backoff grows linearly (backoff, 2*backoff, ...). The last attempt has 

108 nowhere to retry, so it is logged without a backoff. 

109 """ 

110 if attempt + 1 < attempts: 

111 delay = backoff * (attempt + 1) 

112 print(f" Attempt {attempt + 1}/{attempts} failed: {error}; retrying in {delay:.1f}s") 

113 time.sleep(delay) 

114 else: 

115 print(f" Attempt {attempt + 1}/{attempts} failed: {error}") 

116 

117 

118def _fetch_remote_bundles( 

119 repo: str, 

120 branch: str, 

121 attempts: int = _FETCH_ATTEMPTS, 

122 backoff: float = _FETCH_BACKOFF_SECONDS, 

123 timeout: float = _FETCH_TIMEOUT_SECONDS, 

124) -> BundlesDoc: 

125 """Fetch template-bundles.yml from a remote GitHub repository. 

126 

127 Transient network failures (`URLError`/`TimeoutError`) are retried up to 

128 ``attempts`` times with a linear backoff, and each failed attempt is logged 

129 so CI failures are diagnosable. HTTP errors (e.g. 404) are permanent and 

130 returned immediately without retrying. 

131 

132 Args: 

133 repo: GitHub repository in 'owner/repo' format 

134 branch: Branch name 

135 attempts: Total number of fetch attempts (initial try + retries) 

136 backoff: Base seconds to sleep between attempts (multiplied by attempt number) 

137 timeout: Per-request socket timeout in seconds 

138 

139 Returns: 

140 A :class:`BundlesDoc` with the parsed mapping on success, or errors. 

141 """ 

142 # Construct GitHub raw content URL 

143 url = f"https://raw.githubusercontent.com/{repo}/{branch}/.rhiza/template-bundles.yml" 

144 

145 # Validate URL scheme for security (bandit B310) 

146 parsed = urlparse(url) 

147 if parsed.scheme != "https": 

148 return BundlesDoc(None, [f"Invalid URL scheme: {parsed.scheme}. Only https is allowed."]) 

149 

150 # pragma below: equivalent mutant — the final `return BundlesDoc(None, errors)` is only 

151 # reached after a transient-error iteration has reassigned `errors` (success and HTTP 

152 # errors return early), so for attempts >= 1 this initial value is never the one returned. 

153 errors: list[str] = [] # pragma: no mutate 

154 for attempt in range(attempts): 

155 outcome = _fetch_once(url, timeout, repo, branch) 

156 if isinstance(outcome, BundlesDoc): 

157 return outcome # permanent HTTP error — do not retry 

158 if isinstance(outcome, bytes): 

159 return _parse_remote_bundles(outcome) 

160 # Transient failure (network/timeout): record it, then back off and retry. 

161 errors = [outcome] 

162 _log_failed_attempt(attempt, attempts, outcome, backoff) 

163 

164 return BundlesDoc(None, errors)