Coverage for vibefpmgui/dep

1"""Check git dependencies for newer versions (only when the user asks).

3For each git dependency the remote is asked for its tags and branches with

4``git ls-remote`` (nothing is downloaded), and compared with what the project

5requests (tag / branch / rev / nothing) and the commit fpm checked out

6(``build/cache.toml``). A dependency pinned to an older tag can be moved to

7the newest tag: only that tag value in the declaring ``fpm.toml`` is changed.

8"""

10from __future__ import annotations

12import os

13import re

14import subprocess

15import tomllib

16from concurrent.futures import ThreadPoolExecutor, as_completed

17from dataclasses import dataclass

18from pathlib import Path

19from typing import ClassVar

21from qtpy.QtCore import QObject, Qt, QThread, QUrl, Signal, Slot

22from qtpy.QtGui import QColor, QDesktopServices

23from qtpy.QtWidgets import (

24 QAbstractItemView,

25 QDialog,

26 QHBoxLayout,

27 QHeaderView,

28 QLabel,

29 QPushButton,

30 QTableWidget,

31 QTableWidgetItem,

32 QVBoxLayout,

33 QWidget,

34)

36from . import manifest as m

37from .console import NO_WINDOW

40LS_REMOTE_TIMEOUT = 30 # seconds per repository

42# Status kinds (also used for colors)

43UP_TO_DATE, UPDATE, NEW_COMMITS, INFO, ERROR = "up to date", "update", "new commits", "info", "error"

44TAG_AVAILABLE = "tag available" # pinned to a commit; a tag could be used instead

45STATUS_COLORS = {UP_TO_DATE: "#2f9e44", UPDATE: "#e8590c", NEW_COMMITS: "#1c7ed6",

46 TAG_AVAILABLE: "#7048e8", INFO: "#868e96", ERROR: "#e03131"}

49@dataclass

50class DepCheck:

51 name: str

52 git: str

53 ref_kind: str = "" # "tag", "branch", "rev" or "" (default branch)

54 ref: str = ""

55 current: str = "" # commit fpm checked out (build/cache.toml)

56 declared_in: Path | None = None

57 editable: bool = False # the declaring fpm.toml is the user's (root / path dependency)

58 latest_tag: str = ""

59 latest_tag_commit: str = ""

60 branch_commit: str = "" # newest commit of the requested (or default) branch

61 status: str = ""

62 message: str = ""

64 @property

65 def requested(self) -> str:

66 if self.ref_kind == "rev": 66 ↛ 67line 66 didn't jump to line 67 because the condition on line 66 was never true

67 return f"rev {self.ref[:7]}"

68 return f"{self.ref_kind} {self.ref}" if self.ref_kind else "default branch"

71# ------------------------------------------------------------------ collect

72def git_dependencies(graph: m.DependencyGraph) -> list[DepCheck]:

73 """Every git dependency, with how it is requested and where."""

74 checks = []

75 for pkg in graph.packages.values():

76 if not pkg.git or pkg.kind in (m.ROOT, m.DEV_ONLY): # (a dependency's dev-dependency isn't fetched)

77 continue

78 check = DepCheck(pkg.name, pkg.git, current=pkg.git_rev)

79 # Prefer a declaration in a manifest the user can edit.

80 parents = [graph.packages[e.parent] for e in graph.parents(pkg.name) if e.parent in graph.packages]

81 parents.sort(key=lambda p: p.kind not in (m.ROOT, m.PATH))

82 for parent in parents: 82 ↛ 93line 82 didn't jump to line 93 because the loop on line 82 didn't complete

83 spec = _declared_spec(parent.manifest, pkg.name) if parent.manifest else None

84 if spec is not None: 84 ↛ 82line 84 didn't jump to line 82 because the condition on line 84 was always true

85 check.declared_in = parent.manifest

86 check.editable = parent.kind in (m.ROOT, m.PATH)

87 check.git = str(spec.get("git", check.git))

88 for key in ("tag", "branch", "rev"): 88 ↛ 92line 88 didn't jump to line 92 because the loop on line 88 didn't complete

89 if key in spec: 89 ↛ 88line 89 didn't jump to line 88 because the condition on line 89 was always true

90 check.ref_kind, check.ref = key, str(spec[key])

91 break

