Coverage for scripts/sync.py: 100%
101 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-31 06:18 +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 excluded_set,
62 prepare_snapshot,
63)
64from _rhiza_template import Template, load_template # noqa: E402
66EXIT_OK = 0
67EXIT_CONFLICTS = 1
68EXIT_ERROR = 2
71def _now() -> str:
72 """Return the current UTC time as an ISO 8601 ``...Z`` timestamp (seam for tests)."""
73 return datetime.datetime.now(datetime.UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
76# ---------------------------------------------------------------------------
77# Template configuration
78# ---------------------------------------------------------------------------
81def _merge_with_base(
82 ctx: git.GitContext,
83 target: Path,
84 upstream_snapshot: Path,
85 base_sha: str,
86 base_snapshot: Path,
87 git_url: str,
88 include_paths: list[str],
89 excludes: set[str],
90 path_map: dict[str, str],
91) -> bool:
92 """Materialise the base snapshot and three-way merge base->upstream into *target*."""
93 base_clone = Path(tempfile.mkdtemp())
94 try:
95 git.clone(ctx, git_url, base_clone, include_paths, sha=base_sha)
96 prepare_snapshot(base_clone, include_paths, excludes, base_snapshot, path_map)
97 except (subprocess.CalledProcessError, OSError):
98 log("Could not check out base commit — treating all files as new")
99 finally:
100 shutil.rmtree(base_clone, ignore_errors=True)
102 outcome = merge.merge_trees(ctx, target, base_snapshot, upstream_snapshot)
103 if not (outcome.merged or outcome.conflicted or outcome.unmergeable or outcome.deleted):
104 log("Template unchanged since last sync — nothing to apply")
105 return True
107 log(
108 f"Merged {len(outcome.merged)} file(s)"
109 + (f", deleted {len(outcome.deleted)}" if outcome.deleted else "")
110 )
111 for path in outcome.conflicted:
112 log(f" conflict: {path} — `<<<<<<<` markers written")
113 for path in outcome.unmergeable:
114 # Named individually because there is no marker to find these by: the file was
115 # left exactly as the user had it, which is the safe choice but an invisible one.
116 log(f" cannot merge: {path} — locally modified and not text; left untouched")
117 return outcome.clean
120def _run_merge(
121 ctx: git.GitContext,
122 target: Path,
123 template: Template,
124 upstream_snapshot: Path,
125 upstream_sha: str,
126 base_sha: str | None,
127 template_files: list[Path],
128 include_paths: list[str],
129 excludes: set[str],
130 path_map: dict[str, str],
131 lock_path: Path,
132) -> bool:
133 """Apply the upstream snapshot to *target*, clean orphans, write the lock; return clean-ness."""
134 tracked_before = previously_tracked(lock_path)
135 base_snapshot = Path(tempfile.mkdtemp())
136 try:
137 if base_sha:
138 clean = _merge_with_base(
139 ctx,
140 target,
141 upstream_snapshot,
142 base_sha,
143 base_snapshot,
144 template.git_url,
145 include_paths,
146 excludes,
147 path_map,
148 )
149 else:
150 log("First sync — copying all template files")
151 copy_files(upstream_snapshot, target, template_files)
152 clean = True
154 missing = [p for p in template_files if not (target / p).exists()]
155 if missing:
156 log(f"Restoring {len(missing)} template file(s) missing from target")
157 copy_files(upstream_snapshot, target, missing)
159 clean_orphaned_files(target, template_files, excludes, tracked_before)
160 lock = build_lock(upstream_sha, template, [str(p) for p in template_files], _now())
161 write_lock(target, lock, lock_path)
162 finally:
163 shutil.rmtree(base_snapshot, ignore_errors=True)
164 return clean
167def sync(target: Path, branch: str) -> int:
168 """Run the sync and return a process exit code (see the module docstring)."""
169 target = target.resolve()
170 ctx = git.GitContext.default()
172 dirty = git.status_porcelain(ctx, target)
173 if dirty:
174 log("Working tree is not clean — commit or stash your changes before syncing:")
175 for line in dirty:
176 log(f" {line}")
177 return EXIT_ERROR
179 template = load_template(target, target / ".rhiza" / "template.yml")
180 lock_path = resolve_lock_path(target, None)
181 base_sha = read_base_sha(lock_path)
183 log(f"Cloning {template.repository}@{template.ref or branch}")
184 upstream_dir, upstream_sha, include_paths, path_map = clone_template(ctx, template, branch)
185 upstream_snapshot = Path(tempfile.mkdtemp())
186 try:
187 excludes = excluded_set(upstream_dir, template.exclude)
188 template_files = prepare_snapshot(
189 upstream_dir, include_paths, excludes, upstream_snapshot, path_map
190 )
191 log(f"Upstream: {len(template_files)} file(s) to consider")
192 clean = _run_merge(
193 ctx,
194 target,
195 template,
196 upstream_snapshot,
197 upstream_sha,
198 base_sha,
199 template_files,
200 include_paths,
201 excludes,
202 path_map,
203 lock_path,
204 )
205 finally:
206 shutil.rmtree(upstream_snapshot, ignore_errors=True)
207 shutil.rmtree(upstream_dir, ignore_errors=True)
209 if not clean:
210 log("Conflicts remain — resolve `<<<<<<<` markers, then commit.")
211 return EXIT_CONFLICTS
212 log(f"Sync complete — {len(template_files)} file(s) processed")
213 return EXIT_OK
216def main(argv: list[str] | None = None) -> int:
217 """Entry point: parse args, run the sync, and translate failures to exit codes."""
218 parser = argparse.ArgumentParser(
219 description="Sync rhiza template files into this repo (3-way merge)."
220 )
221 parser.add_argument(
222 "target",
223 nargs="?",
224 default=".",
225 help="Repository root to sync (default: current directory).",
226 )
227 parser.add_argument(
228 "--branch",
229 "-b",
230 default="main",
231 help="Template branch to use when template.yml has no `ref` (default: main).",
232 )
233 args = parser.parse_args(argv)
234 try:
235 return sync(Path(args.target), args.branch)
236 except SyncError as exc:
237 log(f"error: {exc}")
238 return EXIT_ERROR
239 except subprocess.CalledProcessError as exc:
240 stderr = exc.stderr.decode() if isinstance(exc.stderr, bytes) else (exc.stderr or "")
241 log(f"error: git failed: {stderr.strip() or exc}")
242 return EXIT_ERROR
243 except RuntimeError as exc:
244 log(f"error: {exc}")
245 return EXIT_ERROR
248if __name__ == "__main__":
249 raise SystemExit(main())