feat: extract raw XZ archives

mybox/extractor.py

@@ -116,12 +116,16 @@ async def extract(self, *, archive: Path, target: Path) -> None:
         pass
 
 
-class Gzip(RawExtractor):
+class ShellRedirectRawExtractor(RawExtractor):
+    def __init__(self, *, driver: Driver, command: str):
+        super().__init__(driver=driver)
+        self.command = command
+
     async def extract(self, *, archive: Path, target: Path) -> None:
         await self.driver.run(
             "sh",
             "-c",
-            f"gunzip < {shlex.quote(str(archive))} > {shlex.quote(str(target))}",
+            f"{self.command} < {shlex.quote(str(archive))} > {shlex.quote(str(target))}",
         )
 
 
@@ -132,7 +136,9 @@ async def extract(self, *, archive: Path, target: Path) -> None:
 
 def _guess_single_extractor(url: str, *, driver: Driver) -> RawExtractor:
     if url.endswith(".gz"):
-        return Gzip(driver=driver)
+        return ShellRedirectRawExtractor(driver=driver, command="gunzip")
+    elif url.endswith(".xz"):
+        return ShellRedirectRawExtractor(driver=driver, command="xzcat")
     else:
         raise ValueError(f"Unknown archive format: {url}")
 

tests/test_extractor.py

@@ -7,7 +7,7 @@
 import pytest
 
 from mybox.driver import LocalDriver
-from mybox.extractor import Tar, get_extractor
+from mybox.extractor import Tar, get_extractor, get_single_extractor
 from mybox.utils import run
 
 
@@ -87,3 +87,22 @@ async def test_get_extractor_unknown():
         ValueError, match="Unknown archive format: https://example.com/"
     ):
         await get_extractor("https://example.com", driver=LocalDriver())
+
+
+@pytest.mark.trio
+@pytest.mark.parametrize("compress_command,extension", [("gzip", "gz"), ("xz", "xz")])
+async def test_single_extractor(compress_command: str, extension: str):
+    with tempfile.TemporaryDirectory() as srcdir:
+        srcpath = Path(srcdir)
+        (srcpath / "myfile").write_text("contents")
+        await run(compress_command, srcpath / "myfile")
+
+        archive = srcpath / f"myfile.{extension}"
+
+        extractor = await get_single_extractor(str(archive), driver=LocalDriver())
+        with tempfile.TemporaryDirectory() as destdir:
+            destpath = Path(destdir) / "target"
+
+            await extractor.extract(archive=archive, target=destpath)
+
+            assert destpath.read_text() == "contents"