Coverage for plugin/scripts/_rhiza_lock.py: 100%
116 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-10-26 04:21 +0000
1"""Read and write `.rhiza/template.lock`, and remove files that left the template.
3The lock records the synced SHA and the exact file list, which two other things depend
4on: the next sync reads the SHA to find its merge base, and `stage_synced.py` reads the
5file list so `/update` can stage template-owned paths only.
7Orphan cleanup lives here because it is the lock's inverse — a file the previous lock
8tracked but the current file set no longer contains is one this sync must delete.
9"""
11from __future__ import annotations
13import os
14import stat
15import sys
16from dataclasses import dataclass
17from pathlib import Path
18from typing import Any
20sys.path.insert(0, str(Path(__file__).resolve().parent))
21from _rhiza_common import escapes_root, log # noqa: E402
22from _rhiza_yaml import as_list # noqa: E402
24# Never removed by orphan cleanup: without it the repo stops being rhiza-managed.
25_PROTECTED = frozenset({Path(".rhiza/template.yml")})
26from _rhiza_template import Template, is_excluded # noqa: E402
27from _rhiza_yaml import dump_yaml, load_yaml # noqa: E402
30@dataclass(frozen=True)
31class OwnershipConflict:
32 """An unmanaged target path that blocks a template file."""
34 path: Path
35 detail: str
36 adoptable: bool
37 alias: bool = False
40def lock_path(target: Path, lock_file: Path | None) -> Path:
41 """Return the lock-file path (explicit override or the default under .rhiza)."""
42 return lock_file if lock_file is not None else target / ".rhiza" / "template.lock"
45def previously_tracked(lock_path: Path) -> set[Path]:
46 """Return the file set recorded in an existing lock's ``files`` field.
48 Entries that would resolve outside the target root are dropped, loudly. This is the
49 first place a lock's paths are read, and the one where they are later **unlinked** —
50 `clean_orphaned_files` joins each onto *target* and deletes it — so a `..` entry here
51 would be a delete outside the repository rather than the read-only join `stage_synced`
52 guards. Nothing upstream produces one today; the containment was implicit, exactly as
53 `_PROTECTED` was before mutation testing found that any string would do.
54 """
55 if not lock_path.exists():
56 return set()
57 try:
58 lock = load_yaml(lock_path)
59 except (OSError, ValueError):
60 return set()
61 tracked: set[Path] = set()
62 for entry in as_list(lock.get("files")):
63 if escapes_root(str(entry)):
64 log(f"Ignoring lock entry outside the repository: {entry}")
65 continue
66 tracked.add(Path(entry))
67 return tracked
70def _path_kind(path: Path) -> str | None:
71 """Return the filesystem kind of *path*, including dangling symlinks."""
72 try:
73 mode = path.lstat().st_mode
74 except (FileNotFoundError, NotADirectoryError):
75 return None
76 if stat.S_ISDIR(mode):
77 return "directory"
78 if stat.S_ISLNK(mode):
79 return "symlink"
80 if stat.S_ISREG(mode):
81 return "file"
82 return "special file"
85def unmanaged_conflicts(
86 target: Path, template_files: list[Path], tracked: set[Path]
87) -> list[OwnershipConflict]:
88 """Find unmanaged files or directories that would block incoming template files.
90 GitHub Actions treats ``action.yml`` and ``action.yaml`` as the metadata file for the
91 same action directory, so an unmanaged sibling with either name also blocks adoption.
92 """
93 conflicts: dict[Path, OwnershipConflict] = {}
95 def add(path: Path, detail: str, *, adoptable: bool, alias: bool = False) -> None:
96 existing = conflicts.get(path)
97 if existing is None or (existing.adoptable and not adoptable):
98 conflicts[path] = OwnershipConflict(path, detail, adoptable, alias)
100 for rel in sorted(template_files):
101 destination = target / rel
102 kind = _path_kind(destination)
103 if kind is not None and rel not in tracked:
104 add(
105 rel,
106 f"unmanaged {kind}",
107 adoptable=kind in {"file", "symlink"},
108 )
110 for parent in rel.parents:
111 if parent == Path("."):
112 continue
113 parent_kind = _path_kind(target / parent)
114 if parent_kind is not None and parent_kind != "directory" and parent not in tracked:
115 add(
116 parent,
117 f"unmanaged {parent_kind} blocks the template directory",
118 adoptable=False,
119 )
121 if rel.name in {"action.yml", "action.yaml"}:
122 alias_name = "action.yaml" if rel.name == "action.yml" else "action.yml"
123 alias = rel.with_name(alias_name)
124 alias_kind = _path_kind(target / alias)
125 if alias_kind is not None and alias not in tracked:
126 add(
127 alias,
128 f"unmanaged {alias_kind} aliases {rel.as_posix()}",
129 adoptable=alias_kind in {"file", "symlink"},
130 alias=True,
131 )
133 return [conflicts[path] for path in sorted(conflicts)]
136def clean_orphaned_files(
137 target: Path, template_files: list[Path], excludes: set[str], previously_tracked: set[Path]
138) -> None:
139 """Delete files tracked by the previous sync that the template no longer provides.
141 An excluded path is never an orphan. Matching goes through :func:`is_excluded` rather
142 than comparing against *excludes* directly, so that a directory entry protects the
143 files under it: `excludes` now holds the configured destination paths verbatim, and a
144 set membership test would read `docs` and `docs/guide.md` as unrelated. Locks written
145 before the exclusion fix list excluded files in `files:`, which makes them look like
146 orphans on the next sync — this is what stops that from deleting them.
147 """
148 orphaned = previously_tracked - set(template_files) - set(_PROTECTED)
149 for rel in sorted(orphaned):
150 if is_excluded(rel.as_posix(), excludes):
151 continue
152 full = target / rel
153 if full.exists():
154 try:
155 full.unlink()
156 log(f"[DEL] {rel}")
157 except OSError as exc:
158 log(f"Failed to delete {rel}: {exc}")
161def _lock_identity(lock: dict[str, Any]) -> tuple[Any, ...]:
162 """Return the content-comparison key for a lock dict, excluding ``synced_at``."""
163 return (
164 str(lock.get("sha", "")),
165 str(lock.get("repo", "")),
166 str(lock.get("host", "")),
167 str(lock.get("ref", "")),
168 as_list(lock.get("include")),
169 as_list(lock.get("exclude")),
170 as_list(lock.get("templates")),
171 as_list(lock.get("files")),
172 str(lock.get("strategy", "")),
173 )
176def build_lock(sha: str, template: Template, files: list[str], synced_at: str) -> dict[str, Any]:
177 """Assemble the ordered lock dict (matching the CLI's field order) for serialisation."""
178 lock: dict[str, Any] = {
179 "sha": sha,
180 "repo": template.repository,
181 "host": template.host,
182 "ref": template.ref,
183 "include": template.include,
184 "exclude": template.exclude,
185 "templates": template.templates,
186 }
187 if template.profiles:
188 lock["profiles"] = template.profiles
189 lock["files"] = files
190 lock["synced_at"] = synced_at
191 lock["strategy"] = "merge"
192 return lock
195def write_lock(target: Path, lock: dict[str, Any], lock_path: Path) -> None:
196 """Write the lock atomically; filter ``files`` to on-disk paths and skip no-op rewrites."""
197 lock = dict(lock)
198 lock["files"] = sorted(f for f in as_list(lock.get("files")) if (target / f).exists())
200 if lock_path.exists():
201 try:
202 existing = load_yaml(lock_path)
203 except (OSError, ValueError):
204 existing = None
205 if existing is not None and _lock_identity(existing) == _lock_identity(lock):
206 log(f"{lock_path.name} is already up to date — skipping write")
207 return
209 lock_path.parent.mkdir(parents=True, exist_ok=True)
210 tmp_path = Path(str(lock_path) + ".tmp")
211 dump_yaml(lock, tmp_path)
212 os.replace(tmp_path, lock_path)
213 log(f"Updated {lock_path.name} -> {str(lock['sha'])[:12]}")
216# ---------------------------------------------------------------------------
217# Merge orchestration
218# ---------------------------------------------------------------------------
221def read_base_sha(lock_path: Path) -> str | None:
222 """Return the previously-synced SHA from the lock, or ``None`` for a first sync."""
223 if not lock_path.exists():
224 return None
225 try:
226 return str(load_yaml(lock_path).get("sha") or "") or None
227 except (OSError, ValueError):
228 return None