feat: add support for debian packages using tar.zst

Author: fyhertz

.github/workflows/cicd.yml

@@ -38,8 +38,8 @@ jobs:
       - name: Run ruff format
         run: poetry run ruff format --diff src tests
 
-      - name: Run ruff
-        run: poetry run ruff src tests
+      - name: Run ruff check
+        run: poetry run ruff check src tests
 
       - name: Run mypy
         run: poetry run mypy --show-error-codes src

.pre-commit-config.yaml

@@ -15,9 +15,9 @@ repos:
         language: system
         types: [python]
         pass_filenames: false
-      - id: ruff
-        name: ruff
-        entry: poetry run ruff src tests --fix
+      - id: ruff-check
+        name: ruff check
+        entry: poetry run ruff check src tests --fix
         language: system
         types: [python]
         pass_filenames: false

pyproject.toml

@@ -79,14 +79,16 @@ build-backend = "poetry.masonry.api"
 
 [tool.ruff]
 line-length = 90
+src = ["src", "tests"]
+
+[tool.ruff.lint]
 select = [
     "E",
     "F",
     "W",
     "I001",
 ]
-src = ["src", "tests"]
 
-[tool.ruff.per-file-ignores]
+[tool.ruff.lint.per-file-ignores]
 "__init__.py" = ["F401"]
 "src/ops2deb/templates.py" = ["E501", "W191"]

src/ops2deb/extracter.py

@@ -5,6 +5,7 @@
 import shutil
 import tarfile
 from pathlib import Path
+from typing import BinaryIO, Literal, Optional
 
 import unix_ar
 import zstandard
@@ -14,6 +15,47 @@
 from ops2deb.utils import log_and_raise
 
 
+# https://github.com/python/cpython/issues/81276
+class TarFile(tarfile.TarFile):
+    """Subclass of tarfile.TarFile that can read and write zstd compressed archives."""
+
+    OPEN_METH = {"zst": "zstopen"} | tarfile.TarFile.OPEN_METH  # type: ignore
+
+    @classmethod
+    def zstopen(
+        cls,
+        name: str,
+        mode: Literal["r", "w", "x"] = "r",
+        fileobj: Optional[BinaryIO] = None,
+    ) -> tarfile.TarFile:
+        if mode not in ("r", "w", "x"):
+            raise NotImplementedError(f"mode `{mode}' not implemented for zst")
+
+        if mode == "r":
+            zfobj = zstandard.open(fileobj or name, "rb")
+        else:
+            zfobj = zstandard.open(
+                fileobj or name,
+                mode + "b",
+                cctx=zstandard.ZstdCompressor(write_checksum=True, threads=-1),
+            )
+        try:
+            tarobj = cls.taropen(name, mode, zfobj)
+        except (OSError, EOFError, zstandard.ZstdError) as exc:
+            zfobj.close()
+            if mode == "r":
+                raise tarfile.ReadError("not a zst file") from exc
+            raise
+        except:
+            zfobj.close()
+            raise
+        # Setting the _extfileobj attribute is important to signal a need to
+        # close this object and thus flush the compressed stream.
+        # Unfortunately, tarfile.pyi doesn't know about it.
+        tarobj._extfileobj = False  # type: ignore
+        return tarobj
+
+
 def _unpack_gz(file_path: str, extract_path: str) -> None:
     output_path = Path(extract_path) / Path(file_path).stem
     with output_path.open("wb") as output:
@@ -35,7 +77,7 @@ def _unpack_deb(file_path: str, extract_path: str) -> None:
         if file_name.startswith("debian-binary"):
             continue
         tarball = ar_file.open(file_name)
-        tar_file = tarfile.open(fileobj=tarball)
+        tar_file = TarFile.open(fileobj=tarball)
         try:
             tar_file.extractall(Path(extract_path) / file_name.split(".")[0])
         finally:

src/ops2deb/updater.py

@@ -180,8 +180,7 @@ async def _find_latest_version(
             if version in blueprint.versions():
                 return None
             logger.info(
-                f"{blueprint.name} can be bumped "
-                f"from {blueprint.version} to {version}"
+                f"{blueprint.name} can be bumped from {blueprint.version} to {version}"
             )
             return LatestRelease(blueprint, version)
         except Ops2debUpdaterError as e:

tests/conftest.py

@@ -128,7 +128,7 @@ def mock_httpx_client():
 
     def async_client_mock(**kwargs):
         kwargs.pop("transport", None)
-        return real_async_client(app=app, **kwargs)
+        return real_async_client(transport=httpx.ASGITransport(app=app), **kwargs)
 
     httpx.AsyncClient = async_client_mock
     yield

tests/test_ops2deb.py

@@ -177,8 +177,7 @@ def test_generate__should_fail_gracefully_when_server_returns_a_404(call_ops2deb
 
     # Then
     expected_error = (
-        "Failed to download http://testserver/1.0.0/404.zip. "
-        "Server responded with 404."
+        "Failed to download http://testserver/1.0.0/404.zip. Server responded with 404."
     )
     assert expected_error in result.stderr
     assert result.exit_code == 77

tests/test_updater.py

@@ -93,7 +93,7 @@ async def test_generic_update_strategy_finds_latest_release_version(
     app = app_factory(versions)
 
     # When
-    async with AsyncClient(app=app) as client:
+    async with AsyncClient(transport=httpx.ASGITransport(app=app)) as client:
         update_strategy = GenericUpdateStrategy(client)
         latest_version = await update_strategy(blueprint)
 
@@ -112,7 +112,7 @@ async def test_generic_update_strategy_finds_latest_release_version_when_version
     app = app_factory(["2.0.0"])
 
     # When
-    async with AsyncClient(app=app) as client:
+    async with AsyncClient(transport=httpx.ASGITransport(app=app)) as client:
         update_strategy = GenericUpdateStrategy(client)
         latest_version = await update_strategy(blueprint)
 
@@ -133,7 +133,7 @@ async def test_github_update_strategy_should_find_expected_blueprint_release(
 ):
     app = github_app_factory(tag_name)
     blueprint = blueprint_factory(fetch=fetch_url)
-    async with AsyncClient(app=app) as client:
+    async with AsyncClient(transport=httpx.ASGITransport(app=app)) as client:
         update_strategy = GithubUpdateStrategy(client)
         assert await update_strategy(blueprint) == "2.3.0"
 
@@ -144,7 +144,7 @@ async def test_github_update_strategy_should_not_return_an_older_version_than_cu
     app = github_app_factory("0.1.0", versions=["1.0.0"])
     url = "https://github.com/owner/name/releases/{{version}}/some-app.tar.gz"
     blueprint = blueprint_factory(fetch=url)
-    async with AsyncClient(app=app) as client:
+    async with AsyncClient(transport=httpx.ASGITransport(app=app)) as client:
         update_strategy = GithubUpdateStrategy(client)
         assert await update_strategy(blueprint) == "1.0.0"
 
@@ -155,7 +155,7 @@ async def test_github_update_strategy_should_fail_gracefully_when_asset_not_foun
     app = github_app_factory("someapp-v2.3")
     url = "https://github.com/owner/name/releases/someapp-v{{version}}/some-app.tar.gz"
     blueprint = blueprint_factory(fetch=url)
-    async with AsyncClient(app=app) as client:
+    async with AsyncClient(transport=httpx.ASGITransport(app=app)) as client:
         with pytest.raises(Ops2debUpdaterError) as e:
             await GithubUpdateStrategy(client)(blueprint)
         assert "Failed to determine latest release URL" in str(e)