92 break

93 checks.append(check)

94 return sorted(checks, key=lambda c: (not c.editable, c.name.lower()))

97def _declared_spec(manifest: Path, name: str) -> dict | None:

98 try:

99 data = m.load_toml(manifest)

100 except (OSError, tomllib.TOMLDecodeError):

101 return None

102 for section in ("dependencies", "dev-dependencies"): 102 ↛ 106line 102 didn't jump to line 106 because the loop on line 102 didn't complete

103 spec = (data.get(section) or {}).get(name)

104 if isinstance(spec, dict): 104 ↛ 102line 104 didn't jump to line 102 because the condition on line 104 was always true

105 return spec

106 return None

107

108

109# -------------------------------------------------------------------- check

110ALLOWED_GIT_PROTOCOLS = "https:http:ssh:git:file"

111

112

113def git_environment(environ=os.environ) -> dict[str, str]:

114 """The environment for git commands that must never wait for a prompt:

115 no password questions (HTTPS), and SSH in batch mode (no host-key or

116 passphrase questions: a key in the agent still works; otherwise it fails

117 at once instead of waiting out the timeout). A GIT_SSH_COMMAND or

118 GIT_SSH of the user's own is left as it is.

119

120 Only real transports are allowed: a URL comes from a project's fpm.toml,

121 which may not be trustworthy, and git's "ext::" transport (and other

122 remote helpers) would run the command in the URL."""

123 env = dict(environ, GIT_TERMINAL_PROMPT="0", GIT_ALLOW_PROTOCOL=ALLOWED_GIT_PROTOCOLS)

124 if not env.get("GIT_SSH_COMMAND") and not env.get("GIT_SSH"):

125 env["GIT_SSH_COMMAND"] = "ssh -o BatchMode=yes"

126 return env

127

128

129def git_error(stderr: str) -> str:

130 """What went wrong, from git's message: its first line (the cause, e.g.

131 "Permission denied (publickey)") and its "fatal:" one, with a hint when

132 SSH couldn't ask for a key or a host key (see git_environment)."""

133 lines = [line.strip() for line in stderr.splitlines() if line.strip()]

134 if not lines:

135 return ""

136 fatal = next((line.removeprefix("fatal: ") for line in lines if line.startswith("fatal:")), "")

137 first = lines[0].removeprefix("fatal: ")

138 text = first if not fatal or fatal == first else f"{first} — {fatal}"

139 if "publickey" in stderr or "Host key verification failed" in stderr:

140 text += (" (SSH can't ask for a key or a host's fingerprint here: load the key into ssh-agent "

141 "and connect once in a terminal, or use the repository's https:// URL)")

142 return text

143

144

145def ls_remote(url: str) -> dict[str, str]:

146 """{ref: commit} for HEAD, branches and tags (annotated tags resolved

147 to their commit). Raises RuntimeError with git's message on failure."""

148 env = git_environment()

149 try:

150 # ("--": a URL from fpm.toml that starts with "-", such as

151 # "--upload-pack=<command>", is a URL to git, never an option.)

152 out = subprocess.run(["git", "ls-remote", "--", url, "HEAD", "refs/heads/*", "refs/tags/*"],

153 capture_output=True, text=True, timeout=LS_REMOTE_TIMEOUT, env=env, check=False,

154 **NO_WINDOW)

155 except FileNotFoundError:

156 raise RuntimeError("git was not found on PATH")

157 except subprocess.TimeoutExpired:

158 raise RuntimeError(f"no answer from the server within {LS_REMOTE_TIMEOUT} s")

159 if out.returncode != 0:

160 raise RuntimeError(git_error(out.stderr or "") or f"git ls-remote failed ({out.returncode})")

161 refs: dict[str, str] = {}

162 for line in out.stdout.splitlines():

163 sha, _, ref = line.partition("\t")

164 if ref.endswith("^{}"):

165 refs[ref[:-3]] = sha # the commit an annotated tag points to

166 else:

167 refs.setdefault(ref, sha)

168 return refs

169

170

171def version_key(tag: str) -> tuple:

172 """Sort key for tags like v1.2.3 / 1.2 / release-2.0.1."""

173 numbers = tuple(int(n) for n in re.findall(r"\d+", tag))

