From 57f484f0dd653ff92af50a84bd07d8db89993fe3 Mon Sep 17 00:00:00 2001 From: Eric Torres Date: Sun, 31 Mar 2019 09:57:19 -0700 Subject: [PATCH] Initial implementation and tests for packagemanager module --- rbackup/package_managers/packagemanager.py | 131 +++++++++++++++++++++ rbackup/tests/test_packagemanager.py | 122 +++++++++++++++++++ 2 files changed, 253 insertions(+) create mode 100644 rbackup/package_managers/packagemanager.py create mode 100644 rbackup/tests/test_packagemanager.py diff --git a/rbackup/package_managers/packagemanager.py b/rbackup/package_managers/packagemanager.py new file mode 100644 index 0000000..9fb81fd --- /dev/null +++ b/rbackup/package_managers/packagemanager.py @@ -0,0 +1,131 @@ +""" +.. author:: Eric Torres +.. module:: rbackup.packagemanager +:synopsis: Module for package manager plugins +""" +import logging +import subprocess +import tarfile + +from collections.abc import Iterable +from pathlib import Path +from tempfile import NamedTemporaryFile + +# ========== Constants ========== + + +# ========== Logging Setup ========== +syslog = logging.getLogger(__name__) + + +# ========== Classes ========== +class PackageManager: + def __init__(self, cachedir, db_path, pkglist_cmd): + """Default constructor for the PackageManager class. + + :param cachedir: path to the package manager cache directory + :type cachedir: str or path-like object + :param db_path: path to the package manager database + :type db_path: str or path-like object + :param pkglist_cmd: command to list installed packages to stdout + :type pkglist_cmd: list + """ + if not isinstance(pkglist_cmd, Iterable) or isinstance(pkglist_cmd, dict): + raise TypeError("pkglist_cmd is the wrong type") + elif not pkglist_cmd: + raise ValueError(f"Package list command is empty: {pkglist_cmd}") + elif any(not isinstance(arg, str) for arg in pkglist_cmd): + raise TypeError(f"{pkglist_cmd} contains a non-str value") + elif any(not bool(arg) for arg in pkglist_cmd): + raise ValueError(f"{pkglist_cmd} contains an empty str value") + + self._cachedir = Path(cachedir) + self._db_path = Path(db_path) + self._pkglist_cmd = pkglist_cmd + + def __init_subclass__(cls, cachedir, db_path, pkglist_cmd): + """Default constructor for all child classes.""" + super().__init_subclass(cachedir, db_path, pkglist_cmd) + + def _gen_pkglist(self): + """Generate a text file listing installed packages + on the system and return the path to that file. + + If there is an error in the package listing command, then + it is to be assumed that no file was created, therefore there + is no file to cleanup. + + Note that this method is internal and is + meant to be called from a subclass in a separate module. + + :returns: path to temporary file + :rtype: path-like object + """ + syslog.info("Creating a package list") + + try: + process = subprocess.run(self._pkglist_cmd, capture_output=True) + except subprocess.CalledProcessError as e: + syslog.error(e) + else: + with NamedTemporaryFile(mode="wb", delete=False) as pkglist: + pkglist.write(process.stdout) + + syslog.info("Package list generation complete") + return Path(pkglist.name) + + def _gen_db_archive(self, *args, **kwargs): + """Generate a database archive for this package manager. + + Note that this method is internal and is + meant to be called from a subclass in a separate module. + + All arguments and keyword-only arguments are passed directly + to the packagemanagerk + + :returns: the path to the created file + :rtype: path-like object + """ + raise NotImplementedError + + syslog.info("Creating a database archive") + + with NamedTemporaryFile(delete=False) as db_archive: + archive = tarfile.open(*args, **kwargs) + + archive.add(cachedir) + + syslog.info("Database archive generation complete") + + return Path(db_archive.name) + + @property + def cache_directory(self): + """Return the cache directory of this package manager. + + :rtype: path-like object + """ + return self._cachedir + + @property + def database_path(self): + """Return the database path of this package manager. + + :rtype: path-like object + """ + return self._db_path + + @property + def pkglist_cmd(self): + """Return the package listing command of this package manager. + + :rtype: iterable or str + """ + return self._pkglist_cmd + + +# ========== Functions ========== +if __name__ == "__main__": + import doctest + + doctest.testmod() diff --git a/rbackup/tests/test_packagemanager.py b/rbackup/tests/test_packagemanager.py new file mode 100644 index 0000000..0c17a65 --- /dev/null +++ b/rbackup/tests/test_packagemanager.py @@ -0,0 +1,122 @@ +""" +.. author:: Eric Torres +:synopsis: Unit tests for the PackageManager module. +""" +import doctest +import subprocess +import unittest + +from hypothesis import given, note +from hypothesis.strategies import ( + booleans, + dictionaries, + integers, + iterables, + lists, + one_of, + none, + text, +) +from pathlib import Path +from rbackup.package_managers.packagemanager import PackageManager +from unittest.mock import patch + +# ========== Constants ========== +TESTING_MODULE = "rbackup.package_managers.packagemanager" + + +# ========== Functions ========== +def load_tests(loader, tests, ignore): + tests.addTests(doctest.DocTestSuite(TESTING_MODULE)) + return tests + + +# ========== Test Cases ========== +class TestCreatePackageManager(unittest.TestCase): + def setUp(self): + self.patched_path = patch(f"{TESTING_MODULE}.Path", autospec=Path) + self.patched_subprocess = patch(f"{TESTING_MODULE}.subprocess.run") + self.patched_tarfile = patch(f"{TESTING_MODULE}.tarfile.open") + self.patched_tempfile = patch(f"{TESTING_MODULE}.NamedTemporaryFile") + + self.mocked_path = self.patched_path.start() + self.mocked_run = self.patched_subprocess.start() + self.mocked_tarfile = self.patched_tarfile.start() + self.mocked_tempfile = self.patched_tempfile.start() + + self.cachedir = "/var/cache/pacman/" + self.db_path = "/var/lib/pacman" + self.pkglist_cmd = ["pacman", "-Qqe"] + self.p = PackageManager(self.cachedir, self.db_path, self.pkglist_cmd) + + @given(one_of(text(min_size=1), iterables(text(min_size=1), min_size=1))) + def test_create_with_valid_values(self, l): + PackageManager("nothing", "nothing", l) + + @given(one_of(none(), booleans(), integers(), dictionaries(text(), text()))) + def test_incorrect_cmd_type(self, cmd): + with self.assertRaises(TypeError): + PackageManager("nothing", "nothing", cmd) + + def test_empty_cmd(self): + with self.assertRaises(ValueError): + PackageManager("nothing", "nothing", []) + PackageManager("nothing", "nothing", set()) + PackageManager("nothing", "nothing", '') + + @given(iterables(one_of(none(), booleans(), integers()), min_size=1)) + def test_wrong_iterable_element_type(self, cmd): + with self.assertRaises(TypeError): + PackageManager("nothing", "nothing", cmd) + + def test_empty_str_in_iterable(self): + with self.assertRaises(ValueError): + PackageManager("nothing", "nothing", ['']) + PackageManager("nothing", "nothing", ['pacman', '']) + + def tearDown(self): + self.patched_path.stop() + self.patched_subprocess.stop() + self.patched_tarfile.stop() + self.patched_tempfile.stop() + + +class TestPackageManagerMethods(unittest.TestCase): + def setUp(self): + self.patched_path = patch(f"{TESTING_MODULE}.Path", autospec=Path) + self.patched_subprocess = patch(f"{TESTING_MODULE}.subprocess.run") + self.patched_tarfile = patch(f"{TESTING_MODULE}.tarfile.open") + self.patched_tempfile = patch(f"{TESTING_MODULE}.NamedTemporaryFile") + + self.mocked_path = self.patched_path.start() + self.mocked_run = self.patched_subprocess.start() + self.mocked_tarfile = self.patched_tarfile.start() + self.mocked_tempfile = self.patched_tempfile.start() + + self.cachedir = "/var/cache/pacman/" + self.db_path = "/var/lib/pacman" + self.pkglist_cmd = ["pacman", "-Qqe"] + self.p = PackageManager(self.cachedir, self.db_path, self.pkglist_cmd) + + def test_pkglist(self): + self.mocked_run.return_value.stdout = "packages" + self.mocked_tempfile.return_value.name = "tempfile" + + pkglist = self.p._gen_pkglist() + + self.mocked_tempfile.return_value.__enter__.return_value.write.assert_called_with( + "packages" + ) + self.assertIsInstance(pkglist, Path) + + def test_pkglist_subprocess_error(self): + self.mocked_run.side_effect = subprocess.CalledProcessError(1, self.pkglist_cmd) + + self.p._gen_pkglist() + self.mocked_tempfile.assert_not_called() + + def tearDown(self): + self.patched_path.stop() + self.patched_subprocess.stop() + self.patched_tarfile.stop() + self.patched_tempfile.stop()