Merge pull request #7 from memory/add-zstandard

Add support for zstandard control archives

Author: memory

Makefile

@@ -45,7 +45,7 @@ else
 endif
 
 ${PYENV}: ${BREW_SSL} ${BREW_READLINE} ${PYENV_BIN}
-	${ARCH_PREFIX} pyenv install ${PYVERSION}
+	${ARCH_PREFIX} pyenv install -s ${PYVERSION}
 
 ${VENV}: ${PYENV}
 	${ARCH_PREFIX} ${PYENV_BIN} virtualenv ${PYVERSION} ${VENV_NAME}

poetry.lock

@@ -13,6 +13,7 @@
 
 # pypi imports
 import six
+import zstandard
 from arpy import Archive
 
 # local imports
@@ -255,9 +256,12 @@ def _process_dpkg_file(self, filename):
         elif b"control.tar.xz" in dpkg_archive.archived_files:
             control_archive = dpkg_archive.archived_files[b"control.tar.xz"]
             control_archive_type = "xz"
+        elif b"control.tar.zst" in dpkg_archive.archived_files:
+            control_archive = dpkg_archive.archived_files[b"control.tar.zst"]
+            control_archive_type = "zst"
         else:
             raise DpkgMissingControlGzipFile(
-                "Corrupt dpkg file: no control.tar.gz/xz file in ar archive."
+                "Corrupt dpkg file: no control.tar.gz/xz/zst file in ar archive."
             )
         self._log.debug("found controlgz: %s", control_archive)
 
@@ -267,12 +271,19 @@ def _process_dpkg_file(self, filename):
                 with tarfile.open(fileobj=io.BytesIO(gzf.read())) as ctar:
                     self._log.debug("opened tar file: %s", ctar)
                     message = self._extract_message(ctar)
-        else:
+        elif control_archive_type == "xz":
             with lzma.open(control_archive) as xzf:
                 self._log.debug("opened xz control archive: %s", xzf)
                 with tarfile.open(fileobj=io.BytesIO(xzf.read())) as ctar:
                     self._log.debug("opened tar file: %s", ctar)
                     message = self._extract_message(ctar)
+        else:
+            zst = zstandard.ZstdDecompressor()
+            with zst.stream_reader(control_archive) as reader:
+                self._log.debug("opened zst control archive: %s", reader)
+                with tarfile.open(fileobj=io.BytesIO(reader.read())) as ctar:
+                    self._log.debug("opened tar file: %s", ctar)
+                    message = self._extract_message(ctar)
 
         for req in REQUIRED_HEADERS:
             if req not in list(map(str.lower, message.keys())):

pydpkg/dpkg.py

@@ -14,11 +14,11 @@ class DpkgVersionError(DpkgError):
 
 
 class DpkgMissingControlFile(DpkgError):
-    """No control file found in control.tar.gz/xz"""
+    """No control file found in control.tar.gz/xz/zst"""
 
 
 class DpkgMissingControlGzipFile(DpkgError):
-    """No control.tar.gz/xz file found in dpkg file"""
+    """No control.tar.gz/xz/zst file found in dpkg file"""
 
 
 class DpkgMissingRequiredHeaderError(DpkgError):

pydpkg/exceptions.py

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "pydpkg"
-version = "1.6.0"
+version = "1.7.0"
 description = "A python library for parsing debian package control headers and version strings"
 authors = ["Nathan J. Mehl <pypi@memory.blank.org>"]
 homepage = "https://github.com/memory/python-dpkg"
@@ -31,6 +31,7 @@ python = "^3.6.2"
 arpy = "^2.2.0"
 six = "^1.16.0"
 PGPy = "0.5.4"
+zstandard = "^0.18.0"
 
 [tool.poetry.dev-dependencies]
 black = "^21.11b1"

pyproject.toml

@@ -1,14 +1,17 @@
 #!/usr/bin/env python
 
 import os
+import pytest
 import unittest
 from email.message import Message
 
 from pydpkg.dpkg import Dpkg