174 return (numbers, not PRERELEASE_RE.search(tag), tag)

175

176

177def newest_tag(tags: list[str], allow_prerelease: bool = False) -> str:

178 candidates = [t for t in tags if re.search(r"\d", t)

179 and (allow_prerelease or not PRERELEASE_RE.search(t))]

180 return max(candidates, key=version_key) if candidates else ""

181

182

183def check_dependency(check: DepCheck) -> DepCheck:

184 try:

185 refs = ls_remote(check.git)

186 except RuntimeError as exc:

187 check.status, check.message = ERROR, f"Could not reach the repository: {exc}"

188 return check

189 tags = {ref[len("refs/tags/"):]: sha for ref, sha in refs.items() if ref.startswith("refs/tags/")}

190 allow_pre = check.ref_kind == "tag" and bool(PRERELEASE_RE.search(check.ref))

191 check.latest_tag = newest_tag(list(tags), allow_pre)

192 check.latest_tag_commit = tags.get(check.latest_tag, "")

193 if check.ref_kind == "branch":

194 check.branch_commit = refs.get(f"refs/heads/{check.ref}", "")

195 else:

196 check.branch_commit = refs.get("HEAD", "")

197 current = check.current

198

199 if check.ref_kind == "tag":

200 if check.ref not in tags:

201 check.status, check.message = ERROR, f"Tag {check.ref} no longer exists in the repository"

202 elif check.latest_tag and version_key(check.latest_tag) > version_key(check.ref):

203 check.status = UPDATE

204 check.message = f"Newer tag available: {check.ref} → {check.latest_tag}"

205 else:

206 check.status, check.message = UP_TO_DATE, f"{check.ref} is the newest tag"

207 elif check.ref_kind == "rev":

208 pinned = check.ref

209 if not check.latest_tag:

210 check.status, check.message = INFO, "Pinned to a commit (the repository has no tags)"

211 elif check.latest_tag_commit and (check.latest_tag_commit.startswith(pinned)

212 or pinned.startswith(check.latest_tag_commit)):

213 check.status = TAG_AVAILABLE

214 check.message = f"Pinned to commit {pinned[:7]}, which is tag {check.latest_tag}"

215 else:

216 check.status = TAG_AVAILABLE

217 check.message = (f"Pinned to a commit; the newest tag is {check.latest_tag} "

218 "(Compare shows whether it is newer)")

219 else: # a branch, or the default branch

220 where = f"branch {check.ref}" if check.ref_kind == "branch" else "the default branch"

221 if not check.branch_commit:

222 check.status, check.message = ERROR, f"Could not find {where} in the repository"

223 elif not current:

224 check.status, check.message = INFO, "Not downloaded yet — Update deps fetches it"

225 elif check.branch_commit.startswith(current) or current.startswith(check.branch_commit):

226 check.status, check.message = UP_TO_DATE, f"Up to date with {where}"

227 else:

228 check.status = NEW_COMMITS

229 check.message = f"{where.capitalize()} has new commits — Update deps fetches them"

230 return check

231

232

233# --------------------------------------------------------------- edit tags

234def set_ref(manifest: Path, name: str, old_key: str, old: str, new_key: str, new: str) -> bool:

235 """Change ``old_key = "old"`` to ``new_key = "new"`` (e.g. a ``rev`` pin

236 to a ``tag``) for one dependency, leaving everything else in the file

237 (comments, layout) untouched. Returns False if the declaration couldn't

238 be found unambiguously."""

239 text = manifest.read_text()

240 tag_re = re.compile(r"(\b)" + re.escape(old_key) + r"""(\s*=\s*)(["'])""" + re.escape(old) + r"""\3""")

241 lines = text.split("\n")

242 target = None

243 # 1. Inline table: name = { git = "...", tag = "old" }

244 key = re.compile(r"""^\s*(?:["']?)""" + re.escape(name) + r"""(?:["']?)\s*=\s*\{""")

245 for i, line in enumerate(lines):

246 if key.match(line) and tag_re.search(line):

247 target = i

248 break

249 # 2. Table form: [dependencies.name] ... tag = "old"

250 if target is None:

251 header = re.compile(r"^\s*\[\s*(?:dev-)?dependencies\.(?:[\"']?)" + re.escape(name)

252 + r"(?:[\"']?)\s*\]")

