Coverage for 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 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 # 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 if not lock_path.exists(): 

36 return set() 

37 try: 

38 lock = load_yaml(lock_path) 

39 except (OSError, ValueError): 

40 return set() 

41 return {Path(f) for f in as_list(lock.get("files"))} 

42 

43 

44def clean_orphaned_files( 

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

46) -> None: 

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

48 orphaned = ( 

49 previously_tracked - set(template_files) - {Path(e) for e in excludes} - set(_PROTECTED) 

50 ) 

51 for rel in sorted(orphaned): 

52 full = target / rel 

53 if full.exists(): 

54 try: 

55 full.unlink() 

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

57 except OSError as exc: 

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

59 

60 

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

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

63 return ( 

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

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

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

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

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

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

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

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

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

73 ) 

74 

75 

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

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

78 lock: dict[str, Any] = { 

79 "sha": sha, 

80 "repo": template.repository, 

81 "host": template.host, 

82 "ref": template.ref, 

83 "include": template.include, 

84 "exclude": template.exclude, 

85 "templates": template.templates, 

86 } 

87 if template.profiles: 

88 lock["profiles"] = template.profiles 

89 lock["files"] = files 

90 lock["synced_at"] = synced_at 

91 lock["strategy"] = "merge" 

92 return lock 

93 

94 

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

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

97 lock = dict(lock) 

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

99 

100 if lock_path.exists(): 

101 try: 

102 existing = load_yaml(lock_path) 

103 except (OSError, ValueError): 

104 existing = None 

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

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

107 return 

108 

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

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

111 dump_yaml(lock, tmp_path) 

112 os.replace(tmp_path, lock_path) 

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

114 

115 

116# --------------------------------------------------------------------------- 

117# Merge orchestration 

118# --------------------------------------------------------------------------- 

119 

120 

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

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

123 if not lock_path.exists(): 

124 return None 

125 try: 

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

127 except (OSError, ValueError): 

128 return None