vibefpmgui.dep_updates

  1"""Check git dependencies for newer versions (only when the user asks).
  2
  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"""
  9
 10from __future__ import annotations
 11
 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
 20
 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)
 35
 36from . import manifest as m
 37from .console import NO_WINDOW
 38
 39PRERELEASE_RE = re.compile(r"(alpha|beta|rc|dev|pre|preview|snapshot|nightly)", re.IGNORECASE)
 40LS_REMOTE_TIMEOUT = 30  # seconds per repository
 41
 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"}
 47
 48
 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 = ""
 63
 64    @property
 65    def requested(self) -> str:
 66        if self.ref_kind == "rev":
 67            return f"rev {self.ref[:7]}"
 68        return f"{self.ref_kind} {self.ref}" if self.ref_kind else "default branch"
 69
 70
 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:
 83            spec = _declared_spec(parent.manifest, pkg.name) if parent.manifest else None
 84            if spec is not None:
 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"):
 89                    if key in spec:
 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()))
 95
 96
 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"):
103        spec = (data.get(section) or {}).get(name)
104        if isinstance(spec, dict):
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)):
256                    if lines[j].lstrip().startswith("["):
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):
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():
285        for dep in folder.iterdir():
286            commit = m.checked_out_commit(dep)
287            if commit:
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("<span style='color:gray'>Asks each repository for its tags and branches "
336                      "(<code>git ls-remote</code>); nothing is downloaded and no code is sent.</span>")
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:
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:
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("<span style='color:gray'>declared by another dependency</span>")
420            row_layout.addWidget(note)
421        if compare and c.status in (UPDATE, NEW_COMMITS) or (
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("<b>This project has no git dependencies.</b>")
457            return
458        if self._done < total:
459            self.summary.setText(f"<b>Checking {total} git dependenc{'ies' if total != 1 else 'y'}…</b>"
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:
468            parts.append(f"{updates} with a newer tag")
469        if commits:
470            parts.append(f"{commits} with new commits")
471        if pinned:
472            parts.append(f"{pinned} pinned to a commit with a tag available")
473        if errors:
474            parts.append(f"{errors} could not be checked")
475        self.summary.setText("<b>" + (", ".join(parts) if parts else "Everything is up to date") + "</b>")
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