253 for i, line in enumerate(lines):

254 if header.match(line):

255 for j in range(i + 1, len(lines)): 255 ↛ 261line 255 didn't jump to line 261 because the loop on line 255 didn't complete

256 if lines[j].lstrip().startswith("["): 256 ↛ 257line 256 didn't jump to line 257 because the condition on line 256 was never true

257 break

258 if tag_re.search(lines[j]):

259 target = j

260 break

261 break

262 if target is None:

263 return False

264 lines[target] = tag_re.sub(lambda mt: f"{mt.group(1)}{new_key}{mt.group(2)}{mt.group(3)}{new}{mt.group(3)}",

265 lines[target], count=1)

266 updated = "\n".join(lines)

267 # Make sure the result parses and says what we meant.

268 try:

269 data = tomllib.loads(updated)

270 except tomllib.TOMLDecodeError:

271 return False

272 spec = (data.get("dependencies") or {}).get(name) or (data.get("dev-dependencies") or {}).get(name)

273 if not isinstance(spec, dict) or spec.get(new_key) != new or (old_key != new_key and old_key in spec): 273 ↛ 274line 273 didn't jump to line 274 because the condition on line 273 was never true

274 return False

275 manifest.write_text(updated)

276 return True

277

278

279def read_cache_revs(project_dir: Path) -> dict[str, str]:

280 """{dependency: commit} for the downloaded git dependencies, from each

281 checkout itself (build/cache.toml's "rev" isn't updated by fpm update)."""

282 revs: dict[str, str] = {}

283 folder = project_dir / "build" / "dependencies"

284 if folder.is_dir(): 284 ↛ 289line 284 didn't jump to line 289 because the condition on line 284 was always true

285 for dep in folder.iterdir():

286 commit = m.checked_out_commit(dep)

287 if commit: 287 ↛ 285line 287 didn't jump to line 285 because the condition on line 287 was always true

288 revs[dep.name] = commit

289 return revs

290

291

292# ------------------------------------------------------------------ worker

293class CheckWorker(QObject):

294 result = Signal(int, object) # row, DepCheck

295 finished = Signal()

296

297 def __init__(self, checks: list[DepCheck]):

298 super().__init__()

299 self.checks = checks

300

301 @Slot()

302 def run(self):

303 try:

304 with ThreadPoolExecutor(max_workers=6) as pool:

305 futures = {pool.submit(check_dependency, c): i for i, c in enumerate(self.checks)}

306 for future in as_completed(futures):

307 row = futures[future]

308 try:

309 check = future.result()

310 except Exception as exc: # noqa: BLE001 - that row's error, not the end of the check

311 check = self.checks[row]

312 check.status, check.message = ERROR, f"Could not be checked: {exc}"

313 self.result.emit(row, check)

314 finally:

315 self.finished.emit() # (always: it ends the thread)

316

317

318# ------------------------------------------------------------------ dialog

319class UpdatesDialog(QDialog):

320 """Results of a dependency check, with actions per row."""

321

322 update_tag_requested = Signal(object) # DepCheck

323 update_deps_requested = Signal()

324

325 COLUMNS: ClassVar[list[str]] = ["Dependency", "Requested", "Current", "Newest", "Status", ""]

326

327 def __init__(self, checks: list[DepCheck], compare_url, parent=None):

328 """``compare_url(git, old, new)``: a web page comparing two commits (or None)."""

329 super().__init__(parent)

330 self.setWindowTitle("Dependency updates")

331 self.checks = checks

332 self._compare_url = compare_url

333 self.summary = QLabel()

334 self.summary.setWordWrap(True)

335 note = QLabel("Asks each repository for its tags and branches "

336 "(<code>git ls-remote</code>); nothing is downloaded and no code is sent.")

337 note.setWordWrap(True)

338 self.table = QTableWidget(len(checks), len(self.COLUMNS))

339 self.table.setHorizontalHeaderLabels(self.COLUMNS)

340 self.table.verticalHeader().setVisible(False)

341 self.table.setEditTriggers(QAbstractItemView.NoEditTriggers)

342 self.table.setSelectionMode(QAbstractItemView.NoSelection)

343 # Compact: one line per row (long text is elided; tooltips have it all),

344 # fixed row height, no grid, alternating shading.

