Linux evo.fastest-server.com 5.14.0-284.1101.el9.tuxcare.11.els11.x86_64 #1 SMP PREEMPT_DYNAMIC Fri Aug 14 13:30:35 UTC 2026 x86_64
LiteSpeed
Server IP : 103.249.112.113 & Your IP : 216.73.217.135
Domains : 988 Domain
User : tanishks
Terminal
Auto Root
Create File
Create Folder
Localroot Suggester
Backdoor Destroyer
Readme
/
usr /
share /
lvemanager-xray /
plugins /
tests /
Delete
Unzip
Name
Size
Permission
Date
Action
test_install_xray_plugin.py
29.14
KB
-rw-r--r--
2026-08-11 11:44
test_upload_reports_destination.py
7.35
KB
-rw-r--r--
2026-08-11 11:44
Save
Rename
""" Product-behaviour and integration-contract tests for the X-Ray plugin installer. 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. Tests cover shared filesystem primitives plus panel destination, selection, lifecycle, cron, and UI/agent-state contracts without requiring a provisioned panel. 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 contextlib import importlib.util import os import shutil import stat import sys import tempfile import types import unittest from unittest import mock 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_open_dir_path_preserves_absolute_root(self): fd = self.base._open_dir_path("/") try: opened = os.fstat(fd) expected = os.stat("/") self.assertEqual((opened.st_dev, opened.st_ino), (expected.st_dev, expected.st_ino)) finally: os.close(fd) 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_creates_missing_parent_chain(self): src = os.path.join(self.tmp, "src.txt") dst = os.path.join(self.tmp, "new", "nested", "dst.txt") self._write(src, b"new tree") self.base.copy_file_or_dir(src, dst) with open(dst, "rb") as fh: self.assertEqual(fh.read(), b"new tree") def test_copy_rejects_symlink_in_intermediate_parent_component(self): src = os.path.join(self.tmp, "src.txt") real_parent = os.path.join(self.tmp, "real-parent") linked_parent = os.path.join(self.tmp, "linked-parent") os.makedirs(os.path.join(real_parent, "nested")) os.symlink(real_parent, linked_parent) self._write(src, b"must not be copied through a symlink") self.base.copy_file_or_dir( src, os.path.join(linked_parent, "nested", "dst.txt"), ) self.assertFalse(os.path.exists(os.path.join(real_parent, "nested", "dst.txt"))) 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") self.base.remove_file_or_dir(target) self.assertFalse(os.path.exists(target)) def test_remove_rejects_symlink_in_intermediate_parent_component(self): real_parent = os.path.join(self.tmp, "real-parent") linked_parent = os.path.join(self.tmp, "linked-parent") os.makedirs(os.path.join(real_parent, "nested")) target = os.path.join(real_parent, "nested", "keep.txt") with open(target, "wb") as fh: fh.write(b"keep") os.symlink(real_parent, linked_parent) self.base.remove_file_or_dir(os.path.join(linked_parent, "nested", "keep.txt")) self.assertTrue(os.path.isfile(target)) class ArgvListHardeningTests(unittest.TestCase): """Argument-injection hardening for the root-run subprocess call sites. The cloudlinux-xray-manager calls (system_id) and the cPanel install/uninstall_plugin calls (theme name) must hand ``exec_command`` a pre-tokenized argv *list*, never a format string that ``parse_command`` re-tokenizes with ``shlex.split``. A list value passes through ``parse_command`` untouched, so whitespace / leading-dash content in ``system_id`` or a theme name can never become extra argv. ``get_system_id`` must additionally reject any value that is not the expected bare-digit token. """ @classmethod def setUpClass(cls): cls.installer = _load_installer_module() cls.Base = cls.installer.Base cls.Cpanel = cls.installer.CpanelPluginInstaller def setUp(self): self.base = self.Base() def test_enable_agent_passes_argv_list(self): captured = [] self.base.exec_command = lambda command, env=None: captured.append(command) or [] self.base.get_system_id = lambda: "1002462490" self.base.enable_agent() self.assertEqual(len(captured), 1) self.assertIsInstance( captured[0], list, "enable_agent must pass an argv list, not a re-tokenizable string", ) self.assertEqual( captured[0], [self.installer.XRAY_MANAGER_UTILITY, "enable-user-agent", "--system_id", "1002462490"], ) def test_disable_agent_passes_argv_list(self): captured = [] self.base.exec_command = lambda command, env=None: captured.append(command) or [] self.base.get_system_id = lambda: "1002462490" self.base.disable_agent() self.assertEqual(len(captured), 1) self.assertIsInstance(captured[0], list) self.assertEqual( captured[0], [self.installer.XRAY_MANAGER_UTILITY, "disable-user-agent", "--system_id", "1002462490"], ) def test_is_agent_running_passes_argv_list(self): captured = [] self.base.exec_command = lambda command, env=None: captured.append(command) or ['{"status": "enabled"}'] self.base.get_system_id = lambda: "1002462490" self.assertTrue(self.base.is_agent_running()) self.assertEqual(len(captured), 1) self.assertIsInstance(captured[0], list) self.assertEqual( captured[0], [self.installer.XRAY_MANAGER_UTILITY, "user-agent-status", "--system_id", "1002462490"], ) def test_is_agent_running_false_and_no_call_when_unregistered(self): # On an unregistered host get_system_id() returns None (a supported # state). is_agent_running must report False without ever invoking the # xray-manager — never pass None into the argv list. captured = [] self.base.exec_command = lambda command, env=None: captured.append(command) or [] self.base.get_system_id = lambda: None self.assertFalse(self.base.is_agent_running()) self.assertEqual(captured, [], "no subprocess call when system_id is None") def test_get_system_id_returns_token_unchanged(self): # get_system_id passes the registration token through unchanged; the # argument-injection fix is the argv-list construction at the call # sites, so a non-list-shaped value can never be re-tokenized. self.installer.get_rhn_systemid_value = lambda *a, **kw: "ID-1002462490" self.assertEqual(self.base.get_system_id(), "1002462490") def test_cpanel_install_plugin_normal_theme_builds_expected_command(self): # Happy path: a normal cPanel theme name must still produce exactly the # installer command per theme, one invocation per discovered theme. inst = self.Cpanel() captured = [] inst.exec_command = lambda command, env=None: captured.append(command) or [] inst.copy_file_or_dir = lambda *a, **kw: None inst.cpanel_fix_feature_manager = lambda: None inst.get_theme_list = lambda: ["paper_lantern", "jupiter"] inst.install_plugin() self.assertEqual( captured, [ [inst.plugin_installer, inst.plugin_tar, "--theme", "paper_lantern"], [inst.plugin_installer, inst.plugin_tar, "--theme", "jupiter"], ], ) def test_cpanel_install_plugin_passes_argv_list(self): inst = self.Cpanel() captured = [] inst.exec_command = lambda command, env=None: captured.append(command) or [] inst.copy_file_or_dir = lambda *a, **kw: None inst.cpanel_fix_feature_manager = lambda: None inst.get_theme_list = lambda: ["paper_lantern --evil"] inst.install_plugin() self.assertEqual(len(captured), 1) self.assertIsInstance( captured[0], list, "install_plugin must pass an argv list so a crafted theme name cannot inject extra argv", ) self.assertEqual( captured[0], [inst.plugin_installer, inst.plugin_tar, "--theme", "paper_lantern --evil"], ) def test_cpanel_uninstall_plugin_normal_theme_builds_expected_command(self): # Happy path: a normal cPanel theme name must still produce exactly the # uninstaller command per theme, one invocation per discovered theme. inst = self.Cpanel() captured = [] inst.exec_command = lambda command, env=None: captured.append(command) or [] inst.remove_file_or_dir = lambda *a, **kw: None inst.get_theme_list = lambda: ["paper_lantern", "jupiter"] inst.uninstall_plugin() self.assertEqual( captured, [ [inst.plugin_uninstaller, inst.plugin_tar, "--theme", "paper_lantern"], [inst.plugin_uninstaller, inst.plugin_tar, "--theme", "jupiter"], ], ) def test_cpanel_uninstall_plugin_passes_argv_list(self): inst = self.Cpanel() captured = [] inst.exec_command = lambda command, env=None: captured.append(command) or [] inst.remove_file_or_dir = lambda *a, **kw: None inst.get_theme_list = lambda: ["paper_lantern --evil"] inst.uninstall_plugin() self.assertEqual(len(captured), 1) self.assertIsInstance(captured[0], list) self.assertEqual( captured[0], [inst.plugin_uninstaller, inst.plugin_tar, "--theme", "paper_lantern --evil"], ) class ConfigurePluginContractTests(unittest.TestCase): """UI visibility and repair rules must stay stable across package upgrades.""" @classmethod def setUpClass(cls): cls.installer = _load_installer_module() cls.Base = cls.installer.Base def setUp(self): self.base = self.Base() self.base.sync_ui_config = mock.Mock() self.base.enable_agent = mock.Mock() self.base.disable_agent = mock.Mock() def test_visible_plugin_syncs_config_and_enables_agent(self): self.base.is_agent_enabled_in_config = mock.Mock(return_value=True) self.base.configure_plugin() self.base.sync_ui_config.assert_called_once_with() self.base.enable_agent.assert_called_once_with() self.base.disable_agent.assert_not_called() def test_hidden_plugin_syncs_config_without_disabling_agent(self): self.base.is_agent_enabled_in_config = mock.Mock(return_value=False) self.base.configure_plugin() self.base.sync_ui_config.assert_called_once_with() self.base.enable_agent.assert_not_called() self.base.disable_agent.assert_not_called() def test_repair_reconfigures_visible_plugin_only_when_agent_is_stopped(self): self.base.is_agent_enabled_in_config = mock.Mock(return_value=True) self.base.is_agent_running = mock.Mock(return_value=False) self.base.configure_plugin = mock.Mock() self.base.check_and_repair_plugin() self.base.configure_plugin.assert_called_once_with() def test_repair_is_noop_when_visible_agent_is_running(self): self.base.is_agent_enabled_in_config = mock.Mock(return_value=True) self.base.is_agent_running = mock.Mock(return_value=True) self.base.configure_plugin = mock.Mock() self.base.check_and_repair_plugin() self.base.configure_plugin.assert_not_called() def test_repair_does_not_query_or_disable_hidden_agent(self): self.base.is_agent_enabled_in_config = mock.Mock(return_value=False) self.base.is_agent_running = mock.Mock() self.base.configure_plugin = mock.Mock() self.base.check_and_repair_plugin() self.base.is_agent_running.assert_not_called() self.base.configure_plugin.assert_not_called() self.base.disable_agent.assert_not_called() class CronContractTests(unittest.TestCase): """Repeated installs overwrite one exact cron entry instead of appending.""" @classmethod def setUpClass(cls): cls.installer = _load_installer_module() def test_add_cron_is_idempotent_and_writes_exact_entry(self): base = self.installer.Base() cron = self.installer.CHECK_PLUGIN_CRON expected_path = "/etc/cron.d/" + cron["name"] expected_entry = "{} {} {}\n".format( cron["schedule"], cron["executor"], cron["command"], ) opened = mock.mock_open() with mock.patch("builtins.open", opened): for _ in range(2): base.add_cron( cron["name"], cron["schedule"], cron["executor"], cron["command"], ) self.assertEqual( opened.call_args_list, [mock.call(expected_path, "w"), mock.call(expected_path, "w")], ) self.assertEqual( opened().write.call_args_list, [mock.call(expected_entry), mock.call(expected_entry)], ) class PanelDestinationContractTests(unittest.TestCase): """Every panel installer must stage the same SPAs at its documented paths.""" @classmethod def setUpClass(cls): cls.installer = _load_installer_module() def test_cpanel_destinations_include_both_spas_and_each_theme_template(self): inst = self.installer.CpanelPluginInstaller() inst.copy_file_or_dir = mock.Mock() inst.exec_command = mock.Mock() inst.cpanel_fix_feature_manager = mock.Mock() inst.get_theme_list = mock.Mock(return_value=["jupiter"]) inst.install_plugin() self.assertEqual( inst.copy_file_or_dir.call_args_list, [ mock.call(self.installer.SOURCE_PATH_ADMIN, inst.destination_admin), mock.call(self.installer.SOURCE_PATH_USER, inst.destination_user), mock.call(inst.template_src, inst.template_dst.format("jupiter")), ], ) def test_plesk_destinations_include_spas_controllers_icon_and_template(self): inst = self.installer.PleskPluginInstaller() inst.copy_file_or_dir = mock.Mock() with mock.patch.object(self.installer.os.path, "isdir", return_value=True): inst.install_plugin() self.assertEqual( inst.copy_file_or_dir.call_args_list, [ mock.call(self.installer.SOURCE_PATH_ADMIN, inst.destination_admin), mock.call(self.installer.SOURCE_PATH_USER, inst.destination_user), mock.call(inst.controller_src, inst.controller_dst), mock.call(inst.send_request_controller_src, inst.send_request_controller_dst), mock.call(inst.icon_src, inst.icon_dst), mock.call(inst.template_src, inst.template_dst), ], ) def test_directadmin_destinations_and_permissions_are_exact(self): inst = self.installer.DirectAdminPluginInstaller() inst.copy_file_or_dir = mock.Mock() inst.safe_recursive_chown = mock.Mock() inst.safe_recursive_chmod = mock.Mock() inst.safe_chmod_nofollow = mock.Mock() inst.install_plugin() self.assertEqual( inst.copy_file_or_dir.call_args_list, [ mock.call(self.installer.SOURCE_PATH_ADMIN, inst.destination_admin_spa), mock.call(inst.source_user_plugin, inst.destination_user_plugin), mock.call(inst.source_index_file, inst.destination_index_file), mock.call(self.installer.SOURCE_PATH_USER, inst.destination_user_spa), ], ) self.assertEqual( inst.safe_recursive_chown.call_args_list, [ mock.call(inst.destination_user_plugin, "diradmin", "diradmin"), mock.call(inst.destination_admin_spa, "diradmin", "diradmin"), ], ) inst.safe_recursive_chmod.assert_called_once_with(inst.destination_user_plugin, 0o755) inst.safe_chmod_nofollow.assert_called_once_with(inst.plugin_conf_file, 0o644) def test_custom_panel_stages_static_and_panel_specific_copies(self): inst = self.installer.PanelIntegrationPluginInstaller() inst.copy_file_or_dir = mock.Mock() inst.get_panel_base_path = mock.Mock(return_value="/opt/vendor/ui/") inst.install_plugin() self.assertEqual( inst.copy_file_or_dir.call_args_list, [ mock.call(self.installer.SOURCE_PATH_ADMIN, inst.destination_admin_py_plugin), mock.call(self.installer.SOURCE_PATH_USER, inst.destination_user_py_plugin), mock.call(self.installer.SOURCE_PATH_ADMIN, "/opt/vendor/ui/assets/xray-admin"), mock.call(self.installer.SOURCE_PATH_USER, "/opt/vendor/ui/assets/xray-user"), ], ) class MainLifecycleContractTests(unittest.TestCase): """Panel selection and install/uninstall/check call ordering are contractual.""" @classmethod def setUpClass(cls): cls.installer = _load_installer_module() def _run(self, args, cpanel=False, directadmin=False, plesk=False, integration=False): main = self.installer.Main() parser = mock.Mock() parser.parse_args.return_value = args main.make_parser = mock.Mock(return_value=parser) selected = mock.Mock() constructors = { "cpanel": mock.Mock(return_value=selected), "directadmin": mock.Mock(return_value=selected), "plesk": mock.Mock(return_value=selected), "integration": mock.Mock(return_value=selected), } patchers = [ mock.patch.object(self.installer.detect, "is_cpanel", create=True, return_value=cpanel), mock.patch.object(self.installer.detect, "is_da", create=True, return_value=directadmin), mock.patch.object(self.installer.detect, "is_plesk", create=True, return_value=plesk), mock.patch.object(self.installer.os.path, "isfile", return_value=integration), mock.patch.object(self.installer, "CpanelPluginInstaller", constructors["cpanel"]), mock.patch.object( self.installer, "DirectAdminPluginInstaller", constructors["directadmin"], ), mock.patch.object(self.installer, "PleskPluginInstaller", constructors["plesk"]), mock.patch.object( self.installer, "PanelIntegrationPluginInstaller", constructors["integration"], ), ] with contextlib.ExitStack() as stack: for patcher in patchers: stack.enter_context(patcher) main.run() return selected, constructors, parser def test_panel_detection_uses_documented_first_match_order(self): no_action = mock.Mock(install=False, uninstall=False, check=False) cases = [ ( {"cpanel": True, "directadmin": True, "plesk": True, "integration": True}, "cpanel", ), ( {"cpanel": False, "directadmin": True, "plesk": True, "integration": True}, "directadmin", ), ( {"cpanel": False, "directadmin": False, "plesk": True, "integration": True}, "plesk", ), ( {"cpanel": False, "directadmin": False, "plesk": False, "integration": True}, "integration", ), ] for detection, expected in cases: with self.subTest(expected=expected): _, constructors, parser = self._run(no_action, **detection) for name, constructor in constructors.items(): if name == expected: constructor.assert_called_once_with() else: constructor.assert_not_called() parser.print_help.assert_called_once_with() def test_install_lifecycle_order_and_exact_cron_contract(self): args = mock.Mock(install=True, uninstall=False, check=False) selected, _, _ = self._run(args, cpanel=True) cron = self.installer.CHECK_PLUGIN_CRON self.assertEqual( selected.method_calls, [ mock.call.install_plugin(), mock.call.generate_translate_templates(), mock.call.configure_plugin(), mock.call.add_cron( cron["name"], cron["schedule"], cron["executor"], cron["command"], ), ], ) def test_uninstall_lifecycle_removes_cron_before_panel_files_and_config(self): args = mock.Mock(install=False, uninstall=True, check=False) selected, _, _ = self._run(args, cpanel=True) self.assertEqual( selected.method_calls, [ mock.call.remove_cron(self.installer.CHECK_PLUGIN_CRON["name"]), mock.call.uninstall_plugin(), mock.call.reset_ui_config(), ], ) def test_check_lifecycle_only_runs_repair(self): args = mock.Mock(install=False, uninstall=False, check=True) selected, _, _ = self._run(args, cpanel=True) self.assertEqual(selected.method_calls, [mock.call.check_and_repair_plugin()]) if __name__ == "__main__": unittest.main()