""" Product-behaviour tests for the xray plugin installer's file-copy primitives. These tests pin the post-fix contract of ``Base.copy_file_or_dir`` and ``Base.remove_file_or_dir`` in ``plugins/install-xray-plugin.py``: the installer must still be able to install a regular file, install a directory tree, replace an existing destination of either kind, and remove files / trees / non-existent paths idempotently. The patch routes every destination syscall through a parent dir fd; these tests verify that the legitimate product behaviour did not regress. The installer module is loaded under a stubbed import environment because it imports CloudLinux-specific runtime modules (``cldetectlib``, ``clcommon.*``) that are not available outside a provisioned CloudLinux OS. We only exercise methods on the ``Base`` class, whose ``__init__`` does nothing and which does not touch the stubbed modules. Run from the repo root with: python3 -m unittest discover -s plugins/tests -p 'test_*.py' or directly: python3 plugins/tests/test_install_xray_plugin.py """ import importlib.util import os import shutil import stat import sys import tempfile import types import unittest def _install_clos_import_stubs(): """Register no-op stand-ins for CloudLinux-only modules. The installer module imports ``cldetectlib``, ``clcommon.utils``, ``clcommon.lib.cledition`` and ``clcommon.ui_config`` at top level. These are present on a CloudLinux OS host but not in a generic Python environment. We register minimal stubs so ``importlib`` can load the module under test. """ if "cldetectlib" not in sys.modules: sys.modules["cldetectlib"] = types.ModuleType("cldetectlib") if "clcommon" not in sys.modules: sys.modules["clcommon"] = types.ModuleType("clcommon") if "clcommon.utils" not in sys.modules: mod = types.ModuleType("clcommon.utils") mod.get_rhn_systemid_value = lambda *a, **kw: None sys.modules["clcommon.utils"] = mod if "clcommon.lib" not in sys.modules: sys.modules["clcommon.lib"] = types.ModuleType("clcommon.lib") if "clcommon.lib.cledition" not in sys.modules: mod = types.ModuleType("clcommon.lib.cledition") mod.is_cl_solo_edition = lambda: False sys.modules["clcommon.lib.cledition"] = mod if "clcommon.ui_config" not in sys.modules: mod = types.ModuleType("clcommon.ui_config") class _UIConfig: def __init__(self, *a, **kw): pass mod.UIConfig = _UIConfig sys.modules["clcommon.ui_config"] = mod def _load_installer_module(): """Load ``install-xray-plugin.py`` from the plugins/ directory. The file name contains a hyphen so it cannot be imported by the normal package mechanism; we use ``importlib`` with an explicit spec. """ _install_clos_import_stubs() here = os.path.dirname(os.path.abspath(__file__)) source_path = os.path.normpath(os.path.join(here, "..", "install-xray-plugin.py")) spec = importlib.util.spec_from_file_location("install_xray_plugin", source_path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module class CopyFileOrDirContractTests(unittest.TestCase): """``Base.copy_file_or_dir`` must install the source at the destination.""" @classmethod def setUpClass(cls): cls.installer = _load_installer_module() cls.Base = cls.installer.Base def setUp(self): self.tmp = tempfile.mkdtemp(prefix="xray-installer-test-") self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True) self.base = self.Base() def _write(self, path, content, mode=0o644): with open(path, "wb") as fh: fh.write(content) os.chmod(path, mode) def test_copy_regular_file_into_empty_destination(self): src = os.path.join(self.tmp, "src.txt") dst = os.path.join(self.tmp, "dst.txt") self._write(src, b"hello xray installer", mode=0o640) self.base.copy_file_or_dir(src, dst) self.assertTrue(os.path.isfile(dst)) with open(dst, "rb") as fh: self.assertEqual(fh.read(), b"hello xray installer") # File permissions should reflect the source mode. self.assertEqual(stat.S_IMODE(os.stat(dst).st_mode), 0o640) def test_copy_regular_file_replaces_existing_file(self): src = os.path.join(self.tmp, "src.txt") dst = os.path.join(self.tmp, "dst.txt") self._write(src, b"new content") self._write(dst, b"old content") self.base.copy_file_or_dir(src, dst) with open(dst, "rb") as fh: self.assertEqual(fh.read(), b"new content") def test_copy_regular_file_replaces_existing_directory(self): # install_plugin re-runs on upgrade; a destination that used to be a # directory must be replaced by a regular file when the source flipped # kind. The contract is "destination ends up matching the source". src = os.path.join(self.tmp, "src.txt") dst = os.path.join(self.tmp, "dst") self._write(src, b"file beats dir") os.mkdir(dst) self._write(os.path.join(dst, "stale.txt"), b"stale") self.base.copy_file_or_dir(src, dst) self.assertTrue(os.path.isfile(dst)) with open(dst, "rb") as fh: self.assertEqual(fh.read(), b"file beats dir") def test_copy_directory_tree_into_empty_destination(self): src = os.path.join(self.tmp, "src-tree") dst = os.path.join(self.tmp, "dst-tree") os.makedirs(os.path.join(src, "sub", "deeper")) self._write(os.path.join(src, "top.txt"), b"top-level", mode=0o644) self._write(os.path.join(src, "sub", "mid.txt"), b"mid-level", mode=0o600) self._write( os.path.join(src, "sub", "deeper", "leaf.txt"), b"leaf-level", mode=0o644, ) self.base.copy_file_or_dir(src, dst) # Every source leaf must be present at the destination with content # preserved. self.assertTrue(os.path.isdir(dst)) with open(os.path.join(dst, "top.txt"), "rb") as fh: self.assertEqual(fh.read(), b"top-level") with open(os.path.join(dst, "sub", "mid.txt"), "rb") as fh: self.assertEqual(fh.read(), b"mid-level") with open(os.path.join(dst, "sub", "deeper", "leaf.txt"), "rb") as fh: self.assertEqual(fh.read(), b"leaf-level") # Per-leaf file mode must be preserved (this is what install_plugin # relies on for the user-plugin assets it later chmods). self.assertEqual( stat.S_IMODE(os.stat(os.path.join(dst, "sub", "mid.txt")).st_mode), 0o600, ) def test_copy_directory_tree_replaces_existing_directory(self): # Upgrade scenario: a previous install left a populated directory at # the destination; the new install must replace it cleanly so stale # files do not linger. src = os.path.join(self.tmp, "src-tree") dst = os.path.join(self.tmp, "dst-tree") os.makedirs(os.path.join(src, "new-only")) self._write(os.path.join(src, "new-only", "fresh.txt"), b"fresh") os.makedirs(os.path.join(dst, "old-only")) self._write(os.path.join(dst, "old-only", "stale.txt"), b"stale") self._write(os.path.join(dst, "stale-top.txt"), b"stale-top") self.base.copy_file_or_dir(src, dst) self.assertTrue(os.path.isfile(os.path.join(dst, "new-only", "fresh.txt"))) self.assertFalse(os.path.exists(os.path.join(dst, "old-only"))) self.assertFalse(os.path.exists(os.path.join(dst, "stale-top.txt"))) class RemoveFileOrDirContractTests(unittest.TestCase): """``Base.remove_file_or_dir`` must remove the entry idempotently.""" @classmethod def setUpClass(cls): cls.installer = _load_installer_module() cls.Base = cls.installer.Base def setUp(self): self.tmp = tempfile.mkdtemp(prefix="xray-installer-test-") self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True) self.base = self.Base() def test_remove_regular_file(self): target = os.path.join(self.tmp, "to-remove.txt") with open(target, "wb") as fh: fh.write(b"x") self.base.remove_file_or_dir(target) self.assertFalse(os.path.exists(target)) def test_remove_directory_tree(self): target = os.path.join(self.tmp, "to-remove") os.makedirs(os.path.join(target, "nested")) with open(os.path.join(target, "nested", "leaf.txt"), "wb") as fh: fh.write(b"x") self.base.remove_file_or_dir(target) self.assertFalse(os.path.exists(target)) def test_remove_nonexistent_path_is_noop(self): # The installer calls remove_file_or_dir against destinations that # may not exist yet on first install; the call must not raise. target = os.path.join(self.tmp, "never-existed") try: self.base.remove_file_or_dir(target) except Exception as exc: # pragma: no cover - assertion failure path self.fail( "remove_file_or_dir raised on a missing path: {!r}".format(exc) ) self.assertFalse(os.path.exists(target)) if __name__ == "__main__": unittest.main()