Coverage for src/lpip/cache.py: 85%

1"""Persistent Git catalog cache and auto-refresh TTL (ADR-0003, ADR-0012).""" 

2 

3from __future__ import annotations 

4 

5import hashlib 

6import json 

7import os 

8import shutil 

9import stat 

10from collections.abc import Mapping 

11from dataclasses import dataclass 

12from datetime import UTC, datetime, timedelta 

13from pathlib import Path 

14 

15from lpip.config import default_user_cache_dir 

16from lpip.gitutils import git_checkout, git_clone, git_fetch 

17from lpip.models import RefreshMode 

18 

19AUTO_REFRESH_TTL = timedelta(hours=24) 

20_METADATA_NAME = "metadata.json" 

21_LAST_SUCCESS_KEY = "last_success" 

22 

23 

24@dataclass(frozen=True) 

25class CatalogCachePaths: 

26 root: Path 

27 repo: Path 

28 metadata: Path 

29 

30 

31def catalog_identity_hash(url: str, revision: str | None) -> str: 

32 payload = f"{url}\n{revision or ''}".encode() 

33 return hashlib.sha256(payload).hexdigest() 

34 

35 

36def catalog_cache_paths( 

37 url: str, 

38 revision: str | None, 

39 *, 

40 cache_home: Path | None = None, 

41) -> CatalogCachePaths: 

42 home = cache_home if cache_home is not None else default_user_cache_dir() 

43 root = home / "catalogs" / catalog_identity_hash(url, revision) 

44 return CatalogCachePaths(root=root, repo=root / "repo", metadata=root / _METADATA_NAME) 

45 

46 

47def read_last_success(paths: CatalogCachePaths) -> datetime | None: 

48 if not paths.metadata.is_file(): 

49 return None 

50 try: 

51 data = json.loads(paths.metadata.read_text(encoding="utf-8")) 

52 except (OSError, json.JSONDecodeError): 

53 return None 

54 raw = data.get(_LAST_SUCCESS_KEY) 

55 if not isinstance(raw, str) or not raw: 55 ↛ 56line 55 didn't jump to line 56 because the condition on line 55 was never true

56 return None 

57 try: 

58 parsed = datetime.fromisoformat(raw) 

59 except ValueError: 

60 return None 

61 if parsed.tzinfo is None: 61 ↛ 62line 61 didn't jump to line 62 because the condition on line 61 was never true

62 parsed = parsed.replace(tzinfo=UTC) 

63 return parsed 

64 

65 

66def write_last_success(paths: CatalogCachePaths, when: datetime | None = None) -> None: 

67 """Record last successful clone/fetch. Call only after Git succeeds.""" 

68 stamp = when or datetime.now(UTC) 

69 if stamp.tzinfo is None: 69 ↛ 70line 69 didn't jump to line 70 because the condition on line 69 was never true

70 stamp = stamp.replace(tzinfo=UTC) 

71 payload: dict[str, object] = {} 

72 if paths.metadata.is_file(): 

73 try: 

74 loaded = json.loads(paths.metadata.read_text(encoding="utf-8")) 

75 if isinstance(loaded, dict): 75 ↛ 79line 75 didn't jump to line 79 because the condition on line 75 was always true

76 payload = loaded 

77 except (OSError, json.JSONDecodeError): 

78 payload = {} 

79 payload[_LAST_SUCCESS_KEY] = stamp.isoformat() 

80 paths.root.mkdir(parents=True, exist_ok=True) 

81 paths.metadata.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") 

82 

83 

84def needs_auto_refresh( 

85 paths: CatalogCachePaths, 

86 *, 

87 now: datetime | None = None, 

88) -> bool: 

89 """True if timestamp is missing or older than 24 hours (ADR-0012).""" 

90 last = read_last_success(paths) 

91 if last is None: 

92 return True 

93 clock = now or datetime.now(UTC) 

94 if clock.tzinfo is None: 94 ↛ 95line 94 didn't jump to line 95 because the condition on line 94 was never true

95 clock = clock.replace(tzinfo=UTC) 

96 return clock - last >= AUTO_REFRESH_TTL 

97 

98 

99def should_fetch( 

100 paths: CatalogCachePaths, 

101 refresh: RefreshMode, 

102 *, 

103 force: bool = False, 

104 now: datetime | None = None, 

105) -> bool: 

106 """Whether to fetch an existing checkout. Absent cache is always cloned.""" 

107 if force or refresh is RefreshMode.ALWAYS: 

108 return True 

109 if refresh is RefreshMode.NEVER: 

110 return False 

111 return needs_auto_refresh(paths, now=now) 

112 

113 

114def _repo_ready(repo: Path) -> bool: 

115 git_dir = repo / ".git" 

116 return git_dir.is_dir() or git_dir.is_file() 

117 

118 

119def ensure_catalog_checkout( 

120 url: str, 

121 revision: str | None, 

122 *, 

123 cache_home: Path | None = None, 

124 refresh: RefreshMode = RefreshMode.AUTO, 

125 force_refresh: bool = False, 

126 now: datetime | None = None, 

127 environ: Mapping[str, str] | None = None, 

128) -> CatalogCachePaths: 

129 """Clone if missing; fetch per refresh policy; stamp only on Git success.""" 

130 paths = catalog_cache_paths(url, revision, cache_home=cache_home) 

131 if not _repo_ready(paths.repo): 

132 git_clone(url, paths.repo, environ=environ) 

133 write_last_success(paths, now) 

134 if revision: 134 ↛ 136line 134 didn't jump to line 136 because the condition on line 134 was always true

135 git_checkout(paths.repo, revision, environ=environ) 

136 return paths 

137 

138 if should_fetch(paths, refresh, force=force_refresh, now=now): 

139 git_fetch(paths.repo, environ=environ) 

140 write_last_success(paths, now) 

141 if revision: 141 ↛ 143line 141 didn't jump to line 143 because the condition on line 141 was always true

142 git_checkout(paths.repo, revision, environ=environ) 

143 return paths 

144 

145 

146def _rmtree(path: Path) -> None: 

147 def _onerror(func: object, name: str, _exc: object) -> None: 

148 os.chmod(name, stat.S_IWRITE) 

149 func(name) # type: ignore[operator] 

150 

151 shutil.rmtree(path, onerror=_onerror) 

152 

153 

154def purge_catalog_cache( 

155 url: str, 

156 revision: str | None, 

157 *, 

158 cache_home: Path | None = None, 

159) -> CatalogCachePaths: 

160 """Remove the cache directory for this catalog identity.""" 

161 paths = catalog_cache_paths(url, revision, cache_home=cache_home) 

162 if paths.root.exists(): 162 ↛ 164line 162 didn't jump to line 164 because the condition on line 162 was always true

163 _rmtree(paths.root) 

164 return paths