Coverage for src/web_signals.py: 100%

1"""Cross-thread signalling for the embedded web UI.

3The web UI (``src.web``) runs in a daemon thread alongside the sync

4loop (``src.sync``). Two things need to flow between them:

61. **Force-sync sentinels** — when the user taps "Sync now" on the

7 dashboard, the web thread touches ``$CONFIG_DIR/.force-sync-<svc>``

8 and the sync loop deletes the sentinel + zeroes the countdown on

9 its next iteration.

112. **Last-sync state** — after each per-service sync run, the sync

12 loop writes a small JSON file the dashboard reads on every

13 refresh (last completion time, file counts, error count).

15Files are chosen over a shared module-level singleton so the same

16mechanism keeps working if a future refactor splits sync + web into

17two processes. They live in ``ICLOUD_DOCKER_CONFIG_DIR`` (default

18``/config``) — same place the keyring and session cookies live, so

19they're persisted across container recreations.

20"""

22__author__ = "Mandar Patil (mandarons@pm.me)"

24import json

25import os

26import time

27from typing import Any

29from src import get_logger

31LOGGER = get_logger()

34def _config_dir() -> str:

35 """Resolve the directory force-sync sentinels + state JSON live in.

37 Mirrors the ICLOUD_DOCKER_CONFIG_DIR / DEFAULT_COOKIE_DIRECTORY

38 setup: same logic the keyring redirect uses, so dev hosts without

39 ``/config`` still work via a tempdir. Reads

40 ``DEFAULT_COOKIE_DIRECTORY`` via ``sys.modules`` so the test

41 fixture's monkeypatch is honoured -- a ``from src import

42 DEFAULT_COOKIE_DIRECTORY`` at module top would bind the value once

43 at import time and miss the redirect.

44 """

45 import sys

47 # DEFAULT_COOKIE_DIRECTORY is "<config_dir>/session_data"; strip the

48 # trailing component to recover the config dir.

49 cookie_dir = sys.modules["src"].DEFAULT_COOKIE_DIRECTORY

50 return os.path.dirname(cookie_dir) or "/config"

53_VALID_SERVICES = ("drive", "photos")

56def _sentinel_path(service: str) -> str:

57 return os.path.join(_config_dir(), f".force-sync-{service}")

60def _state_path() -> str:

61 return os.path.join(_config_dir(), ".last-sync-state.json")

64def request_force_sync(service: str) -> bool:

65 """Touch the sentinel for ``service``.

67 Returns True on success, False on validation/IO failure. Idempotent —

68 re-tapping while a previous request is still queued is a no-op.

69 """

70 if service not in _VALID_SERVICES:

71 return False

72 path = _sentinel_path(service)

73 try:

74 os.makedirs(os.path.dirname(path), exist_ok=True)

75 with open(path, "w") as f:

76 f.write(str(time.time()))

77 return True

78 except OSError as e:

79 LOGGER.warning(f"web_signals: failed to write {path}: {e!s}")

80 return False

83def pending_force_syncs() -> list[str]:

84 """Return services currently queued for an immediate sync.

86 Used by the dashboard to render "Queued ✓" instead of "Sync now".

87 """

88 return [

89 service

90 for service in _VALID_SERVICES

91 if os.path.isfile(_sentinel_path(service))

92 ]

95def consume_force_sync(service: str) -> bool:

96 """Atomically check + delete the sentinel. Sync loop calls this on

97 each iteration; True means "user requested an immediate run."

99 ``os.unlink`` raises ``FileNotFoundError`` if another caller beat us

100 to it — treated as "no request" rather than an error.

101 """

102 if service not in _VALID_SERVICES:

103 return False

104 try:

105 os.unlink(_sentinel_path(service))

106 return True

107 except FileNotFoundError:

108 return False

109 except OSError as e:

110 LOGGER.warning(f"web_signals: failed to consume {service} sentinel: {e!s}")

111 return False

112

113

114def record_sync_completion(

115 service: str,

116 *,

117 files_downloaded: int | None = None,

118 files_skipped: int | None = None,

119 files_removed: int | None = None,

120 errors: int | None = None,

121 duration_seconds: float | None = None,

122) -> None:

