Coverage for scripts/_rhiza_bundles.py: 100%

1"""Resolve profiles and bundles to the file paths a sync should copy. 

2 

3The largest single job in the sync and the one with no I/O at all: `profiles` expand to 

4bundle names, bundle names expand to `(source, dest)` file entries, and the result is an 

5ordered, de-duplicated path list plus a remap table. 

6 

7It also owns the path-safety check. `template-bundles.yml` comes from the template repo, 

8so a `dest` is untrusted input that gets joined onto the target directory — an absolute 

9path, a drive letter or a `..` component would write outside the project. 

10""" 

11 

12from __future__ import annotations 

13 

14import sys 

15from dataclasses import dataclass 

16from pathlib import Path, PurePosixPath 

17from typing import Any 

18 

19sys.path.insert(0, str(Path(__file__).resolve().parent)) 

20from _rhiza_common import SyncError # noqa: E402 

21from _rhiza_template import Template # noqa: E402 

22from _rhiza_yaml import as_list # noqa: E402 

23 

24 

25def _ensure_safe_bundle_path(value: str) -> None: 

26 """Reject a bundle path that could escape the project directory. 

27 

28 ``template-bundles.yml`` is untrusted (fetched from the template repo) and a 

29 remapped ``dest`` is joined onto the target directory, so an absolute path, a 

30 Windows drive letter, or a ``..`` component could write outside the project. 

31 

32 Raises: 

33 SyncError: If *value* is absolute, uses a drive letter, or traverses up. 

34 """ 

35 normalized = value.replace("\\", "/") 

36 pure = PurePosixPath(normalized) 

37 has_drive = len(normalized) >= 2 and normalized[0].isalpha() and normalized[1] == ":" 

38 if pure.is_absolute() or has_drive or ".." in pure.parts: 

39 raise SyncError( 

40 f"Unsafe bundle path {value!r}: paths must be relative to the project root " 

41 "(no absolute paths, drive letters, or '..' traversal)." 

42 ) 

43 

44 

45def _bundle_file_entries(raw_files: Any) -> list[tuple[str, str]]: 

46 """Coerce a bundle's ``files`` field into validated ``(source, dest)`` pairs.""" 

47 entries: list[tuple[str, str]] = [] 

48 for entry in as_list(raw_files) if isinstance(raw_files, str) else (raw_files or []): 

49 if isinstance(entry, str): 

50 source = dest = entry 

51 elif isinstance(entry, dict) and "source" in entry: 

52 source = str(entry["source"]) 

53 dest = str(entry.get("dest", source)) 

54 else: 

55 raise SyncError( 

56 f"Bundle file entry must be a string or a {{source, dest}} map, got: {entry!r}" 

57 ) 

58 _ensure_safe_bundle_path(source) 

59 _ensure_safe_bundle_path(dest) 

60 entries.append((source, dest)) 

61 return entries 

62 

63 

64@dataclass(frozen=True) 

65class Bundles: 

66 """The bundle/profile definitions from `template-bundles.yml` that sync needs.""" 

67 

68 requires: dict[str, list[str]] 

69 files: dict[str, list[tuple[str, str]]] 

70 profiles: dict[str, list[str]] 

71 

72 @classmethod 

73 def from_config(cls, config: dict[str, Any]) -> Bundles: 

74 """Parse a `template-bundles.yml` dict into requires/files/profiles maps.""" 

75 raw_bundles = config.get("bundles") or {} 

76 raw_profiles = config.get("profiles") or {} 

77 requires: dict[str, list[str]] = {} 

78 files: dict[str, list[tuple[str, str]]] = {} 

79 for name, data in raw_bundles.items(): 

80 data = data or {} 

81 requires[name] = as_list(data.get("requires")) 

82 files[name] = _bundle_file_entries(data.get("files")) 

83 profiles = { 

84 name: as_list((data or {}).get("bundles")) for name, data in raw_profiles.items() 

85 } 

86 return cls(requires=requires, files=files, profiles=profiles) 

87 

88 def _order(self, names: list[str], *, strict: bool) -> list[str]: 

89 """Return *names* plus their ``requires`` dependencies in dependency-first order.""" 

90 order: list[str] = [] 

91 resolved: set[str] = set() 

92 resolving: set[str] = set() 

93 

94 def _collect(name: str) -> None: 

95 if name not in self.requires: 

96 if strict: 

97 raise SyncError(f"Bundle '{name}' does not exist") 

98 return 

99 if name in resolving: 

100 if strict: 

101 raise SyncError(f"Circular dependency detected for bundle '{name}'") 

102 return 

103 if name in resolved: 

104 return 

105 resolving.add(name) 

106 for dependency in self.requires[name]: 

107 _collect(dependency) 

108 resolving.discard(name) 

109 resolved.add(name) 

110 order.append(name) 

111 

112 for name in names: 

113 _collect(name) 

114 return order 

115 

116 def resolve_to_paths(self, names: list[str]) -> list[str]: 

117 """Resolve bundle *names* (and dependencies) to a deduplicated source-path list.""" 

118 paths: list[str] = [] 

119 seen: set[str] = set() 

120 for name in self._order(names, strict=True): 

121 entries = self.files[name] 

122 sources = [source for source, _ in entries] if entries else [f"bundles/{name}/"] 

123 for source in sources: 

124 if source not in seen: 

125 seen.add(source) 

126 paths.append(source) 

127 return paths 

128 

129 def resolve_to_path_map(self, names: list[str]) -> dict[str, str]: 

130 """Return a source->dest map for remapped entries (and dir bundles map to '').""" 

131 resolved = set(self.resolve_to_paths(names)) 

132 path_map: dict[str, str] = {} 

133 for name in self._order(names, strict=False): 

134 entries = self.files[name] 

135 if entries: 

136 for source, dest in entries: 

137 if source in resolved and source != dest: 

138 path_map[source] = dest 

139 else: 

140 path_map[f"bundles/{name}/"] = "" 

141 return path_map 

142 

143 

144def resolve_bundle_names(template: Template, bundles: Bundles) -> list[str]: 

145 """Expand configured profiles to bundle names and merge with explicit templates.""" 

146 if not template.profiles: 

147 return template.templates 

148 names: list[str] = [] 

149 for profile in template.profiles: 

150 if profile not in bundles.profiles: 

151 available = ", ".join(sorted(bundles.profiles)) or "none" 

152 raise SyncError(f"Profile '{profile}' was not found. Available profiles: {available}") 

153 for bundle in bundles.profiles[profile]: 

154 if bundle not in names: 

155 names.append(bundle) 

156 return list(dict.fromkeys(names + template.templates)) 

157 

158 

159# --------------------------------------------------------------------------- 

160# Cloning + snapshot preparation 

161# ---------------------------------------------------------------------------