-from pydpkg.exceptions import DpkgVersionError
+from pydpkg.exceptions import DpkgVersionError, DpkgMissingControlGzipFile
 
 TEST_DPKG_GZ_FILE = "testdeb_1:0.0.0-test_all.deb"
 TEST_DPKG_XZ_FILE = "sample_package_xz.deb"
+TEST_DPKG_ZST_FILE = "sample_package_zst.deb"
+TEST_DPKG_BAD_FILE = "sample_package_badcontrol.deb"
 
 
 class DpkgGzTest(unittest.TestCase):
@@ -67,6 +70,41 @@ def test_message(self):
         self.assertIsInstance(self.dpkg.message, type(Message()))
 
 
+class DpkgZstTest(unittest.TestCase):
+    def setUp(self):
+        dpkgfile = os.path.join(os.path.dirname(__file__), TEST_DPKG_ZST_FILE)
+        self.dpkg = Dpkg(dpkgfile)
+
+    def test_get_versions(self):
+        self.assertEqual(self.dpkg.epoch, 0)
+        self.assertEqual(self.dpkg.upstream_version, "0.0.1")
+        self.assertEqual(self.dpkg.debian_revision, "0")
+
+    def test_get_message_headers(self):
+        self.assertEqual(self.dpkg.package, "samplepackage.test")
+        self.assertEqual(self.dpkg.PACKAGE, "samplepackage.test")
+        self.assertEqual(self.dpkg["package"], "samplepackage.test")
+        self.assertEqual(self.dpkg["PACKAGE"], "samplepackage.test")
+        self.assertEqual(self.dpkg.get("package"), "samplepackage.test")
+        self.assertEqual(self.dpkg.get("PACKAGE"), "samplepackage.test")
+        self.assertEqual(self.dpkg.get("nonexistent"), None)
+        self.assertEqual(self.dpkg.get("nonexistent", "foo"), "foo")
+
+    def test_missing_header(self):
+        self.assertRaises(KeyError, self.dpkg.__getitem__, "xyzzy")
+        self.assertRaises(AttributeError, self.dpkg.__getattr__, "xyzzy")
+
+    def test_message(self):
+        self.assertIsInstance(self.dpkg.message, type(Message()))
+
+
+class DpkgBadTest(unittest.TestCase):
+    def test_bad_control(self):
+        with pytest.raises(DpkgMissingControlGzipFile):
+            dpkgfile = os.path.join(os.path.dirname(__file__), TEST_DPKG_BAD_FILE)
+            Dpkg(dpkgfile).message
+
+
 class DpkgVersionsTest(unittest.TestCase):
     def test_get_epoch(self):
         self.assertEqual(Dpkg.get_epoch("0"), (0, "0"))
@@ -222,15 +260,17 @@ def test_compare_versions(self):
         )
 
         # unicode me harder
-        self.assertEqual(Dpkg.compare_versions(u"2:0.0.44-1", u"2:0.0.44-nobin"), -1)
-        self.assertEqual(Dpkg.compare_versions(u"2:0.0.44-nobin", u"2:0.0.44-1"), 1)
-        self.assertEqual(Dpkg.compare_versions(u"2:0.0.44-1", u"2:0.0.44-1"), 0)
+        self.assertEqual(Dpkg.compare_versions("2:0.0.44-1", "2:0.0.44-nobin"), -1)
+        self.assertEqual(Dpkg.compare_versions("2:0.0.44-nobin", "2:0.0.44-1"), 1)
+        self.assertEqual(Dpkg.compare_versions("2:0.0.44-1", "2:0.0.44-1"), 0)
 
 
 if __name__ == "__main__":
     suite = unittest.TestLoader().loadTestsFromTestCase(DpkgGzTest)
     unittest.TextTestRunner(verbosity=2).run(suite)
     suite = unittest.TestLoader().loadTestsFromTestCase(DpkgXzTest)
     unittest.TextTestRunner(verbosity=2).run(suite)
+    suite = unittest.TestLoader().loadTestsFromTestCase(DpkgZstTest)
+    unittest.TextTestRunner(verbosity=2).run(suite)
     suite = unittest.TestLoader().loadTestsFromTestCase(DpkgVersionsTest)
     unittest.TextTestRunner(verbosity=2).run(suite)