Coverage for src / rhiza / models / bundle.py: 100%
187 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-07-16 12:51 +0000
1"""Bundle models for Rhiza configuration."""
3from dataclasses import dataclass, field
4from pathlib import PurePosixPath
5from typing import Any
7from rhiza.models._base import YamlSerializable
8from rhiza.models._git.helpers import _normalize_to_list
11def _ensure_safe_bundle_path(value: str) -> None:
12 """Reject a bundle path that could escape the project directory.
14 ``template.yml`` is effectively untrusted input (it is fetched from the
15 template repository), and a remapped ``dest`` is joined onto the target
16 directory to decide where a file is written. An absolute path, a Windows
17 drive letter, or a ``..`` traversal component could therefore write outside
18 the project. This validates both ``source`` and ``dest`` at the trust
19 boundary so no such path can reach the copy step.
21 Args:
22 value: A ``source`` or ``dest`` path from bundle config.
24 Raises:
25 ValueError: If *value* is absolute, uses a drive letter, or contains a
26 ``..`` traversal component.
27 """
28 # Normalise separators so a Windows-style path cannot slip past the checks.
29 normalized = value.replace("\\", "/")
30 pure = PurePosixPath(normalized)
31 has_drive = len(normalized) >= 2 and normalized[0].isalpha() and normalized[1] == ":"
32 if pure.is_absolute() or has_drive or ".." in pure.parts:
33 raise ValueError( # noqa: TRY003
34 f"Unsafe bundle path {value!r}: paths must be relative to the project "
35 "root (no absolute paths, drive letters, or '..' traversal)."
36 )
39@dataclass(frozen=True)
40class BundleFileEntry:
41 """A file entry in a bundle, with optional source→destination path remapping.
43 When ``source`` and ``dest`` differ, the file is read from ``source`` in the
44 template repository but written to ``dest`` in the downstream project.
46 Attributes:
47 source: Path of the file in the template repository.
48 dest: Path where the file should be placed in the downstream project.
49 """
51 source: str
52 dest: str
54 def __post_init__(self) -> None:
55 """Validate that both paths stay within the project (no escape)."""
56 _ensure_safe_bundle_path(self.source)
57 _ensure_safe_bundle_path(self.dest)
59 @property
60 def is_remapped(self) -> bool:
61 """True when the destination path differs from the source path."""
62 return self.source != self.dest
64 @classmethod
65 def from_config_entry(cls, entry: "str | dict[str, str]") -> "BundleFileEntry":
66 """Parse a file entry from a YAML config value (string or dict).
68 Args:
69 entry: Either a plain path string or a ``{source, dest}`` dict.
71 Returns:
72 A :class:`BundleFileEntry` instance.
74 Raises:
75 TypeError: If the entry is a dict without a ``source`` key.
76 """
77 if isinstance(entry, str):
78 return cls(source=entry, dest=entry)
79 if not isinstance(entry, dict) or "source" not in entry:
80 raise TypeError( # noqa: TRY003
81 f"File entry must be a string or a dict with a 'source' key, got: {entry!r}"
82 )
83 source = entry["source"]
84 dest = entry.get("dest", source)
85 return cls(source=source, dest=dest)
87 def to_config_entry(self) -> "str | dict[str, str]":
88 """Serialize back to the YAML config representation."""
89 if not self.is_remapped:
90 return self.source
91 return {"source": self.source, "dest": self.dest}
93 def remap_expanded_path(self, expanded_source: str) -> str:
94 """Map an expanded source path to its destination path.
96 Handles both exact file matches and directory-prefix matches.
98 Args:
99 expanded_source: A source path produced by expanding this entry.
101 Returns:
102 The corresponding destination path.
103 """
104 if not self.is_remapped:
105 return expanded_source
106 if expanded_source == self.source:
107 return self.dest
108 src_prefix = self.source.rstrip("/") + "/"
109 if expanded_source.startswith(src_prefix):
110 dest_prefix = self.dest.rstrip("/") + "/"
111 return dest_prefix + expanded_source[len(src_prefix) :]
112 return expanded_source
115@dataclass(frozen=True, kw_only=True)
116class ProfileDefinition:
117 """Represents a single profile from template-bundles.yml.
119 Attributes:
120 description: Human-readable description of the profile.
121 bundles: List of bundle names included in this profile.
122 """
124 description: str = ""
125 bundles: list[str] = field(default_factory=list)
128@dataclass(frozen=True, kw_only=True)
129class BundleDefinition:
130 """Represents a single bundle from template-bundles.yml.
132 Attributes:
133 description: Human-readable description of the bundle.
134 files: Explicit file entries (legacy format only — new bundles own files via
135 their ``bundles/<name>/`` directory in the template repository).
136 requires: List of bundle names that this bundle requires.
137 recommends: List of bundle names that this bundle recommends (soft deps).
138 standalone: Whether this bundle is standalone (no dependencies).
139 required: Whether this bundle is mandatory (always included).
140 notes: Free-form notes for maintainers (not synced to downstream projects).
141 """
143 description: str
144 standalone: bool = True
145 required: bool = False
146 files: list[BundleFileEntry] = field(default_factory=list)
147 requires: list[str] = field(default_factory=list)
148 recommends: list[str] = field(default_factory=list)
149 notes: str = ""
152def _parse_bundle_files(raw_files: Any) -> list[BundleFileEntry]:
153 """Coerce a bundle's raw ``files`` field into a list of file entries.
155 Args:
156 raw_files: The raw value of the ``files`` key (list, string, or absent).
158 Returns:
159 Parsed file entries; an empty list when *raw_files* is neither a list
160 nor a string.
161 """
162 if isinstance(raw_files, list):
163 return [BundleFileEntry.from_config_entry(e) for e in raw_files]
164 if isinstance(raw_files, str):
165 return [BundleFileEntry.from_config_entry(e) for e in _normalize_to_list(raw_files)]
166 return []
169def _parse_bundle_definitions(bundles_config: Any) -> dict[str, BundleDefinition]:
170 """Parse the ``bundles`` mapping of a config dict into definitions.
172 Args:
173 bundles_config: The raw ``bundles`` value from the configuration.
175 Returns:
176 Mapping of bundle name to :class:`BundleDefinition`.
178 Raises:
179 TypeError: If *bundles_config* or any entry is not a dictionary.
180 """
181 if not isinstance(bundles_config, dict):
182 msg = "Bundles must be a dictionary"
183 raise TypeError(msg)
185 bundles: dict[str, BundleDefinition] = {}
186 for bundle_name, bundle_data in bundles_config.items():
187 if not isinstance(bundle_data, dict):
188 msg = f"Bundle '{bundle_name}' must be a dictionary"
189 raise TypeError(msg)
190 bundles[bundle_name] = BundleDefinition(
191 description=bundle_data.get("description", ""),
192 files=_parse_bundle_files(bundle_data.get("files")),
193 requires=_normalize_to_list(bundle_data.get("requires")),
194 recommends=_normalize_to_list(bundle_data.get("recommends")),
195 standalone=bundle_data.get("standalone", True),
196 required=bool(bundle_data.get("required", False)),
197 notes=bundle_data.get("notes") or "",
198 )
199 return bundles
202def _parse_profile_definitions(profiles_config: Any) -> dict[str, ProfileDefinition]:
203 """Parse the ``profiles`` mapping of a config dict into definitions.
205 Args:
206 profiles_config: The raw ``profiles`` value from the configuration
207 (``None`` is treated as an empty mapping).
209 Returns:
210 Mapping of profile name to :class:`ProfileDefinition`.
212 Raises:
213 TypeError: If *profiles_config* or any entry is not a dictionary.
214 """
215 if profiles_config is None:
216 profiles_config = {}
217 elif not isinstance(profiles_config, dict):
218 msg = "Profiles must be a dictionary"
219 raise TypeError(msg)
221 profiles: dict[str, ProfileDefinition] = {}
222 for profile_name, profile_data in profiles_config.items():
223 if not isinstance(profile_data, dict):
224 msg = f"Profile '{profile_name}' must be a dictionary"
225 raise TypeError(msg)
226 profiles[profile_name] = ProfileDefinition(
227 description=profile_data.get("description", ""),
228 bundles=_normalize_to_list(profile_data.get("bundles")),
229 )
230 return profiles
233@dataclass(frozen=True, kw_only=True)
234class RhizaBundles(YamlSerializable):
235 """Represents the structure of template-bundles.yml.
237 Attributes:
238 version: Optional version string of the bundles configuration format.
239 bundles: Dictionary mapping bundle names to their definitions.
240 """
242 version: str | None = None
243 bundles: dict[str, BundleDefinition] = field(default_factory=dict)
244 profiles: dict[str, ProfileDefinition] = field(default_factory=dict)
246 @staticmethod
247 def _bundle_to_entry(bundle: BundleDefinition) -> dict[str, Any]:
248 """Serialise a single bundle definition into its config dict, omitting falsy fields."""
249 entry: dict[str, Any] = {"description": bundle.description}
250 if bundle.required:
251 entry["required"] = bundle.required
252 if bundle.standalone:
253 entry["standalone"] = bundle.standalone
254 if bundle.requires:
255 entry["requires"] = bundle.requires
256 if bundle.recommends:
257 entry["recommends"] = bundle.recommends
258 if bundle.files:
259 entry["files"] = [f.to_config_entry() for f in bundle.files]
260 if bundle.notes:
261 entry["notes"] = bundle.notes
262 return entry
264 @staticmethod
265 def _profile_to_entry(profile: ProfileDefinition) -> dict[str, Any]:
266 """Serialise a single profile definition into its config dict."""
267 entry: dict[str, Any] = {}
268 if profile.description:
269 entry["description"] = profile.description
270 entry["bundles"] = profile.bundles
271 return entry
273 @property
274 def config(self) -> dict[str, Any]:
275 """Return the bundles' current state as a configuration dictionary."""
276 config: dict[str, Any] = {}
278 if self.version is not None:
279 config["version"] = self.version
281 config["bundles"] = {name: self._bundle_to_entry(bundle) for name, bundle in self.bundles.items()}
283 if self.profiles:
284 config["profiles"] = {name: self._profile_to_entry(profile) for name, profile in self.profiles.items()}
286 return config
288 @classmethod
289 def from_config(cls, config: dict[str, Any]) -> "RhizaBundles":
290 """Create a RhizaBundles instance from a configuration dictionary.
292 Args:
293 config: Dictionary containing bundles configuration.
295 Returns:
296 A new RhizaBundles instance.
298 Raises:
299 TypeError: If bundle data has invalid types.
300 """
301 version = config.get("version")
302 bundles = _parse_bundle_definitions(config.get("bundles", {}))
303 profiles = _parse_profile_definitions(config.get("profiles", {}))
304 return cls(version=version, bundles=bundles, profiles=profiles)
306 def _resolve_bundle_order(self, bundle_names: list[str], *, strict: bool) -> list[str]:
307 """Return *bundle_names* and their ``requires`` dependencies in topological order.
309 Args:
310 bundle_names: Bundle names to resolve.
311 strict: When True, raise ``ValueError`` for unknown bundles or
312 circular dependencies; when False, silently skip them.
314 Returns:
315 Dependency-first ordering of the resolved bundle names.
317 Raises:
318 ValueError: If ``strict`` and a bundle is missing or forms a cycle.
319 """
320 order: list[str] = []
321 resolved: set[str] = set()
322 resolving: set[str] = set()
324 def _collect(name: str) -> None:
325 """Recursively resolve a single bundle's dependencies in topological order."""
326 if name not in self.bundles:
327 if strict:
328 msg = f"Bundle '{name}' does not exist"
329 raise ValueError(msg)
330 return
331 if name in resolving:
332 if strict:
333 msg = f"Circular dependency detected for bundle '{name}'"
334 raise ValueError(msg)
335 return
336 if name in resolved:
337 return
339 resolving.add(name)
340 for dependency in self.bundles[name].requires:
341 _collect(dependency)
342 resolving.remove(name)
343 resolved.add(name)
344 order.append(name)
346 for name in bundle_names:
347 _collect(name)
348 return order
350 def resolve_to_paths(self, bundle_names: list[str]) -> list[str]:
351 """Convert bundle names to deduplicated file paths.
353 Args:
354 bundle_names: List of bundle names to resolve.
356 Returns:
357 Deduplicated list of file paths from all bundles and their dependencies.
359 Raises:
360 ValueError: If a bundle doesn't exist or circular dependency detected.
361 """
362 paths: list[str] = []
363 seen: set[str] = set()
365 def _add(path: str) -> None:
366 """Append a path once, tracking it in the dedup set."""
367 if path not in seen:
368 paths.append(path)
369 seen.add(path)
371 for bundle_name in self._resolve_bundle_order(bundle_names, strict=True):
372 bundle = self.bundles[bundle_name]
373 if bundle.files:
374 for entry in bundle.files:
375 _add(entry.source)
376 else:
377 _add(f"bundles/{bundle_name}/")
379 return paths
381 def resolve_to_path_map(self, bundle_names: list[str]) -> dict[str, str]:
382 """Return a source→destination mapping for all remapped file entries.
384 Plain (non-remapped) entries are excluded — callers can assume an
385 absent key means ``dest == source``.
387 Args:
388 bundle_names: List of bundle names to resolve (dependencies included).
390 Returns:
391 Dict mapping source path → destination path for remapped entries only.
392 """
393 path_map: dict[str, str] = {}
394 resolved = self.resolve_to_paths(bundle_names)
395 resolved_set = set(resolved)
397 for bundle_name in self._resolve_bundle_order(bundle_names, strict=False):
398 bundle = self.bundles[bundle_name]
399 if bundle.files:
400 for entry in bundle.files:
401 if entry.source in resolved_set and entry.is_remapped:
402 path_map[entry.source] = entry.dest
403 else:
404 path_map[f"bundles/{bundle_name}/"] = ""
406 return path_map
408 def resolve_profile_to_paths(self, profile_name: str) -> list[str]:
409 """Resolve a profile name to deduplicated file paths.
411 Args:
412 profile_name: Name of the profile to resolve.
414 Returns:
415 Deduplicated list of file paths from all bundles in the profile.
417 Raises:
418 ValueError: If the profile doesn't exist or a referenced bundle doesn't exist.
419 """
420 if profile_name not in self.profiles:
421 msg = f"Profile '{profile_name}' does not exist"
422 raise ValueError(msg)
423 return self.resolve_to_paths(self.profiles[profile_name].bundles)