Coverage for plugin/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, is_excluded # 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 _remap_path(source: str, path_map: dict[str, str]) -> str: 

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

74 if source in path_map: 

75 return path_map[source] 

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

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

78 if source.startswith(src_prefix): 

79 suffix = source[len(src_prefix) :] 

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

81 return source 

82 

83 

84def prepare_snapshot( 

85 clone_dir: Path, 

86 include_paths: list[str], 

87 excludes: set[str], 

88 snapshot_dir: Path, 

89 path_map: dict[str, str], 

90) -> list[Path]: 

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

92 

93 The remap happens **before** the exclusion test, and the order is the whole point: 

94 `exclude:` is declared in destination paths, so testing the source path meant a 

95 bundle-sourced file was never matched. `bundles/python-core/.pre-commit-config.yaml` 

96 is not `.pre-commit-config.yaml`, so the file was copied, listed in the lock, and 

97 staged into the PR by `stage_synced.py` — all against an explicit exclusion. 

98 """ 

99 template_files: list[Path] = [] 

100 for f in _expand_paths(clone_dir, include_paths): 

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

102 rel_dest = _remap_path(rel_source, path_map) 

103 if is_excluded(rel_dest, excludes): 

104 continue 

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