from __future__ import annotations import subprocess from pathlib import Path import pytest from pyro_mcp.vm_network import NetworkConfig, TapNetworkManager def test_tap_network_manager_allocation_disabled_by_default( monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.delenv("PYRO_VM_ENABLE_NETWORK", raising=False) manager = TapNetworkManager() config = manager.allocate("abcdef123456") assert manager.enabled is False assert config.tap_name == "pyroabcdef12" assert config.guest_ip.startswith("172.29.") metadata = manager.to_metadata(config) assert metadata["network_enabled"] == "false" def test_tap_network_manager_network_info() -> None: manager = TapNetworkManager(enabled=False) config = NetworkConfig( vm_id="abcdef123456", tap_name="pyroabcdef12", guest_ip="172.29.100.2", gateway_ip="172.29.100.1", subnet_cidr="172.29.100.0/24", mac_address="06:00:ab:cd:ef:12", ) info = manager.network_info(config) assert info["tap_name"] == "pyroabcdef12" assert info["outbound_connectivity_expected"] is False def test_tap_network_manager_enabled_runs_host_commands() -> None: commands: list[list[str]] = [] def runner(command: list[str]) -> subprocess.CompletedProcess[str]: commands.append(command) return subprocess.CompletedProcess(command, 0, "", "") manager = TapNetworkManager(enabled=True, runner=runner) config = manager.allocate("abcdef123456") manager.cleanup(config) assert commands[0][:4] == ["ip", "tuntap", "add", "dev"] assert commands[-1][:3] in (["ip", "link", "del"], ["nft", "delete", "table"]) def test_tap_network_manager_missing_host_support(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(Path, "exists", lambda self: False if str(self) == "/dev/net/tun" else True) manager = TapNetworkManager( enabled=True, runner=lambda command: subprocess.CompletedProcess(command, 0, "", ""), ) with pytest.raises(RuntimeError, match="/dev/net/tun"): manager.allocate("abcdef123456")