60 lines
2.1 KiB
Python
60 lines
2.1 KiB
Python
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")
|