Port to ostree API and drop CLI usage

flatpak_builder_lint/checks/flathub_json.py

@@ -108,13 +108,12 @@ def check_repo(self, path: str) -> None:
         if not ref:
             return
 
-        flathub_json = ostree.get_flathub_json(path, ref)
-        if not flathub_json:
-            return
-
         with tempfile.TemporaryDirectory() as tmpdir:
             ostree.extract_subpath(path, ref, "/metadata", tmpdir)
             metadata = builddir.parse_metadata(tmpdir)
             if not metadata:
                 return
+            flathub_json = ostree.get_flathub_json(path, ref, tmpdir)
+            if not flathub_json:
+                return
             self._check_metadata(metadata, flathub_json)

flatpak_builder_lint/checks/screenshots.py

@@ -1,3 +1,4 @@
+import glob
 import os
 import tempfile
 
@@ -16,10 +17,7 @@ def check_repo(self, path: str) -> None:
         if appid.endswith(".BaseApp"):
             return
 
-        refs_cmd = ostree.cli(path, "refs", "--list")
-        if refs_cmd["returncode"] != 0:
-            raise RuntimeError("Failed to list refs")
-        refs = refs_cmd["stdout"].splitlines()
+        refs = ostree.get_refs(path, None)
 
         with tempfile.TemporaryDirectory() as tmpdir:
             ostree.extract_subpath(path, ref, "files/share", tmpdir)
@@ -72,14 +70,14 @@ def check_repo(self, path: str) -> None:
                 "https://dl.flathub.org/media",
             )
 
