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, "", "")