123 """Persist per-service stats after a sync run completes.

124

125 All counters are optional — passing ``None`` leaves the previous

126 value alone. Writes atomically (temp + rename) so a partial write

127 can't corrupt the file.

128 """

129 if service not in _VALID_SERVICES:

130 return

131 state = _load_state()

132 entry = state.get(service, {})

133 entry["completed_at"] = time.time()

134 if files_downloaded is not None:

135 entry["files_downloaded"] = int(files_downloaded)

136 if files_skipped is not None:

137 entry["files_skipped"] = int(files_skipped)

138 if files_removed is not None:

139 entry["files_removed"] = int(files_removed)

140 if errors is not None:

141 entry["errors"] = int(errors)

142 if duration_seconds is not None:

143 entry["duration_seconds"] = float(duration_seconds)

144 state[service] = entry

145 _save_state(state)

146

147

148def get_sync_state(service: str) -> dict[str, Any]:

149 """Return the persisted last-sync state for ``service``.

150

151 Empty dict on missing/corrupt file — the dashboard renders absence

152 gracefully.

153 """

154 if service not in _VALID_SERVICES:

155 return {}

156 return _load_state().get(service, {})

157

158

159def _load_state() -> dict[str, dict[str, Any]]:

160 path = _state_path()

161 if not os.path.isfile(path):

162 return {}

163 try:

164 with open(path) as f:

165 data = json.load(f)

166 if isinstance(data, dict):

167 return data

168 except (OSError, json.JSONDecodeError) as e:

169 LOGGER.warning(f"web_signals: failed to load {path}: {e!s} — treating as empty")

170 return {}

171

172

173def _save_state(state: dict[str, dict[str, Any]]) -> None:

174 path = _state_path()

175 tmp = path + ".tmp"

176 try:

177 os.makedirs(os.path.dirname(path), exist_ok=True)

178 with open(tmp, "w") as f:

179 json.dump(state, f, indent=2)

180 os.rename(tmp, path)

181 except OSError as e:

182 LOGGER.warning(f"web_signals: failed to save {path}: {e!s}")

183 try:

184 os.unlink(tmp)

185 except OSError:

186 pass

187

188

189_TRUST_STATE_KEY = "_trust"

190

191

192def record_trust_state(

193 *,

194 expires_at_iso: str | None,

195 warned_for_expires_at: str | None = None,

196) -> None:

197 """Persist Apple trust-cookie expiry + whether we've warned for it.

198

199 Stored under a reserved ``_trust`` key in the same JSON file as

200 per-service sync state. ``warned_for_expires_at`` carries the iso

201 timestamp the most recent threshold-cross warning was fired for, so

202 a cookie refresh (new expires_at) automatically rearms warning

203 eligibility -- compare ``warned_for_expires_at`` against the live

204 ``expires_at_iso`` to decide whether to fire again.

205 """

206 state = _load_state()

207 entry = state.get(_TRUST_STATE_KEY, {})

208 entry["expires_at"] = expires_at_iso

209 entry["last_updated"] = time.time()

210 if warned_for_expires_at is not None:

211 entry["warned_for_expires_at"] = warned_for_expires_at

212 state[_TRUST_STATE_KEY] = entry

213 _save_state(state)

214

215

216def get_trust_state() -> dict[str, Any]:

217 """Return persisted trust state. Empty dict if never recorded."""

218 return _load_state().get(_TRUST_STATE_KEY, {})

219

220

221def format_relative_time(epoch_seconds: float, *, now: float | None = None) -> str:

222 """Human-friendly relative time for dashboard display.

223

224 "Just now" / "5 min ago" / "2 h ago" / "3 d ago". Avoids

225 "1 hour ago" pluralisation gymnastics by sticking to compact

226 unit suffixes.

227 """

228 if not epoch_seconds:

229 return ""

230 if now is None:

231 now = time.time()

232 delta = max(0, now - epoch_seconds)

233 if delta < 30:

234 return "Just now"

235 if delta < 120:

236 return f"{int(delta)} sec ago"

237 if delta < 3600:

238 return f"{int(delta // 60)} min ago"

239 if delta < 86400:

240 return f"{int(delta // 3600)} h ago"

241 return f"{int(delta // 86400)} d ago"