from __future__ import annotations import os import stat import tempfile import unittest from pathlib import Path from support import FakeRunner, FakeStore, make_config, plan from netbox_store_agent.executor import OperationProcessor from netbox_store_agent.journal import Journal, ManagedPlugin from netbox_store_agent.protocol import OperationRequest class ExecutorTests(unittest.TestCase): def setUp(self) -> None: self.temporary = tempfile.TemporaryDirectory() self.root = Path(self.temporary.name) self.key = "20f4274f-d4e5-42bf-9164-967b1a774481" def tearDown(self) -> None: self.temporary.cleanup() def request(self, action: str, *, version: str | None = None, token: str | None = None) -> OperationRequest: return OperationRequest( "eea17d87-8944-4ee2-a076-363338ab746d", action, "demo-plugin", version, token, "alice", ) def seed(self, journal: Journal, *, enabled: bool) -> ManagedPlugin: release = plan() plugin = ManagedPlugin( "demo-plugin", "demo-plugin", "demo_plugin", "1.2.3", enabled, (release.requirement(),), ) journal.upsert_managed_plugin(plugin) return plugin def test_dry_run_verifies_artifact_without_runner_or_state(self) -> None: config = make_config(self.root, dry_run=True) journal = Journal(config.agent.journal_path) store = FakeStore() runner = FakeRunner() journal.submit(self.key, self.request("install", version="1.2.3", token="c" * 64)) OperationProcessor(config, journal, store=store, runner=runner).process(self.key) operation = journal.get_operation(self.key) self.assertEqual(operation["state"], "dry_run") self.assertIn(("download", "1.2.3"), store.calls) self.assertEqual(runner.commands, []) self.assertIsNone(journal.get_managed_plugin("demo-plugin")) self.assertFalse(config.paths.include_path.exists()) if os.name == "posix": self.assertEqual(stat.S_IMODE(config.paths.temp_dir.stat().st_mode), 0o711) def test_install_is_disabled_and_uses_fixed_pip_argv(self) -> None: config = make_config(self.root, dry_run=False) journal = Journal(config.agent.journal_path) runner = FakeRunner() journal.submit(self.key, self.request("install", version="1.2.3", token="c" * 64)) OperationProcessor(config, journal, store=FakeStore(), runner=runner).process(self.key) operation = journal.get_operation(self.key) self.assertEqual(operation["state"], "succeeded") managed = journal.get_managed_plugin("demo-plugin") self.assertIsNotNone(managed) self.assertFalse(managed.enabled) self.assertEqual(len(runner.commands), 1) command = runner.commands[0] self.assertIn("--no-index", command) self.assertIn("--no-deps", command) self.assertIn("--require-hashes", command) def test_approval_token_mismatch_fails_before_download(self) -> None: config = make_config(self.root, dry_run=False) journal = Journal(config.agent.journal_path) store = FakeStore() runner = FakeRunner() journal.submit(self.key, self.request("install", version="1.2.3", token="d" * 64)) OperationProcessor(config, journal, store=store, runner=runner).process(self.key) self.assertEqual(journal.get_operation(self.key)["state"], "failed") self.assertFalse(any(call[0] == "download" for call in store.calls)) self.assertEqual(runner.commands, []) def test_enable_runs_migrate_collectstatic_and_both_service_restart(self) -> None: config = make_config(self.root, dry_run=False) journal = Journal(config.agent.journal_path) self.seed(journal, enabled=False) runner = FakeRunner() journal.submit(self.key, self.request("enable")) OperationProcessor(config, journal, store=FakeStore(), runner=runner).process(self.key) self.assertEqual(journal.get_operation(self.key)["state"], "succeeded") self.assertTrue(journal.get_managed_plugin("demo-plugin").enabled) flattened = [item for command in runner.commands for item in command] self.assertIn("migrate", flattened) self.assertIn("collectstatic", flattened) self.assertIn("netbox", flattened) self.assertIn("netbox-rq", flattened) def test_uninstall_requires_disabled(self) -> None: config = make_config(self.root, dry_run=False) journal = Journal(config.agent.journal_path) self.seed(journal, enabled=True) runner = FakeRunner() journal.submit(self.key, self.request("uninstall")) OperationProcessor(config, journal, store=FakeStore(), runner=runner).process(self.key) self.assertEqual(journal.get_operation(self.key)["state"], "failed") self.assertEqual(runner.commands, []) def test_disable_restarts_services_and_persists_disabled_state(self) -> None: config = make_config(self.root, dry_run=False) journal = Journal(config.agent.journal_path) self.seed(journal, enabled=True) runner = FakeRunner() journal.submit(self.key, self.request("disable")) OperationProcessor(config, journal, store=FakeStore(), runner=runner).process(self.key) self.assertEqual(journal.get_operation(self.key)["state"], "succeeded") self.assertFalse(journal.get_managed_plugin("demo-plugin").enabled) self.assertEqual(len(runner.commands), 1) self.assertIn("restart", runner.commands[0]) def test_uninstall_disabled_plugin_uses_fixed_package_and_deletes_state(self) -> None: config = make_config(self.root, dry_run=False) journal = Journal(config.agent.journal_path) self.seed(journal, enabled=False) runner = FakeRunner() journal.submit(self.key, self.request("uninstall")) OperationProcessor(config, journal, store=FakeStore(), runner=runner).process(self.key) self.assertEqual(journal.get_operation(self.key)["state"], "succeeded") self.assertIsNone(journal.get_managed_plugin("demo-plugin")) self.assertEqual(len(runner.commands), 1) self.assertEqual(runner.commands[0][-1], "demo-plugin") def test_failed_pip_attempt_is_conservatively_manual_recovery(self) -> None: config = make_config(self.root, dry_run=False) journal = Journal(config.agent.journal_path) runner = FakeRunner(fail_step="install") journal.submit(self.key, self.request("install", version="1.2.3", token="c" * 64)) OperationProcessor(config, journal, store=FakeStore(), runner=runner).process(self.key) self.assertEqual(journal.get_operation(self.key)["state"], "manual_recovery") def test_failure_after_pip_requires_manual_recovery_and_restores_files(self) -> None: config = make_config(self.root, dry_run=False) config.paths.include_path.write_text("old include", encoding="utf-8") config.paths.requirements_path.write_text("old requirements", encoding="utf-8") journal = Journal(config.agent.journal_path) self.seed(journal, enabled=True) runner = FakeRunner(fail_step="restart") journal.submit(self.key, self.request("update", version="1.2.3", token="c" * 64)) OperationProcessor(config, journal, store=FakeStore(), runner=runner).process(self.key) self.assertEqual(journal.get_operation(self.key)["state"], "manual_recovery") self.assertEqual(config.paths.include_path.read_text(), "old include") self.assertEqual(config.paths.requirements_path.read_text(), "old requirements") def test_self_management_fails_before_store_access(self) -> None: config = make_config(self.root, dry_run=True) journal = Journal(config.agent.journal_path) store = FakeStore() request = OperationRequest( "eea17d87-8944-4ee2-a076-363338ab746d", "install", "netbox-plugin-store", "1.2.3", "c" * 64, "alice", ) journal.submit(self.key, request) OperationProcessor(config, journal, store=store, runner=FakeRunner()).process(self.key) self.assertEqual(journal.get_operation(self.key)["state"], "failed") self.assertEqual(store.calls, []) if __name__ == "__main__": unittest.main()