from __future__ import annotations import json import shutil from pathlib import Path import pytest from pyro_mcp.daily_loop import DailyLoopManifest, prepare_manifest_path, write_prepare_manifest from pyro_mcp.runtime import doctor_report, resolve_runtime_paths, runtime_capabilities from pyro_mcp.vm_environments import EnvironmentStore, get_environment def _materialize_installed_environment( environment_store: EnvironmentStore, *, name: str, ) -> None: spec = get_environment(name, runtime_paths=environment_store._runtime_paths) install_dir = environment_store._install_dir(spec) install_dir.mkdir(parents=True, exist_ok=True) (install_dir / "vmlinux").write_text("kernel\n", encoding="utf-8") (install_dir / "rootfs.ext4").write_text("rootfs\n", encoding="utf-8") (install_dir / "environment.json").write_text( json.dumps( { "name": spec.name, "version": spec.version, "source": "test-cache", "source_digest": spec.source_digest, } ), encoding="utf-8", ) def test_resolve_runtime_paths_default_bundle() -> None: paths = resolve_runtime_paths() assert paths.firecracker_bin.exists() assert paths.jailer_bin.exists() assert paths.guest_agent_path is not None assert paths.guest_agent_path.exists() assert paths.guest_init_path is not None assert paths.guest_init_path.exists() assert paths.artifacts_dir.exists() assert paths.manifest.get("platform") == "linux-x86_64" def test_resolve_runtime_paths_missing_manifest( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: empty_root = tmp_path / "bundle" empty_root.mkdir(parents=True, exist_ok=True) monkeypatch.setenv("PYRO_RUNTIME_BUNDLE_DIR", str(empty_root)) with pytest.raises(RuntimeError, match="manifest not found"): resolve_runtime_paths() def test_resolve_runtime_paths_checksum_mismatch( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: source = resolve_runtime_paths() copied_bundle = tmp_path / "bundle" copied_platform = copied_bundle / "linux-x86_64" copied_platform.mkdir(parents=True, exist_ok=True) (copied_bundle / "NOTICE").write_text( source.notice_path.read_text(encoding="utf-8"), encoding="utf-8" ) manifest = json.loads(source.manifest_path.read_text(encoding="utf-8")) (copied_platform / "manifest.json").write_text( json.dumps(manifest, indent=2), encoding="utf-8", ) firecracker_path = copied_platform / "bin" / "firecracker" firecracker_path.parent.mkdir(parents=True, exist_ok=True) firecracker_path.write_text("tampered\n", encoding="utf-8") (copied_platform / "bin" / "jailer").write_bytes(source.jailer_bin.read_bytes()) guest_agent_path = source.guest_agent_path if guest_agent_path is None: raise AssertionError("expected guest agent in runtime bundle") guest_init_path = source.guest_init_path if guest_init_path is None: raise AssertionError("expected guest init in runtime bundle") copied_guest_dir = copied_platform / "guest" copied_guest_dir.mkdir(parents=True, exist_ok=True) (copied_guest_dir / "pyro_guest_agent.py").write_text( guest_agent_path.read_text(encoding="utf-8"), encoding="utf-8", ) (copied_guest_dir / "pyro-init").write_text( guest_init_path.read_text(encoding="utf-8"), encoding="utf-8", ) monkeypatch.setenv("PYRO_RUNTIME_BUNDLE_DIR", str(copied_bundle)) with pytest.raises(RuntimeError, match="checksum mismatch"): resolve_runtime_paths() def test_resolve_runtime_paths_guest_init_checksum_mismatch( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: source = resolve_runtime_paths() copied_bundle = tmp_path / "bundle" shutil.copytree(source.bundle_root.parent, copied_bundle) copied_platform = copied_bundle / "linux-x86_64" copied_guest_init = copied_platform / "guest" / "pyro-init" copied_guest_init.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8") monkeypatch.setenv("PYRO_RUNTIME_BUNDLE_DIR", str(copied_bundle)) with pytest.raises(RuntimeError, match="checksum mismatch"): resolve_runtime_paths() def test_resolve_runtime_paths_guest_init_manifest_malformed( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: source = resolve_runtime_paths() copied_bundle = tmp_path / "bundle" shutil.copytree(source.bundle_root.parent, copied_bundle) manifest_path = copied_bundle / "linux-x86_64" / "manifest.json" manifest = json.loads(manifest_path.read_text(encoding="utf-8")) guest = manifest.get("guest") if not isinstance(guest, dict): raise AssertionError("expected guest manifest section") guest["init"] = {"path": "guest/pyro-init"} manifest_path.write_text(json.dumps(manifest, indent=2), encoding="utf-8") monkeypatch.setenv("PYRO_RUNTIME_BUNDLE_DIR", str(copied_bundle)) with pytest.raises(RuntimeError, match="runtime guest init manifest entry is malformed"): resolve_runtime_paths() def test_doctor_report_has_runtime_fields() -> None: report = doctor_report() assert "runtime_ok" in report assert "kvm" in report assert "networking" in report assert "daily_loop" in report if report["runtime_ok"]: runtime = report.get("runtime") assert isinstance(runtime, dict) assert "firecracker_bin" in runtime assert "guest_agent_path" in runtime assert "guest_init_path" in runtime assert "component_versions" in runtime assert "environments" in runtime networking = report["networking"] assert isinstance(networking, dict) assert "tun_available" in networking def test_doctor_report_daily_loop_statuses( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: monkeypatch.setenv("PYRO_ENVIRONMENT_CACHE_DIR", str(tmp_path)) cold_report = doctor_report(environment="debian:12") cold_daily_loop = cold_report["daily_loop"] assert cold_daily_loop["status"] == "cold" assert cold_daily_loop["installed"] is False paths = resolve_runtime_paths() environment_store = EnvironmentStore(runtime_paths=paths, cache_dir=tmp_path) _materialize_installed_environment(environment_store, name="debian:12") installed_report = doctor_report(environment="debian:12") installed_daily_loop = installed_report["daily_loop"] assert installed_daily_loop["status"] == "cold" assert installed_daily_loop["installed"] is True manifest_path = prepare_manifest_path( tmp_path, platform="linux-x86_64", environment="debian:12", ) write_prepare_manifest( manifest_path, DailyLoopManifest( environment="debian:12", environment_version="1.0.0", platform="linux-x86_64", catalog_version=environment_store.catalog_version, bundle_version=( None if paths.manifest.get("bundle_version") is None else str(paths.manifest.get("bundle_version")) ), prepared_at=123.0, network_prepared=True, last_prepare_duration_ms=456, ), ) warm_report = doctor_report(environment="debian:12") warm_daily_loop = warm_report["daily_loop"] assert warm_daily_loop["status"] == "warm" assert warm_daily_loop["network_prepared"] is True stale_manifest = json.loads(manifest_path.read_text(encoding="utf-8")) stale_manifest["catalog_version"] = "0.0.0" manifest_path.write_text(json.dumps(stale_manifest), encoding="utf-8") stale_report = doctor_report(environment="debian:12") stale_daily_loop = stale_report["daily_loop"] assert stale_daily_loop["status"] == "stale" assert "catalog version changed" in str(stale_daily_loop["reason"]) def test_runtime_capabilities_reports_real_bundle_flags() -> None: paths = resolve_runtime_paths() capabilities = runtime_capabilities(paths) assert capabilities.supports_vm_boot is True assert capabilities.supports_guest_exec is True assert capabilities.supports_guest_network is True