Coverage for plugin/scripts/_rhiza_lock.py: 100%

1"""Read and write `.rhiza/template.lock`, and remove files that left the template. 

2 

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. 

6 

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""" 

10 

11from __future__ import annotations 

12 

13import os 

14import sys 

15from pathlib import Path 

16from typing import Any 

17 

18sys.path.insert(0, str(Path(__file__).resolve().parent)) 

19from _rhiza_common import escapes_root, log # noqa: E402 

20from _rhiza_yaml import as_list # noqa: E402 

21 

22# Never removed by orphan cleanup: without it the repo stops being rhiza-managed. 

23_PROTECTED = frozenset({Path(".rhiza/template.yml")}) 

24from _rhiza_template import Template, is_excluded # noqa: E402 

25from _rhiza_yaml import dump_yaml, load_yaml # noqa: E402 

26 

27 

28def lock_path(target: Path, lock_file: Path | None) -> Path: 

29 """Return the lock-file path (explicit override or the default under .rhiza).""" 

30 return lock_file if lock_file is not None else target / ".rhiza" / "template.lock" 

31 

32 

33def previously_tracked(lock_path: Path) -> set[Path]: 

34 """Return the file set recorded in an existing lock's ``files`` field. 

35 

36 Entries that would resolve outside the target root are dropped, loudly. This is the 

37 first place a lock's paths are read, and the one where they are later **unlinked** — 

38 `clean_orphaned_files` joins each onto *target* and deletes it — so a `..` entry here 

39 would be a delete outside the repository rather than the read-only join `stage_synced` 

40 guards. Nothing upstream produces one today; the containment was implicit, exactly as 

41 `_PROTECTED` was before mutation testing found that any string would do. 

42 """ 

43 if not lock_path.exists(): 

44 return set() 

45 try: 

46 lock = load_yaml(lock_path) 

47 except (OSError, ValueError): 

48 return set() 

49 tracked: set[Path] = set() 

50 for entry in as_list(lock.get("files")): 

51 if escapes_root(str(entry)): 

52 log(f"Ignoring lock entry outside the repository: {entry}") 

53 continue 

54 tracked.add(Path(entry)) 

55 return tracked 

56 

57 

58def clean_orphaned_files( 

59 target: Path, template_files: list[Path], excludes: set[str], previously_tracked: set[Path] 

60) -> None: 

61 """Delete files tracked by the previous sync that the template no longer provides. 

62 

63 An excluded path is never an orphan. Matching goes through :func:`is_excluded` rather 

64 than comparing against *excludes* directly, so that a directory entry protects the 

65 files under it: `excludes` now holds the configured destination paths verbatim, and a 

66 set membership test would read `docs` and `docs/guide.md` as unrelated. Locks written 

67 before the exclusion fix list excluded files in `files:`, which makes them look like 

68 orphans on the next sync — this is what stops that from deleting them. 

69 """ 

70 orphaned = previously_tracked - set(template_files) - set(_PROTECTED) 

71 for rel in sorted(orphaned): 

72 if is_excluded(rel.as_posix(), excludes): 

73 continue 

74 full = target / rel 

75 if full.exists(): 

76 try: 

77 full.unlink() 

78 log(f"[DEL] {rel}") 

79 except OSError as exc: 

80 log(f"Failed to delete {rel}: {exc}") 

81 

82 

83def _lock_identity(lock: dict[str, Any]) -> tuple[Any, ...]: 

84 """Return the content-comparison key for a lock dict, excluding ``synced_at``.""" 

85 return ( 

86 str(lock.get("sha", "")), 

87 str(lock.get("repo", "")), 

88 str(lock.get("host", "")), 

89 str(lock.get("ref", "")), 

90 as_list(lock.get("include")), 

91 as_list(lock.get("exclude")), 

92 as_list(lock.get("templates")), 

93 as_list(lock.get("files")), 

94 str(lock.get("strategy", "")), 

95 ) 

96 

97 

98def build_lock(sha: str, template: Template, files: list[str], synced_at: str) -> dict[str, Any]: 

99 """Assemble the ordered lock dict (matching the CLI's field order) for serialisation.""" 

100 lock: dict[str, Any] = { 

101 "sha": sha, 

102 "repo": template.repository, 

103 "host": template.host, 

104 "ref": template.ref, 

105 "include": template.include, 

106 "exclude": template.exclude, 

107 "templates": template.templates, 

108 } 

109 if template.profiles: 

110 lock["profiles"] = template.profiles 

111 lock["files"] = files 

112 lock["synced_at"] = synced_at 

113 lock["strategy"] = "merge" 

114 return lock 

115 

116 

117def write_lock(target: Path, lock: dict[str, Any], lock_path: Path) -> None: 

118 """Write the lock atomically; filter ``files`` to on-disk paths and skip no-op rewrites.""" 

119 lock = dict(lock) 

120 lock["files"] = sorted(f for f in as_list(lock.get("files")) if (target / f).exists()) 

121 

122 if lock_path.exists(): 

123 try: 

124 existing = load_yaml(lock_path) 

125 except (OSError, ValueError): 

126 existing = None 

127 if existing is not None and _lock_identity(existing) == _lock_identity(lock): 

128 log(f"{lock_path.name} is already up to date — skipping write") 

129 return 

130 

131 lock_path.parent.mkdir(parents=True, exist_ok=True) 

132 tmp_path = Path(str(lock_path) + ".tmp") 

133 dump_yaml(lock, tmp_path) 

134 os.replace(tmp_path, lock_path) 

135 log(f"Updated {lock_path.name} -> {str(lock['sha'])[:12]}") 

136 

137 

138# --------------------------------------------------------------------------- 

139# Merge orchestration 

140# --------------------------------------------------------------------------- 

141 

142 

143def read_base_sha(lock_path: Path) -> str | None: 

144 """Return the previously-synced SHA from the lock, or ``None`` for a first sync.""" 

145 if not lock_path.exists(): 

146 return None 

147 try: 

148 return str(load_yaml(lock_path).get("sha") or "") or None 

149 except (OSError, ValueError): 

150 return None