Coverage for scripts/_rhiza_snapshot.py: 100%

1"""Materialise a template ref as a flat snapshot directory. 

2 

3Clone the template (sparse, to the included paths only), then copy the included, 

4non-excluded files into a snapshot directory *at their destination paths*. The result is 

5a plain tree that mirrors what the target repo should contain. 

6 

7`_rhiza_merge.py` consumes two of these — one at the previously-synced ref, one at the 

8new one — which is what makes the merge a three-way merge over ordinary directories. 

9""" 

10 

11from __future__ import annotations 

12 

13import os 

14import shutil 

15import sys 

16import tempfile 

17from pathlib import Path 

18 

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

20import _rhiza_git as git # noqa: E402 

21from _rhiza_bundles import Bundles, resolve_bundle_names # noqa: E402 

22from _rhiza_common import log # noqa: E402 

23from _rhiza_template import Template # noqa: E402 

24from _rhiza_yaml import load_yaml # noqa: E402 

25 

26 

27def clone_template( 

28 ctx: git.GitContext, template: Template, branch: str 

29) -> tuple[Path, str, list[str], dict[str, str]]: 

30 """Clone the upstream template and resolve the include paths + path map. 

31 

32 Returns ``(upstream_dir, upstream_sha, include_paths, path_map)``; the caller 

33 owns *upstream_dir* and must remove it. 

34 """ 

35 rhiza_branch = template.ref or branch 

36 include_paths = list(template.include) 

37 upstream_dir = Path(tempfile.mkdtemp()) 

38 path_map: dict[str, str] = {} 

39 

40 if template.profiles or template.templates: 

41 git.clone(ctx, template.git_url, upstream_dir, [template.bundles_path], branch=rhiza_branch) 

42 bundles = Bundles.from_config(load_yaml(upstream_dir / template.bundles_path)) 

43 names = resolve_bundle_names(template, bundles) 

44 resolved = bundles.resolve_to_paths(names) 

45 path_map = bundles.resolve_to_path_map(names) 

46 include_paths = list(dict.fromkeys(resolved + include_paths)) 

47 git.update_sparse_checkout(ctx, upstream_dir, include_paths) 

48 else: 

49 git.clone(ctx, template.git_url, upstream_dir, include_paths, branch=rhiza_branch) 

50 

51 upstream_sha = git.get_head_sha(ctx, upstream_dir) 

52 log(f"Upstream HEAD: {upstream_sha[:12]}") 

53 return upstream_dir, upstream_sha, include_paths, path_map 

54 

55 

56def _expand_paths(base_dir: Path, paths: list[str]) -> list[Path]: 

57 """Expand file/directory *paths* under *base_dir* into a flat list of files.""" 

58 all_files: list[Path] = [] 

59 for rel in paths: 

60 full = base_dir / rel 

61 if full.is_file(): 

62 all_files.append(full) 

63 elif full.is_dir(): 

64 all_files.extend( 

65 Path(dirpath) / fname 

66 for dirpath, _, filenames in os.walk(full, followlinks=True) 

67 for fname in filenames 

68 ) 

69 return all_files 

70 

71 

72def excluded_set(base_dir: Path, excluded_paths: list[str]) -> set[str]: 

73 """Return excluded relative-path strings for *base_dir* (always includes template.yml).""" 

74 result = {str(f.relative_to(base_dir)) for f in _expand_paths(base_dir, excluded_paths)} 

75 result.add(".rhiza/template.yml") 

76 return result 

77 

78 

79def _remap_path(source: str, path_map: dict[str, str]) -> str: 

80 """Translate *source* to its destination via *path_map* (exact or directory-prefix).""" 

81 if source in path_map: 

82 return path_map[source] 

83 for src, dest in path_map.items(): 

84 src_prefix = src.rstrip("/") + "/" 

85 if source.startswith(src_prefix): 

86 suffix = source[len(src_prefix) :] 

87 return dest.rstrip("/") + "/" + suffix if dest.rstrip("/") else suffix 

88 return source 

89 

90 

91def prepare_snapshot( 

92 clone_dir: Path, 

93 include_paths: list[str], 

94 excludes: set[str], 

95 snapshot_dir: Path, 

96 path_map: dict[str, str], 

97) -> list[Path]: 

98 """Copy included, non-excluded files from *clone_dir* into *snapshot_dir* at dest paths.""" 

99 template_files: list[Path] = [] 

100 for f in _expand_paths(clone_dir, include_paths): 

101 rel_source = str(f.relative_to(clone_dir)) 

102 if rel_source in excludes: 

103 continue 

104 rel_dest = _remap_path(rel_source, path_map) 

105 dst = snapshot_dir / rel_dest 

106 dst.parent.mkdir(parents=True, exist_ok=True) 

107 shutil.copy2(f, dst) 

108 template_files.append(Path(rel_dest)) 

109 return template_files 

110 

111 

112def copy_files(snapshot_dir: Path, target: Path, files: list[Path]) -> None: 

113 """Copy each of *files* from *snapshot_dir* into *target*, creating parents.""" 

114 for rel in sorted(files): 

115 dst = target / rel 

116 dst.parent.mkdir(parents=True, exist_ok=True) 

117 shutil.copy2(snapshot_dir / rel, dst) 

118 

119 

120# --------------------------------------------------------------------------- 

121# Lock file + orphan cleanup 

122# ---------------------------------------------------------------------------