Netbox-Store/netbox_plugin/tests/test_lifecycle.py

import hashlib
import tempfile
import unittest
from pathlib import Path

import _bootstrap  # noqa: F401

from netbox_plugin_store.client import CatalogPlugin
from netbox_plugin_store.commands import CommandResult
from netbox_plugin_store.lifecycle import LifecycleError, LifecycleRequest, LifecycleService
from netbox_plugin_store.runtime import RuntimeSettings


ARTIFACT = b"approved wheel bytes"


def catalog_plugin():
    return CatalogPlugin.from_mapping(
        {
            "slug": "example-plugin",
            "name": "Example",
            "package_name": "netbox-example",
            "import_name": "netbox_example",
            "latest_version": "1.0.0",
            "approved": True,
            "releases": [
                {
                    "version": "1.0.0",
                    "download_url": "https://store.example/example.whl",
                    "sha256": hashlib.sha256(ARTIFACT).hexdigest(),
                    "approved_payload_sha256": "b" * 64,
                    "approved": True,
                    "immutable": True,
                    "min_netbox_version": "4.6.5",
                    "max_netbox_version": "4.6.8",
                }
            ],
        }
    )


class FakeClient:
    def __init__(self, plugin):
        self.plugin = plugin
        self.downloads = 0

    def get_plugin(self, slug):
        return self.plugin

    def download_artifact(self, url, destination, **kwargs):
        self.downloads += 1
        destination.write_bytes(ARTIFACT)
        return hashlib.sha256(ARTIFACT).hexdigest(), len(ARTIFACT)


class FakeRepository:
    def __init__(self):
        self.audits = {}
        self.statuses = []

    def create_audit(self, request, actor_id):
        self.audits[1] = {"status": "queued"}
        return 1

    def mark_running(self, audit_id):
        self.audits[audit_id]["status"] = "running"

    def mark_success(self, audit_id, result):
        self.audits[audit_id].update(status="succeeded", result=result)

    def mark_handed_off(self, audit_id, operation_id, result):
        self.audits[audit_id].update(status="handed-off", operation_id=operation_id, result=result)

    def mark_failed(self, audit_id, error, result):
        self.audits[audit_id].update(status="failed", error=error)

    def update_status(self, plugin, **values):
        self.statuses.append(values)

    def has_other_pending_mutation(self, slug, audit_id):
        return False


class FakeRunner:
    def __init__(self):
        self.calls = []

    def run(self, argv, **kwargs):
        self.calls.append((list(argv), kwargs))
        return CommandResult(0, "ok", "")


def runtime(root: Path, *, execution_mode="direct", allow=True):
    config = root / "configuration.py"
    config.write_text("PLUGINS = ['netbox_plugin_store']\n", encoding="utf-8")
    return RuntimeSettings.from_mapping(
        {
            "store_url": "https://store.example",
            "allowed_store_urls": ["https://store.example"],
            "allowed_artifact_urls": ["https://store.example"],
            "configuration_path": str(config),
            "requirements_path": str(root / "local_requirements.txt"),
            "manage_path": str(root / "manage.py"),
            "lock_path": str(root / "operation.lock"),
            "backup_dir": str(root / "backups"),
            "execution_mode": execution_mode,
            "allow_lifecycle_mutations": allow,
            "run_migrations": False,
            "collect_static": False,
            "allow_package_index": False,
        },
        configuration_dir=root,
        netbox_root=root,
        base_dir=root,
        netbox_version="4.6.8",
    )


class LifecycleTests(unittest.TestCase):
    def test_dry_run_has_no_download_or_subprocess(self):
        with tempfile.TemporaryDirectory() as temp_name:
            root = Path(temp_name)
            client = FakeClient(catalog_plugin())
            repository = FakeRepository()
            runner = FakeRunner()
            service = LifecycleService(runtime(root), client, repository, runner=runner, version_provider=lambda _: "")
            result = service.execute(LifecycleRequest("example-plugin", "install", "1.0.0", True))
            self.assertEqual(result.state, "dry-run")
            self.assertEqual(client.downloads, 0)
            self.assertEqual(runner.calls, [])

    def test_direct_install_is_disabled_and_uses_no_index_and_pip_check(self):
        with tempfile.TemporaryDirectory() as temp_name:
            root = Path(temp_name)
            client = FakeClient(catalog_plugin())
            repository = FakeRepository()
            runner = FakeRunner()
            service = LifecycleService(runtime(root), client, repository, runner=runner, version_provider=lambda _: "")
            result = service.execute(LifecycleRequest("example-plugin", "install", "1.0.0", False))
            self.assertFalse(result.enabled)
            self.assertFalse(result.restart_required)
            self.assertNotIn("netbox_example", (root / "configuration.py").read_text(encoding="utf-8"))
            requirement = (root / "local_requirements.txt").read_text(encoding="utf-8")
            self.assertIn("#sha256=", requirement)
            self.assertIn("--no-index", runner.calls[0][0])
            self.assertEqual(runner.calls[1][0][-2:], ["pip", "check"])

    def test_uninstall_requires_explicit_disable_first(self):
        with tempfile.TemporaryDirectory() as temp_name:
            root = Path(temp_name)
            settings = runtime(root)
            settings.configuration_path.write_text(
                "PLUGINS = ['netbox_plugin_store', 'netbox_example']\n", encoding="utf-8"
            )
            runner = FakeRunner()
            service = LifecycleService(
                settings,
                FakeClient(catalog_plugin()),
                FakeRepository(),
                runner=runner,
                version_provider=lambda _: "1.0.0",
            )
            with self.assertRaises(LifecycleError):
                service.execute(LifecycleRequest("example-plugin", "uninstall", dry_run=False))
            self.assertEqual(runner.calls, [])


if __name__ == "__main__":
    unittest.main()