Netbox-Store/host_agent/tests/support.py

from __future__ import annotations

import hashlib
import json
import sys
import zipfile
from pathlib import Path
from typing import Any

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))

from netbox_store_agent.catalog import HttpStatusError, PluginMetadata, ReleasePlan
from netbox_store_agent.config import (
    AgentSettings,
    CommandSettings,
    Config,
    PathSettings,
    PolicySettings,
    StoreSettings,
)
from netbox_store_agent.errors import ExecutionError
from netbox_store_agent.runner import CommandResult


def make_config(root: Path, *, dry_run: bool = True, require_peers: bool = False) -> Config:
    root = root.resolve()
    managed_root = root / "netbox"
    managed_root.mkdir(parents=True, exist_ok=True)
    state = root / "state"
    return Config(
        agent=AgentSettings(
            socket_path=root / "agent.sock",
            journal_path=state / "journal.sqlite3",
            lock_path=state / "lifecycle.lock",
            backup_dir=state / "backups",
            dry_run=dry_run,
            require_root=False,
            require_peer_credentials=require_peers,
            allowed_peer_uids=(0,),
            allowed_peer_gids=(0,),
            socket_mode=0o660,
            socket_uid=0,
            socket_gid=0,
            max_request_bytes=65536,
            connection_timeout_seconds=1,
            worker_threads=1,
        ),
        store=StoreSettings(
            base_url="http://store.test",
            plugin_endpoint_template="/api/v1/plugins/{plugin_slug}",
            release_endpoint_template="/api/v1/plugins/{plugin_slug}/releases/{version}",
            timeout_seconds=1,
            max_catalog_bytes=1024 * 1024,
            max_artifact_bytes=1024 * 1024,
            allow_private_addresses=True,
            allow_http_for_testing=True,
            allowed_hosts=("store.test", "artifacts.test"),
            bearer_token_file=None,
            ca_file=None,
        ),
        paths=PathSettings(
            allowed_root=managed_root,
            include_path=managed_root / "store_plugins.py",
            requirements_path=managed_root / "store_requirements.txt",
            temp_dir=state / "tmp",
        ),
        commands=CommandSettings(
            python_path=root / "bin" / "python",
            manage_path=managed_root / "manage.py",
            systemctl_path=root / "bin" / "systemctl",
            services=("netbox", "netbox-rq"),
            command_timeout_seconds=10,
        ),
        policy=PolicySettings(
            netbox_version="4.6.8",
            min_supported_netbox="4.6.5",
            max_supported_netbox="4.6.8",
            self_plugin_slugs=("netbox-store", "netbox-plugin-store", "netbox_plugin_store"),
            allow_prereleases=False,
            require_release_for_enable=True,
        ),
    )


def release_json(version: str = "1.2.3", **overrides: Any) -> dict[str, Any]:
    result: dict[str, Any] = {
        "version": version,
        "download_url": f"http://artifacts.test/demo_plugin-{version}-py3-none-any.whl",
        "sha256": "a" * 64,
        "artifact_size": 100,
        "artifact_kind": "wheel",
        "artifact_filename": f"demo_plugin-{version}-py3-none-any.whl",
        "commit_sha": "",
        "min_netbox_version": "4.6.5",
        "max_netbox_version": "4.6.8",
        "published_at": None,
        "approved": True,
        "status": "approved",
        "immutable": True,
        "approved_payload_sha256": "c" * 64,
    }
    result.update(overrides)
    return result


def plugin_json(releases: list[dict[str, Any]] | None = None, **overrides: Any) -> dict[str, Any]:
    result: dict[str, Any] = {
        "api_version": "v1",
        "slug": "demo-plugin",
        "name": "Demo",
        "summary": "Summary",
        "description": "Description",
        "repository_url": "https://git.test/demo",
        "latest_version": "1.2.3",
        "package_name": "demo-plugin",
        "import_name": "demo_plugin",
        "min_netbox_version": "4.6.5",
        "max_netbox_version": "4.6.8",
        "approved": True,
        "status": "approved",
        "releases": releases or [],
    }
    result.update(overrides)
    return result


class FakeTransport:
    def __init__(self, responses: dict[str, Any], artifact: bytes | None = None):
        self.responses = responses
        self.artifact = artifact or b""
        self.json_urls: list[str] = []
        self.download_urls: list[str] = []

    def get_json(self, url: str, max_bytes: int) -> Any:
        self.json_urls.append(url)
        if url not in self.responses:
            raise HttpStatusError(404, "missing")
        value = self.responses[url]
        if isinstance(value, Exception):
            raise value
        assert len(json.dumps(value)) <= max_bytes
        return value

    def download(
        self,
        url: str,
        destination: Path,
        *,
        max_bytes: int,
        expected_size: int,
        expected_sha256: str,
    ) -> None:
        self.download_urls.append(url)
        if len(self.artifact) != expected_size:
            raise AssertionError("fixture size mismatch")
        if hashlib.sha256(self.artifact).hexdigest() != expected_sha256:
            raise AssertionError("fixture digest mismatch")
        destination.write_bytes(self.artifact)


def wheel_bytes(path: Path, version: str = "1.2.3") -> bytes:
    wheel = path / f"demo_plugin-{version}-py3-none-any.whl"
    with zipfile.ZipFile(wheel, "w") as archive:
        archive.writestr("demo_plugin/__init__.py", "")
        archive.writestr(
            f"demo_plugin-{version}.dist-info/WHEEL",
            "Wheel-Version: 1.0\nGenerator: tests\nRoot-Is-Purelib: true\nTag: py3-none-any\n",
        )
    return wheel.read_bytes()


def plan(version: str = "1.2.3") -> ReleasePlan:
    plugin = PluginMetadata(
        "demo-plugin", "demo-plugin", "demo_plugin", "4.6.5", "4.6.8", ()
    )
    return ReleasePlan(
        plugin,
        version,
        f"http://artifacts.test/demo_plugin-{version}-py3-none-any.whl",
        f"demo_plugin-{version}-py3-none-any.whl",
        "b" * 64,
        1,
        "c" * 64,
    )


class FakeStore:
    def __init__(self, release: ReleasePlan | None = None):
        self.release = release or plan()
        self.calls: list[tuple[str, ...]] = []

    def get_plugin(self, slug: str) -> PluginMetadata:
        self.calls.append(("plugin", slug))
        return self.release.plugin

    def get_release(self, slug: str, version: str) -> ReleasePlan:
        self.calls.append(("release", slug, version))
        return self.release

    def download_release(self, release: ReleasePlan, directory: Path) -> Path:
        self.calls.append(("download", release.version))
        target = directory / release.filename
        target.write_bytes(b"x")
        return target


class FakeRunner:
    def __init__(self, fail_step: str | None = None):
        self.commands: list[tuple[str, ...]] = []
        self.fail_step = fail_step

    def run(self, argv: list[str]) -> CommandResult:
        command = tuple(argv)
        self.commands.append(command)
        if self.fail_step and self.fail_step in command:
            raise ExecutionError("injected command failure")
        return CommandResult(command, "", "")