Implement and test PackageManager.gen_db_archive()

rbackup/package_managers/packagemanager.py

@@ -12,7 +12,7 @@ from pathlib import Path
 from tempfile import NamedTemporaryFile
 
 # ========== Constants ==========
-
+VALID_DB_COMPRESS_MODES = ["bzip2", "gz", "lzma", "xz"]
 
 # ========== Logging Setup ==========
 syslog = logging.getLogger(__name__)
@@ -83,21 +83,29 @@ class PackageManager:
             All arguments and keyword-only arguments are passed directly
             to the packagemanagerk
 
+        :param compress: compression mode
+        :type compress: str
         :returns: the path to the created file
         :rtype: path-like object
+        :raises: ValueError if compress is not in packagemanager.VALID_DB_COMPRESS_MODES
         """
-        raise NotImplementedError
-
         syslog.info("Creating a database archive")
 
-        with NamedTemporaryFile(delete=False) as db_archive:
-            archive = tarfile.open(*args, **kwargs)
+        if compress is not None and compress not in VALID_DB_COMPRESS_MODES:
+            raise ValueError(f"{compress} is not a valid compress mode")
 
-            archive.add(cachedir)
+        archivemode = "w" if compress is None else f"w:{compress}"
+        archivesuffix = ".tar" if compress is None else f".tar.{compress}"
+
+        with NamedTemporaryFile(delete=False, suffix=archivesuffix) as tmpfile:
+            archive_path = Path(tmpfile.name)
+
+        with tarfile.open(name=archive_path, mode=archivemode) as db_archive:
+            db_archive.add(self.database_path)
 
         syslog.info("Database archive generation complete")
 
-        return Path(db_archive.name)
+        return archive_path
 
     @property
     def cache_directory(self):

rbackup/tests/test_packagemanager.py

@@ -115,6 +115,32 @@ class TestPackageManagerMethods(unittest.TestCase):
         self.p._gen_pkglist()
         self.mocked_tempfile.assert_not_called()
 
+    def test_db_archive(self):
+        p = Path("tmpfile")
+        self.mocked_path.return_value = p
+
+        archive = self.p.gen_db_archive()
+
+        self.assertIsInstance(archive, Path)
+        self.mocked_tempfile.assert_called_with(delete=False, suffix=".tar")
+        self.mocked_tarfile.assert_called_with(name=p, mode="w")
+
+    def test_db_archive_compress_mode(self):
+        p = Path("tmpfile")
+        compress = "xz"
+        self.mocked_path.return_value = p
+
+        archive = self.p.gen_db_archive(compress)
+
+        self.assertIsInstance(archive, Path)
+        self.mocked_tempfile.assert_called_with(delete=False, suffix=".tar.xz")
+        self.mocked_tarfile.assert_called_with(name=p, mode="w:xz")
+
+    @given(from_regex(r"(?!gz|bz2|lzma|xz)"))
+    def test_db_archive_invalid_compress_mode(self, invalid_mode):
+        with self.assertRaises(ValueError):
+            self.p.gen_db_archive(invalid_mode)
+
     def tearDown(self):
         self.patched_path.stop()
         self.patched_subprocess.stop()