345 self.table.setWordWrap(False)

346 self.table.setTextElideMode(Qt.ElideRight)

347 self.table.setShowGrid(False)

348 self.table.setAlternatingRowColors(True)

349 self.row_height = self.table.fontMetrics().height() + 12

350 self.table.verticalHeader().setSectionResizeMode(QHeaderView.Fixed)

351 self.table.verticalHeader().setDefaultSectionSize(self.row_height)

352 hh = self.table.horizontalHeader()

353 for col in range(len(self.COLUMNS) - 1):

354 hh.setSectionResizeMode(col, QHeaderView.ResizeToContents)

355 hh.setSectionResizeMode(4, QHeaderView.Stretch)

356 for row, c in enumerate(checks):

357 self._set_row(row, c, checking=True)

358 close = QPushButton("Close")

359 close.clicked.connect(self.accept)

360 self.update_all = QPushButton("Update deps")

361 self.update_all.setToolTip("Run fpm update (fetches new commits of branch dependencies)")

362 self.update_all.clicked.connect(self.update_deps_requested)

363 self.update_all.hide()

364 buttons = QHBoxLayout()

365 buttons.addWidget(self.update_all)

366 buttons.addStretch(1)

367 buttons.addWidget(close)

368 layout = QVBoxLayout(self)

369 layout.addWidget(self.summary)

370 layout.addWidget(note)

371 layout.addWidget(self.table, 1)

372 layout.addLayout(buttons)

373 header = self.table.horizontalHeader().sizeHint().height()

374 self.resize(1120, 150 + header + self.row_height * max(3, min(len(checks), 12)))

375 self._done = 0

376 self._update_summary()

377

378 def _set_row(self, row: int, c: DepCheck, checking: bool = False):

379 def cell(col, text, tip="", color=None):

380 item = QTableWidgetItem(text)

381 if tip:

382 item.setToolTip(tip)

383 if color:

384 item.setForeground(QColor(color))

385 self.table.setItem(row, col, item)

386

387 where = "" if c.declared_in is None else f"declared in {c.declared_in}"

388 cell(0, c.name + ("" if c.editable else " (indirect)"), f"{c.git}\n{where}")

389 cell(1, c.requested, f"{c.ref_kind} {c.ref}" if c.ref_kind else "the repository's default branch")

390 cell(2, c.current[:7] if c.current else "—", c.current)

391 if checking:

392 cell(3, "…")

393 cell(4, "checking…", color="#868e96")

394 return

395 newest = c.latest_tag or "—"

396 if c.ref_kind in ("branch", "") and c.branch_commit: 396 ↛ 397line 396 didn't jump to line 397 because the condition on line 396 was never true

397 newest = f"{c.branch_commit[:7]}" + (f" (tag {c.latest_tag})" if c.latest_tag else "")

398 cell(3, newest, c.latest_tag_commit or c.branch_commit)

399 cell(4, c.message, c.message, STATUS_COLORS.get(c.status))

400 actions = QWidget()

401 row_layout = QHBoxLayout(actions)

402 row_layout.setContentsMargins(4, 0, 4, 0)

403 row_layout.setSpacing(4)

404 new_commit = c.latest_tag_commit if c.status in (UPDATE, TAG_AVAILABLE) else c.branch_commit

405 compare = self._compare_url(c.git, c.current, new_commit) if c.current and new_commit else None

406 if c.status == UPDATE and c.editable: 406 ↛ 411line 406 didn't jump to line 411 because the condition on line 406 was always true

407 button = self._small_button(f"Update to {c.latest_tag}")

408 button.setToolTip(f"Change tag = \"{c.ref}\" to \"{c.latest_tag}\" in {c.declared_in}")

409 button.clicked.connect(lambda _=False, ch=c: self.update_tag_requested.emit(ch))

410 row_layout.addWidget(button)

411 elif c.status == TAG_AVAILABLE and c.editable:

412 same = c.latest_tag_commit and (c.latest_tag_commit.startswith(c.ref) or c.ref.startswith(c.latest_tag_commit))

413 button = self._small_button(f"{'Use' if same else 'Switch to'} tag {c.latest_tag}")

