182 lines
8.4 KiB
Python
182 lines
8.4 KiB
Python
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()
|