Coverage for plugin/scripts/status.py: 100%
129 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-15 14:46 +0000
1#!/usr/bin/env python3
2"""Show the current rhiza sync status from `.rhiza/template.lock`.
4A stdlib-only port of the `rhiza status` command, bundled with this plugin so
5`/rhiza:status` works without the `rhiza` CLI (or PyYAML) installed. It reads
6the authoritative lock written by the last sync and reports the template
7repository, ref, SHA, sync timestamp, strategy, and included templates/paths.
8With `--files` it also renders the managed files as a directory tree — the view
9that used to live in the separate `/rhiza:tree` command.
11Usage:
12 uv run --python 3.12 --no-project python scripts/status.py [TARGET] [--json] [--files] [--check]
14 TARGET repository root to inspect (default: current directory)
15 --json emit a single JSON object on stdout instead of human-readable lines
16 --files append the managed files as a `tree`-style listing (human output)
17 --check compare the pinned ref to the latest upstream release (needs network)
19When no lock is present the repo has never been synced; this prints a hint to
20stderr and exits 0 (nothing to report is not an error). The `--json` payload
21mirrors `rhiza status --json` field-for-field (including `files`), so tools like
22stats.py can read either interchangeably. Where the CLI uses `rich` for the file
23tree, this renders plain ASCII connectors so it stays dependency-free.
25Everything is offline and deterministic except `--check`, which shells out to
26`git ls-remote --tags` (no `gh`, no auth for public repos) to see whether a newer
27release exists; a network or git failure there is reported, never fatal.
28"""
30from __future__ import annotations
32import argparse
33import json
34import re
35import shutil
36import subprocess # nosec B404
37import sys
38from pathlib import Path
39from typing import Any
41_SEMVER_RE = re.compile(r"^v?(\d+)\.(\d+)\.(\d+)$")
43sys.path.insert(0, str(Path(__file__).resolve().parent))
44from _rhiza_yaml import load_yaml # noqa: E402
46LOCK_REL = Path(".rhiza") / "template.lock"
49def _as_list(value: Any) -> list[str]:
50 """Normalise a scalar/None/list lock field to a list of strings."""
51 if value is None:
52 return []
53 if isinstance(value, list):
54 return [str(x) for x in value]
55 return [str(value)]
58def _status_dict(lock: dict[str, Any]) -> dict[str, Any]:
59 """Build the machine-readable status payload from a parsed lock.
61 Mirrors the field set emitted by `rhiza status --json`.
62 """
63 host = str(lock.get("host", "github"))
64 repo = str(lock.get("repo", ""))
65 return {
66 "repository": f"{host}/{repo}" if repo else host,
67 "host": host,
68 "repo": repo,
69 "ref": str(lock.get("ref", "main")),
70 "sha": str(lock.get("sha", "")),
71 "synced_at": str(lock.get("synced_at", "")),
72 "strategy": str(lock.get("strategy", "")),
73 "templates": _as_list(lock.get("templates")),
74 "include": _as_list(lock.get("include")),
75 "files": _as_list(lock.get("files")),
76 }
79def _build_tree(paths: list[str]) -> dict[str, Any]:
80 """Fold a flat path list into a nested {name: subtree} dict."""
81 root: dict[str, Any] = {}
82 for path in sorted(paths):
83 node = root
84 for part in Path(path).parts:
85 node = node.setdefault(part, {})
86 return root
89def _render(node: dict[str, Any], prefix: str = "") -> list[str]:
90 """Render a nested tree dict into Unix-`tree`-style lines."""
91 lines: list[str] = []
92 items = sorted(node.items())
93 for i, (name, child) in enumerate(items):
94 last = i == len(items) - 1
95 lines.append(f"{prefix}{'└── ' if last else '├── '}{name}")
96 if child:
97 lines.extend(_render(child, prefix + (" " if last else "│ ")))
98 return lines
101def _print_file_tree(files: list[str]) -> None:
102 """Print the managed files as a `tree`-style listing plus a total count."""
103 if not files:
104 print("No files are tracked in template.lock", file=sys.stderr)
105 return
106 print("\nFiles managed by Rhiza:")
107 print(".")
108 for line in _render(_build_tree(files)):
109 print(line)
110 print(f"\n{len(files)} file{'s' if len(files) != 1 else ''} managed by Rhiza")
113def _parse_semver(tag: str) -> tuple[int, int, int] | None:
114 """Parse a `vX.Y.Z` tag into a (major, minor, patch) tuple, or None."""
115 m = _SEMVER_RE.match(tag)
116 return (int(m.group(1)), int(m.group(2)), int(m.group(3))) if m else None
119def _remote_url(host: str, repo: str) -> str:
120 """Build an HTTPS clone URL for the given host alias and owner/repo slug."""
121 base = "gitlab.com" if "gitlab" in host else "github.com"
122 return f"https://{base}/{repo}"
125def _remote_tags(host: str, repo: str) -> list[str]:
126 """Return the remote's tag names via `git ls-remote --tags`, or [] on failure.
128 The git binary is resolved through `shutil.which` rather than left to a ``PATH``
129 lookup at exec time, matching every other call site here. A bare ``"git"`` is
130 resolved against whatever ``PATH`` happens to hold, which is the one difference
131 between this and the rest of the module (ruff's S607).
132 """
133 git = shutil.which("git")
134 if git is None: # pragma: no cover - git is present everywhere this runs
135 return []
136 try:
137 proc = subprocess.run( # nosec B603 # noqa: S603 - a fixed argv, never a shell
138 [git, "ls-remote", "--tags", _remote_url(host, repo)],
139 capture_output=True,
140 text=True,
141 check=True,
142 timeout=20,
143 )
144 except (OSError, subprocess.SubprocessError):
145 return []
146 tags: set[str] = set()
147 for line in proc.stdout.splitlines():
148 _, _, name = line.partition("refs/tags/")
149 if name:
150 tags.add(name.removesuffix("^{}"))
151 return sorted(tags)
154Release = tuple[str, tuple[int, ...]]
157def _releases(tags: list[str]) -> list[Release]:
158 """Pair each tag that parses as a release with its version, dropping the rest."""
159 return [(tag, version) for tag in tags if (version := _parse_semver(tag)) is not None]
162def _behind_count(releases: list[Release], current: tuple[int, ...]) -> int:
163 """Count how many of *releases* are newer than *current*."""
164 return sum(1 for _, version in releases if version > current)
167def _outdated_message(ref: str, tags: list[str]) -> str:
168 """Compose a one-line 'update available?' summary for *ref* against *tags*."""
169 releases = _releases(tags)
170 if not releases:
171 return "Update : could not determine the latest release"
172 latest_tag, _ = max(releases, key=lambda pair: pair[1])
173 current = _parse_semver(ref)
174 if current is None:
175 return (
176 f"Update : latest release is {latest_tag} "
177 f"(current ref '{ref}' is not a release tag) — run /rhiza:update, or sync.py"
178 )
179 behind = _behind_count(releases, current)
180 if behind == 0:
181 return f"Update : up to date ({latest_tag} is the latest release)"
182 plural = "s" if behind != 1 else ""
183 return (
184 f"Update : {ref} → {latest_tag} ({behind} release{plural} behind) "
185 "— run /rhiza:update, or sync.py"
186 )
189def _print_outdated(payload: dict[str, Any]) -> None:
190 """Print the outdated-check line for a status payload (network via git)."""
191 if not payload["repo"]:
192 print("Update : no template repository recorded in the lock")
193 return
194 tags = _remote_tags(payload["host"], payload["repo"])
195 print(_outdated_message(payload["ref"], tags))
198def _print_summary(payload: dict[str, Any]) -> None:
199 """Print the human-readable header block for a lock file.
201 `templates` and `include` are alternatives, not both — the lock records whichever
202 selection mode the pointer used, so printing one or the other keeps the output honest
203 about which mode is in force.
204 """
205 print(f"Repository : {payload['repository']}")
206 print(f"Ref : {payload['ref']}")
207 sha = payload["sha"]
208 print(f"SHA : {sha[:12]}" if sha else "SHA : (unknown)")
209 print(f"Synced at : {payload['synced_at'] or '(unknown)'}")
210 print(f"Strategy : {payload['strategy'] or '(unknown)'}")
211 if payload["templates"]:
212 print(f"Templates : {', '.join(payload['templates'])}")
213 elif payload["include"]:
214 print(f"Include : {', '.join(payload['include'])}")
217def status(
218 target: Path,
219 *,
220 json_output: bool = False,
221 show_files: bool = False,
222 check: bool = False,
223) -> int:
224 """Print the sync status; return a process exit code."""
225 lock_path = (target / LOCK_REL).resolve()
226 if not lock_path.exists():
227 print(
228 "No template.lock found — this repo has never been synced. "
229 "Run /rhiza:update, or sync.py directly.",
230 file=sys.stderr,
231 )
232 return 0
234 try:
235 lock = load_yaml(lock_path)
236 except (OSError, ValueError) as exc:
237 print(f"Could not read {lock_path}: {exc}", file=sys.stderr)
238 return 1
240 payload = _status_dict(lock)
241 if json_output:
242 print(json.dumps(payload, indent=2))
243 return 0
245 _print_summary(payload)
246 if check:
247 _print_outdated(payload)
248 if show_files:
249 _print_file_tree(payload["files"])
250 return 0
253def main(argv: list[str] | None = None) -> int:
254 """Entry point: print status; return an exit code."""
255 parser = argparse.ArgumentParser(
256 description="Show the current rhiza sync status from .rhiza/template.lock.",
257 )
258 parser.add_argument(
259 "target",
260 nargs="?",
261 default=".",
262 help="Repository root to inspect (default: current directory).",
263 )
264 parser.add_argument(
265 "--json",
266 dest="json_output",
267 action="store_true",
268 help="Emit the status as a single JSON object on stdout.",
269 )
270 parser.add_argument(
271 "--files",
272 "--tree",
273 dest="show_files",
274 action="store_true",
275 help="Append the managed files as a tree-style listing (human output).",
276 )
277 parser.add_argument(
278 "--check",
279 dest="check",
280 action="store_true",
281 help="Compare the pinned ref to the latest upstream release (needs network).",
282 )
283 args = parser.parse_args(argv)
284 return status(
285 Path(args.target),
286 json_output=args.json_output,
287 show_files=args.show_files,
288 check=args.check,
289 )
292if __name__ == "__main__":
293 raise SystemExit(main())