Coverage for plugin/scripts/sync.py: 100%
101 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-15 14:46 +0000
1#!/usr/bin/env python3
2"""Sync rhiza template files into this repo using a 3-way merge.
4A stdlib-only port of the `rhiza sync` command, bundled with this plugin so
5`/rhiza:sync` (and `/rhiza:update`) work without the `rhiza` CLI installed:
6clone the upstream template, materialise the previously-synced snapshot beside
7it, and three-way merge the two into the working tree — preserving local edits
8and leaving conflict markers where both sides changed a region.
10The merge itself lives in `_rhiza_merge.py`, which compares the two snapshot
11directories directly. It used to render a `git diff --no-index` of one against
12the other, apply it with `git apply -3`, then parse that diff text back into a
13file list to merge each file with `git merge-file` anyway. Since nothing runs
14`git apply` now, **`.rej` files are no longer produced**.
16Usage:
17 uv run --python 3.12 --no-project python scripts/sync.py [TARGET] [--branch BRANCH]
19 TARGET repository root to sync (default: current directory)
20 --branch template branch to use when template.yml has no `ref`
21 (default: main)
23Requires `git` on PATH and Python >= 3.11 (uses ``datetime.UTC``); run it under
24``uv run --python 3.12`` since the system ``python3`` may be older (macOS ships
253.9). **Mutates the working tree.** Exit codes:
26 0 synced cleanly (or already up to date)
27 1 synced with conflicts — resolve the `<<<<<<<` markers, then commit (this is
28 the expected outcome when local edits collide with upstream). Also returned
29 when a locally-modified binary file could not be merged, which is reported
30 by name since it leaves no marker behind.
31 2 could not sync (dirty tree, invalid template.yml, or a git failure)
32"""
34from __future__ import annotations
36import argparse
37import datetime
38import shutil
39import subprocess # nosec B404
40import sys
41import tempfile
42from pathlib import Path
44sys.path.insert(0, str(Path(__file__).resolve().parent))
45import _rhiza_git as git # noqa: E402
46import _rhiza_merge as merge # noqa: E402
47from _rhiza_common import SyncError, log # noqa: E402
48from _rhiza_lock import ( # noqa: E402
49 build_lock,
50 clean_orphaned_files,
51 previously_tracked,
52 read_base_sha,
53 write_lock,
54)
55from _rhiza_lock import (
56 lock_path as resolve_lock_path,
57)
58from _rhiza_snapshot import ( # noqa: E402
59 clone_template,
60 copy_files,
61 prepare_snapshot,
62)
63from _rhiza_template import Template, load_template, normalise_excludes # noqa: E402
65EXIT_OK = 0
66EXIT_CONFLICTS = 1
67EXIT_ERROR = 2
70def _now() -> str:
71 """Return the current UTC time as an ISO 8601 ``...Z`` timestamp (seam for tests)."""
72 return datetime.datetime.now(datetime.UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
75# ---------------------------------------------------------------------------
76# Template configuration
77# ---------------------------------------------------------------------------
80def _merge_with_base(
81 ctx: git.GitContext,
82 target: Path,
83 upstream_snapshot: Path,
84 base_sha: str,
85 base_snapshot: Path,
86 git_url: str,
87 include_paths: list[str],
88 excludes: set[str],
89 path_map: dict[str, str],
90) -> bool:
91 """Materialise the base snapshot and three-way merge base->upstream into *target*."""
92 base_clone = Path(tempfile.mkdtemp())
93 try:
94 git.clone(ctx, git_url, base_clone, include_paths, sha=base_sha)
95 prepare_snapshot(base_clone, include_paths, excludes, base_snapshot, path_map)
96 except (subprocess.CalledProcessError, OSError):
97 log("Could not check out base commit — treating all files as new")
98 finally:
99 shutil.rmtree(base_clone, ignore_errors=True)
101 outcome = merge.merge_trees(ctx, target, base_snapshot, upstream_snapshot)
102 if not (outcome.merged or outcome.conflicted or outcome.unmergeable or outcome.deleted):
103 log("Template unchanged since last sync — nothing to apply")
104 return True
106 log(
107 f"Merged {len(outcome.merged)} file(s)"
108 + (f", deleted {len(outcome.deleted)}" if outcome.deleted else "")
109 )
110 for path in outcome.conflicted:
111 log(f" conflict: {path} — `<<<<<<<` markers written")
112 for path in outcome.unmergeable:
113 # Named individually because there is no marker to find these by: the file was
114 # left exactly as the user had it, which is the safe choice but an invisible one.
115 log(f" cannot merge: {path} — locally modified and not text; left untouched")
116 return outcome.clean
119def _run_merge(
120 ctx: git.GitContext,
121 target: Path,
122 template: Template,
123 upstream_snapshot: Path,
124 upstream_sha: str,
125 base_sha: str | None,
126 template_files: list[Path],
127 include_paths: list[str],
128 excludes: set[str],
129 path_map: dict[str, str],
130 lock_path: Path,
131) -> bool:
132 """Apply the upstream snapshot to *target*, clean orphans, write the lock; return clean-ness."""
133 tracked_before = previously_tracked(lock_path)
134 base_snapshot = Path(tempfile.mkdtemp())
135 try:
136 if base_sha:
137 clean = _merge_with_base(
138 ctx,
139 target,
140 upstream_snapshot,
141 base_sha,
142 base_snapshot,
143 template.git_url,
144 include_paths,
145 excludes,
146 path_map,
147 )
148 else:
149 log("First sync — copying all template files")
150 copy_files(upstream_snapshot, target, template_files)
151 clean = True
153 missing = [p for p in template_files if not (target / p).exists()]
154 if missing:
155 log(f"Restoring {len(missing)} template file(s) missing from target")
156 copy_files(upstream_snapshot, target, missing)
158 clean_orphaned_files(target, template_files, excludes, tracked_before)
159 lock = build_lock(upstream_sha, template, [str(p) for p in template_files], _now())
160 write_lock(target, lock, lock_path)
161 finally:
162 shutil.rmtree(base_snapshot, ignore_errors=True)
163 return clean
166def sync(target: Path, branch: str) -> int:
167 """Run the sync and return a process exit code (see the module docstring)."""
168 target = target.resolve()
169 ctx = git.GitContext.default()
171 dirty = git.status_porcelain(ctx, target)
172 if dirty:
173 log("Working tree is not clean — commit or stash your changes before syncing:")
174 for line in dirty:
175 log(f" {line}")
176 return EXIT_ERROR
178 template = load_template(target, target / ".rhiza" / "template.yml")
179 lock_path = resolve_lock_path(target, None)
180 base_sha = read_base_sha(lock_path)
182 log(f"Cloning {template.repository}@{template.ref or branch}")
183 upstream_dir, upstream_sha, include_paths, path_map = clone_template(ctx, template, branch)
184 upstream_snapshot = Path(tempfile.mkdtemp())
185 try:
186 excludes = normalise_excludes(template.exclude)
187 template_files = prepare_snapshot(
188 upstream_dir, include_paths, excludes, upstream_snapshot, path_map
189 )
190 log(f"Upstream: {len(template_files)} file(s) to consider")
191 clean = _run_merge(
192 ctx,
193 target,
194 template,
195 upstream_snapshot,
196 upstream_sha,
197 base_sha,
198 template_files,
199 include_paths,
200 excludes,
201 path_map,
202 lock_path,
203 )
204 finally:
205 shutil.rmtree(upstream_snapshot, ignore_errors=True)
206 shutil.rmtree(upstream_dir, ignore_errors=True)
208 if not clean:
209 log("Conflicts remain — resolve `<<<<<<<` markers, then commit.")
210 return EXIT_CONFLICTS
211 log(f"Sync complete — {len(template_files)} file(s) processed")
212 return EXIT_OK
215def main(argv: list[str] | None = None) -> int:
216 """Entry point: parse args, run the sync, and translate failures to exit codes."""
217 parser = argparse.ArgumentParser(
218 description="Sync rhiza template files into this repo (3-way merge)."
219 )
220 parser.add_argument(
221 "target",
222 nargs="?",
223 default=".",
224 help="Repository root to sync (default: current directory).",
225 )
226 parser.add_argument(
227 "--branch",
228 "-b",
229 default="main",
230 help="Template branch to use when template.yml has no `ref` (default: main).",
231 )
232 args = parser.parse_args(argv)
233 try:
234 return sync(Path(args.target), args.branch)
235 except SyncError as exc:
236 log(f"error: {exc}")
237 return EXIT_ERROR
238 except subprocess.CalledProcessError as exc:
239 stderr = exc.stderr.decode() if isinstance(exc.stderr, bytes) else (exc.stderr or "")
240 log(f"error: git failed: {stderr.strip() or exc}")
241 return EXIT_ERROR
242 except RuntimeError as exc:
243 log(f"error: {exc}")
244 return EXIT_ERROR
247if __name__ == "__main__":
248 raise SystemExit(main())