414 button.setToolTip(f"Replace rev = \"{c.ref[:12]}…\" with tag = \"{c.latest_tag}\" in {c.declared_in}"

415 + ("" if same else "\n(Compare first: the tag may be older or newer than the pinned commit)"))

416 button.clicked.connect(lambda _=False, ch=c: self.update_tag_requested.emit(ch))

417 row_layout.addWidget(button)

418 elif c.status in (UPDATE, TAG_AVAILABLE):

419 note = QLabel("declared by another dependency")

420 row_layout.addWidget(note)

421 if compare and c.status in (UPDATE, NEW_COMMITS) or ( 421 ↛ 423line 421 didn't jump to line 423 because the condition on line 421 was never true

422 compare and c.status == TAG_AVAILABLE and not c.latest_tag_commit.startswith(c.current[:7])):

423 link = self._small_button("Compare")

424 link.setToolTip(compare)

425 link.clicked.connect(lambda _=False, u=compare: QDesktopServices.openUrl(QUrl(u)))

426 row_layout.addWidget(link)

427 row_layout.addStretch(1) # buttons keep their natural width

428 self.table.setCellWidget(row, 5, actions)

429

430 def _small_button(self, text: str) -> QPushButton:

431 """A button that fits a compact row (macOS buttons are tall by default)."""

432 button = QPushButton(text)

433 button.setAttribute(Qt.WA_MacSmallSize)

434 button.setAttribute(Qt.WA_LayoutUsesWidgetRect)

435 button.setFixedHeight(self.row_height - 4)

436 return button

437

438 def show_result(self, row: int, check: DepCheck):

439 self.checks[row] = check

440 self._set_row(row, check)

441 self._done += 1

442 self.table.resizeColumnsToContents()

443 # Size the buttons column to its widest row (cell widgets aren't

444 # counted by resizeColumnsToContents); Status takes the rest.

445 widest = max((w.sizeHint().width() for w in

446 (self.table.cellWidget(r, 5) for r in range(self.table.rowCount())) if w), default=0)

447 hh = self.table.horizontalHeader()

448 hh.setSectionResizeMode(5, QHeaderView.Fixed)

449 self.table.setColumnWidth(5, widest + 8)

450 hh.setSectionResizeMode(4, QHeaderView.Stretch)

451 self._update_summary()

452

453 def _update_summary(self):

454 total = len(self.checks)

455 if not total:

456 self.summary.setText("This project has no git dependencies.")

457 return

458 if self._done < total:

459 self.summary.setText(f"Checking {total} git dependenc{'ies' if total != 1 else 'y'}…"

460 f" ({self._done} done)")

461 return

462 updates = sum(c.status == UPDATE for c in self.checks)

463 commits = sum(c.status == NEW_COMMITS for c in self.checks)

464 pinned = sum(c.status == TAG_AVAILABLE for c in self.checks)

465 errors = sum(c.status == ERROR for c in self.checks)

466 parts = []

467 if updates: 467 ↛ 469line 467 didn't jump to line 469 because the condition on line 467 was always true

468 parts.append(f"{updates} with a newer tag")

469 if commits: 469 ↛ 470line 469 didn't jump to line 470 because the condition on line 469 was never true

470 parts.append(f"{commits} with new commits")

471 if pinned: 471 ↛ 472line 471 didn't jump to line 472 because the condition on line 471 was never true

472 parts.append(f"{pinned} pinned to a commit with a tag available")

473 if errors: 473 ↛ 474line 473 didn't jump to line 474 because the condition on line 473 was never true

474 parts.append(f"{errors} could not be checked")

475 self.summary.setText("" + (", ".join(parts) if parts else "Everything is up to date") + "")

476 self.update_all.setVisible(commits > 0)

477

478

479def start_check(checks: list[DepCheck], dialog: UpdatesDialog, parent: QObject) -> QThread:

480 thread = QThread(parent)

481 worker = CheckWorker(checks)

482 worker.moveToThread(thread)

483 thread.started.connect(worker.run)

484 worker.result.connect(dialog.show_result)

485 worker.finished.connect(thread.quit)

486 thread.finished.connect(worker.deleteLater)

487 thread.finished.connect(thread.deleteLater)

488 thread._worker = worker # keep a reference

489 thread.start()

490 return thread

Coverage for vibefpmgui/dep_updates.py: 90%

361 statements