-            screenshots = appstream.components(appstream_path)[0].xpath(
-                "screenshots/screenshot/image"
-            )
-
-            sc_values = list(
-                appstream.components(appstream_path)[0].xpath(
-                    "screenshots/screenshot/image/text()"
-                )
+            sc_values = set(
+                [
+                    os.path.basename(i)
+                    for i in appstream.components(appstream_path)[0].xpath(
+                        "screenshots/screenshot/image/text()"
+                    )
+                    if i.endswith(".png")
+                ]
             )
 
             if not sc_values:
@@ -102,40 +100,17 @@ def check_repo(self, path: str) -> None:
             for arch in arches:
                 if f"screenshots/{arch}" not in refs:
                     self.errors.add("appstream-screenshots-not-mirrored-in-ostree")
-                    return
-
-                ostree_screenshots_cmd = ostree.cli(
-                    path, "ls", "-R", f"screenshots/{arch}"
+                media_path = os.path.join(tmpdir, "app-info", f"screenshots-{arch}")
+                media_glob_path = f"{media_path}/**"
+                ostree.extract_subpath(path, f"screenshots/{arch}", "/", media_path)
+
+                ref_sc_files = set(
+                    [
+                        os.path.basename(path)
+                        for path in glob.glob(media_glob_path, recursive=True)
+                        if path.endswith(".png")
+                    ]
                 )
-                if ostree_screenshots_cmd["returncode"] != 0:
-                    raise RuntimeError(
-                        "Failed to list screenshot refs: {}".format(
-                            ostree_screenshots_cmd["stderr"].strip()
-                        )
-                    )
 
-                ostree_screenshots = []
-                for ostree_screenshot in ostree_screenshots_cmd["stdout"].splitlines():
-                    (
-                        mode,
-                        _,
-                        _,
-                        _,
-                        ostree_screenshot_filename,
-                    ) = ostree_screenshot.split()
-                    if mode[0] != "-":
-                        continue
-                    ostree_screenshots.append(ostree_screenshot_filename[1:])
-
-                for screenshot in screenshots:
-                    if screenshot.attrib.get("type") == "thumbnail":
-                        if screenshot.text.startswith("https://dl.flathub.org/media/"):
-                            screenshot_fn = "/".join(screenshot.text.split("/")[4:])
-                        else:
-                            screenshot_fn = "/".join(screenshot.text.split("/")[5:])
-
-                        if f"{screenshot_fn}" not in ostree_screenshots:
-                            self.warnings.add(
-                                "appstream-screenshots-files-not-found-in-ostree"
-                            )
-                            return
+                if not (ref_sc_files & sc_values):
+                    self.errors.add("appstream-screenshots-files-not-found-in-ostree")

flatpak_builder_lint/ostree.py

@@ -1,40 +1,32 @@
-import errno
 import json
 import os
-import subprocess
-from typing import Optional, TypedDict
+from typing import List, Optional
 
-from .builddir import parse_metadata
+import gi
 
+gi.require_version("OSTree", "1.0")
+from gi.repository import Gio, GLib, OSTree  # noqa: E402
 
-class CliResult(TypedDict):
-    stdout: str
-    stderr: str
-    returncode: int
 
+def open_ostree_repo(repo_path: str) -> OSTree.Repo:
+    repo = OSTree.Repo.new(Gio.File.new_for_path(repo_path))
 
-def cli(repo: str, *args: str) -> CliResult:
-    if not os.path.exists(repo):
-        raise OSError(errno.ENOENT, f"No such ostree repo: {repo}")
+    if not repo.open(None):
+        raise Exception("Failed to open the OSTree repository")
 
-    ret = subprocess.run(
-        ["ostree", f"--repo={repo}", *args],
-        capture_output=True,
-    )
+    return repo
 
-    return {
-        "stdout": ret.stdout.decode("utf-8"),
-        "stderr": ret.stderr.decode("utf-8"),
-        "returncode": ret.returncode,
-    }
 
+def get_refs(repo_path: str, ref_prefix: str | None) -> set[str]:
+    repo = open_ostree_repo(repo_path)
+    _, refs = repo.list_refs(ref_prefix, None)
 
-def get_primary_ref(repo: str) -> Optional[str]:
-    refs_cmd = cli(repo, "refs", "--list")
-    if refs_cmd["returncode"] != 0:
-        raise RuntimeError("Failed to list refs")
+    return set(refs.keys())
+
+
+def get_primary_ref(repo_path: str) -> Optional[str]:
+    refs = get_refs(repo_path, None)
 
-    refs = refs_cmd["stdout"].splitlines()
     ref: str
 
     for ref in refs:
@@ -44,30 +36,6 @@ def get_primary_ref(repo: str) -> Optional[str]:
     return None
 
 
-def get_text_file(repo: str, ref: str, path: str) -> Optional[str]:
-    cmd = cli(repo, "cat", ref, path)
-    if cmd["returncode"] == 0:
-        return cmd["stdout"]
-
-    return None
-
-
-def get_metadata(repo: str, primary_ref: Optional[str]) -> Optional[dict]:
-    if not primary_ref:
-        ref = get_primary_ref(repo)
-        if not ref:
-            return None
-    else:
-        ref = primary_ref
-
-    cat_metadata_cmd = cli(repo, "cat", ref, "/metadata")
-    if cat_metadata_cmd["returncode"] == 0:
-        metadata = parse_metadata(cat_metadata_cmd["stdout"])
-        return metadata
-
-    return None
-
-
 def infer_appid(path: str) -> Optional[str]:
     ref = get_primary_ref(path)
     if ref:
@@ -76,38 +44,47 @@ def infer_appid(path: str) -> Optional[str]:
     return None
 
 
-def get_flathub_json(repo: str, ref: str) -> Optional[dict]:
-    flathub_json_path = "/files/flathub.json"
-    flathub_json_raw = get_text_file(repo, ref, flathub_json_path)
-
-    if not flathub_json_raw:
-        return None
+def extract_subpath(
+    repo_path: str,
+    ref: str,
+    subpath: str,
+    dest: str,
+    should_pass: Optional[bool] = False,
+) -> None:
+    repo = open_ostree_repo(repo_path)
+    opts = OSTree.RepoCheckoutAtOptions()
+    # https://gitlab.gnome.org/GNOME/pygobject/-/issues/639
+    opts.mode = int(OSTree.RepoCheckoutMode.USER)  # type: ignore
+    opts.overwrite_mode = int(OSTree.RepoCheckoutOverwriteMode.ADD_FILES)  # type: ignore
+    opts.subpath = subpath
+
+    _, rev = repo.resolve_rev(ref, True)
+
+    # https://sourceware.org/git/?p=glibc.git;a=blob;f=io/fcntl.h;h=f157991782681caabe9bd7edb46ec205731965af;hb=HEAD#l149
+    AT_FDCWD = -100
+    if rev:
+        if should_pass:
+            try:
+                repo.checkout_at(opts, AT_FDCWD, dest, rev, None)
+            except GLib.Error as err:
+                if err.matches(Gio.io_error_quark(), Gio.IOErrorEnum.NOT_FOUND):
+                    pass
+                else:
+                    raise
+        else:
+            repo.checkout_at(opts, AT_FDCWD, dest, rev, None)
+
+
+def get_flathub_json(
+    repo_path: str, ref: str, dest: str
+) -> dict[str, str | bool | List[str]]:
+    extract_subpath(repo_path, ref, "/files/flathub.json", dest, True)
+
+    flathub_json_path = os.path.join(dest, "flathub.json")
+    flathub_json: dict = {}
+
+    if os.path.exists(flathub_json_path):
+        with open(flathub_json_path) as fp:
+            flathub_json = json.load(fp)
 
-    flathub_json: dict = json.loads(flathub_json_raw)
     return flathub_json
-
-
-def extract_subpath(repo: str, ref: str, subpath: str, dest: str) -> CliResult:
-    cmd = cli(
-        repo,
-        "checkout",
-        "--union-add",
-        "--user-mode",
-        f"--subpath={subpath}",
-        ref,
-        dest,
-    )
-
-    if cmd["returncode"] != 0:
-        raise RuntimeError(
-            "Failed to extract {} from ostree repo: {}".format(
-                subpath, cmd["stderr"].strip()
-            )
-        )
-
-    return cmd
-
-
-def list_ref(repo: str, ref: str) -> CliResult:
-    cmd = cli(repo, "ls", "--R", ref)
-    return cmd