512 lines
26 KiB
Python
512 lines
26 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import unittest
|
|
from copy import deepcopy
|
|
from dataclasses import replace
|
|
from ipaddress import IPv4Address, IPv4Network
|
|
from unittest.mock import patch
|
|
|
|
from uvm.config import Settings
|
|
from uvm.domain import PortRule, VmRecord
|
|
from uvm.errors import CommandError, ConflictError, UvmError, ValidationError
|
|
from uvm.network import NetworkManager
|
|
from uvm.system import CommandResult
|
|
|
|
|
|
LINK_INVENTORY = ("ip", "-j", "-d", "link", "show")
|
|
|
|
|
|
class FakeRunner:
|
|
def __init__(self) -> None:
|
|
self.calls: list[tuple[str, ...]] = []
|
|
self.links: dict[str, dict] = {}
|
|
self.inventory: str | None = None
|
|
self.failures: dict[tuple[str, ...], BaseException] = {}
|
|
self.default_routes = "default via 192.0.2.1 dev eth0\n"
|
|
self.rules: dict[tuple[str, str], list[tuple[str, ...]]] = {
|
|
("nat", "POSTROUTING"): [], ("filter", "FORWARD"): [],
|
|
}
|
|
|
|
def run(self, command, *, check=True, capture=False, timeout=None) -> CommandResult:
|
|
del capture, timeout
|
|
args = tuple(str(part) for part in command)
|
|
self.calls.append(args)
|
|
if failure := self.failures.get(args):
|
|
if check or not isinstance(failure, CommandError):
|
|
raise failure
|
|
return CommandResult(args=args, returncode=1, stderr=str(failure))
|
|
if args[:5] == LINK_INVENTORY:
|
|
links = list(self.links.values()) if args == LINK_INVENTORY else [self.links[args[6]]]
|
|
stdout = self.inventory if self.inventory is not None else json.dumps(links)
|
|
return CommandResult(args=args, returncode=0, stdout=stdout)
|
|
if args[:3] == ("ip", "link", "show"):
|
|
return CommandResult(args=args, returncode=0 if args[3] in self.links else 1)
|
|
if args[:4] == ("ip", "tuntap", "add", "dev"):
|
|
if args[4] in self.links:
|
|
raise CommandError("interface already exists")
|
|
self.links[args[4]] = {
|
|
"ifname": args[4], "flags": [],
|
|
"linkinfo": {"info_kind": "tun", "info_data": {"type": "tap"}},
|
|
}
|
|
elif args[:3] == ("ip", "link", "add"):
|
|
self.links[args[3]] = {
|
|
"ifname": args[3], "flags": [],
|
|
"linkinfo": {"info_kind": "bridge", "info_data": {"stp_state": 0}},
|
|
}
|
|
elif args[:3] == ("ip", "link", "set"):
|
|
name, *options = args[4:] if args[3] == "dev" else args[3:]
|
|
link = self.links[name]
|
|
if options[0] == "alias":
|
|
link["ifalias"] = options[1]
|
|
elif options[0] == "master":
|
|
link["master"] = options[1]
|
|
elif options in (["up"], ["down"]):
|
|
link["flags"] = ["UP"] if options == ["up"] else []
|
|
elif options[:3] == ["type", "bridge_slave", "group_fwd_mask"]:
|
|
link["linkinfo"]["info_slave_kind"] = "bridge"
|
|
link["linkinfo"]["info_slave_data"] = {"group_fwd_mask": int(options[3])}
|
|
else:
|
|
raise AssertionError(f"unexpected link command: {args}")
|
|
elif args[:3] == ("ip", "link", "del"):
|
|
self.links.pop(args[3], None)
|
|
elif args[:3] == ("ip", "addr", "replace") or args[:2] == ("sysctl", "-w"):
|
|
pass
|
|
elif args[0] == "iptables":
|
|
if args[1:4] != ("--wait", "5", "-t"):
|
|
raise AssertionError(f"unexpected iptables options: {args}")
|
|
table, operation, chain, *rule = args[4:]
|
|
rules = self.rules[table, chain]
|
|
if operation == "-C":
|
|
return CommandResult(args=args, returncode=0 if tuple(rule) in rules else 1)
|
|
if operation == "-I":
|
|
position, *rule = rule
|
|
rules.insert(int(position) - 1, tuple(rule))
|
|
else:
|
|
raise AssertionError(f"unexpected iptables mutation: {args}")
|
|
elif args != ("ip", "route", "show", "default"):
|
|
raise AssertionError(f"unexpected host command: {args}")
|
|
if args == ("ip", "route", "show", "default"):
|
|
return CommandResult(args=args, returncode=0, stdout=self.default_routes)
|
|
return CommandResult(args=args, returncode=0)
|
|
|
|
|
|
class NetworkManagerTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.runner = FakeRunner()
|
|
self.network = NetworkManager(Settings(), runner=self.runner) # type: ignore[arg-type]
|
|
self.vm = VmRecord(
|
|
id="vm-0123456789abcdef0123456789abcdef",
|
|
cpu=1,
|
|
ram_mib=512,
|
|
guest_ip="10.42.0.2",
|
|
gateway="10.42.0.1",
|
|
tap="uvm-test",
|
|
mac="02:fc:00:00:00:01",
|
|
socket="/tmp/firecracker.sock",
|
|
config="/tmp/config.json",
|
|
log="/tmp/firecracker.log",
|
|
ports=(PortRule(22),),
|
|
bridge="uvm0",
|
|
)
|
|
self.tap_link = {
|
|
"ifname": self.vm.tap, "ifalias": f"uvm:{self.vm.id}", "flags": [],
|
|
"master": "uvm0",
|
|
"linkinfo": {"info_kind": "tun", "info_data": {"type": "tap"}},
|
|
}
|
|
|
|
def test_allocates_first_available_guest_address(self) -> None:
|
|
self.assertEqual(self.network.allocate_ip([], None), IPv4Address("10.42.0.2"))
|
|
|
|
def test_rejects_gateway_and_used_requested_addresses(self) -> None:
|
|
with self.assertRaises(ValidationError):
|
|
self.network.allocate_ip([self.vm], IPv4Address("10.42.0.1"))
|
|
with self.assertRaises(ValidationError):
|
|
self.network.allocate_ip([self.vm], IPv4Address("10.42.0.2"))
|
|
with self.assertRaises(ValidationError):
|
|
self.network.allocate_ip([], IPv4Address("10.42.0.0"))
|
|
with self.assertRaises(ValidationError):
|
|
self.network.allocate_ip([], IPv4Address("10.42.0.255"))
|
|
|
|
def test_mac_and_tap_names_are_bounded_and_deterministic(self) -> None:
|
|
self.assertEqual(NetworkManager.mac_for(1), "02:fc:00:00:00:01")
|
|
tap = NetworkManager.tap_name("vm-0123456789abcdef")
|
|
self.assertLessEqual(len(tap), 15)
|
|
self.assertTrue(tap.startswith("uvm-"))
|
|
|
|
def test_bridge_setup_adds_a_missing_bridge_and_one_nat_rule(self) -> None:
|
|
self.network.ensure_bridge()
|
|
|
|
self.assertIn(("ip", "link", "add", "uvm0", "type", "bridge"), self.runner.calls)
|
|
self.assertIn(
|
|
("ip", "addr", "replace", "10.42.0.1/24", "dev", "uvm0"),
|
|
self.runner.calls,
|
|
)
|
|
self.assertIn(("sysctl", "-w", "net.ipv4.ip_forward=1"), self.runner.calls)
|
|
nat, = self.runner.rules["nat", "POSTROUTING"]
|
|
self.assertEqual(nat[:6], ("-s", "10.42.0.0/24", "!", "-o", "uvm0", "-m"))
|
|
self.assertEqual(nat[-2:], ("-j", "MASQUERADE"))
|
|
inbound, outbound = self.runner.rules["filter", "FORWARD"]
|
|
self.assertEqual(outbound[:4], ("-i", "uvm0", "-s", "10.42.0.0/24"))
|
|
self.assertEqual(inbound[:8], (
|
|
"-o", "uvm0", "-d", "10.42.0.0/24", "-m", "conntrack",
|
|
"--ctstate", "ESTABLISHED,RELATED",
|
|
))
|
|
self.assertEqual(inbound[-2:], ("-j", "ACCEPT"))
|
|
self.assertEqual(outbound[-2:], ("-j", "ACCEPT"))
|
|
|
|
def test_egress_is_installed_offline_and_survives_uplink_changes_without_extra_rules(self) -> None:
|
|
self.runner.default_routes = ""
|
|
self.network.ensure_bridge()
|
|
offline_rules = deepcopy(self.runner.rules)
|
|
self.assertEqual([len(rules) for rules in offline_rules.values()], [1, 2])
|
|
|
|
for route in (
|
|
"default via 192.0.2.1 dev eth0\n",
|
|
"default via 198.51.100.1 dev wlan0\n",
|
|
"default dev tun0\ndefault via 198.51.100.1 dev wlan0 metric 600\n",
|
|
"",
|
|
):
|
|
with self.subTest(route=route):
|
|
self.runner.default_routes = route
|
|
self.runner.calls.clear()
|
|
self.network.ensure_bridge()
|
|
self.assertEqual(self.runner.rules, offline_rules)
|
|
self.assertFalse(any("-I" in call for call in self.runner.calls))
|
|
self.assertFalse(any("eth0" in rule or "wlan0" in rule or "tun0" in rule
|
|
for rules in offline_rules.values() for rule in rules))
|
|
|
|
def test_egress_exceptions_precede_existing_drops_and_preserve_foreign_and_legacy_rules(self) -> None:
|
|
existing = [
|
|
("-j", "HOST-FIREWALL"), ("-j", "DROP"),
|
|
("-i", "uvm0", "-o", "eth0", "-s", "10.42.0.0/24", "-j", "ACCEPT"),
|
|
]
|
|
self.runner.rules["filter", "FORWARD"] = list(existing)
|
|
self.network.ensure_bridge()
|
|
rules = list(self.runner.rules["filter", "FORWARD"])
|
|
self.assertEqual(rules[2:], existing)
|
|
self.assertTrue(all("uvm-egress" in rule and rule[-1] == "ACCEPT" for rule in rules[:2]))
|
|
self.network.ensure_bridge()
|
|
self.assertEqual(self.runner.rules["filter", "FORWARD"], rules)
|
|
|
|
def test_iptables_inspection_errors_are_not_treated_as_missing_rules(self) -> None:
|
|
check = ("iptables", "--wait", "5", "-t", "nat", "-C", "POSTROUTING",
|
|
"-s", "10.42.0.0/24", "!", "-o", "uvm0",
|
|
"-m", "comment", "--comment", "uvm-egress", "-j", "MASQUERADE")
|
|
with patch.object(self.runner, "run", return_value=CommandResult(check, 4, stderr="permission denied")) as run:
|
|
with self.assertRaisesRegex(CommandError, "could not inspect.*permission denied"):
|
|
self.network._ensure_masquerade()
|
|
self.assertEqual(run.call_count, 1)
|
|
|
|
def test_bridge_and_guest_network_support_non_default_prefixes(self) -> None:
|
|
settings = Settings(
|
|
network=IPv4Network("10.50.0.0/16"),
|
|
gateway=IPv4Address("10.50.0.1"),
|
|
)
|
|
runner = FakeRunner()
|
|
NetworkManager(settings, runner=runner).ensure_bridge() # type: ignore[arg-type]
|
|
|
|
self.assertIn(
|
|
("ip", "addr", "replace", "10.50.0.1/16", "dev", "uvm0"),
|
|
runner.calls,
|
|
)
|
|
for rules in runner.rules.values():
|
|
for rule in rules:
|
|
self.assertIn("10.50.0.0/16", rule)
|
|
self.assertNotIn("10.42.0.0/24", rule)
|
|
|
|
def test_refuses_to_adopt_an_existing_tap_device(self) -> None:
|
|
for alias in (None, "foreign-owner", f"uvm:{self.vm.id}"):
|
|
for owned in (False, True):
|
|
with self.subTest(alias=alias, owned=owned):
|
|
link = {**deepcopy(self.tap_link), "ifalias": alias}
|
|
self.runner.links = {self.vm.tap: link}
|
|
self.runner.calls.clear()
|
|
self.vm.tap_owned = owned
|
|
with self.assertRaisesRegex(ConflictError, "already exists"):
|
|
self.network.prepare_tap(self.vm)
|
|
self.assertEqual(self.runner.calls, [LINK_INVENTORY])
|
|
self.assertEqual(self.runner.links, {self.vm.tap: link})
|
|
self.assertIs(self.vm.tap_owned, owned)
|
|
|
|
def test_refuses_to_reconfigure_an_existing_non_bridge_interface(self) -> None:
|
|
self.runner.links["uvm0"] = {"ifname": "uvm0", "linkinfo": {"info_kind": "dummy"}}
|
|
|
|
with self.assertRaises(UvmError):
|
|
self.network.ensure_bridge()
|
|
|
|
self.assertNotIn(
|
|
("ip", "addr", "replace", "10.42.0.1/24", "dev", "uvm0"),
|
|
self.runner.calls,
|
|
)
|
|
|
|
def test_existing_stp_disabled_bridge_is_reused(self) -> None:
|
|
self.runner.links["uvm0"] = {
|
|
"ifname": "uvm0", "linkinfo": {"info_kind": "bridge", "info_data": {"stp_state": 0}},
|
|
}
|
|
self.network.ensure_bridge()
|
|
self.assertNotIn(("ip", "link", "add", "uvm0", "type", "bridge"), self.runner.calls)
|
|
self.assertIn(("ip", "addr", "replace", "10.42.0.1/24", "dev", "uvm0"), self.runner.calls)
|
|
|
|
def test_stp_enabled_bridge_is_rejected_before_any_mutation(self) -> None:
|
|
for stp_state in (1, 2):
|
|
with self.subTest(stp_state=stp_state):
|
|
self.runner.links["uvm0"] = {
|
|
"ifname": "uvm0",
|
|
"linkinfo": {"info_kind": "bridge", "info_data": {"stp_state": stp_state}},
|
|
}
|
|
self.runner.calls.clear()
|
|
with self.assertRaisesRegex(UvmError, "STP-disabled"):
|
|
self.network.ensure_bridge()
|
|
self.assertEqual(self.runner.calls, [
|
|
("ip", "link", "show", "uvm0"), (*LINK_INVENTORY, "dev", "uvm0"),
|
|
])
|
|
|
|
def test_invalid_bridge_inventory_is_rejected_before_any_mutation(self) -> None:
|
|
self.runner.links["uvm0"] = {"ifname": "uvm0"}
|
|
for inventory in ("invalid-json", "null", "{}", "[]", "[{}]"):
|
|
with self.subTest(inventory=inventory):
|
|
self.runner.inventory = inventory
|
|
self.runner.calls.clear()
|
|
with self.assertRaises(UvmError):
|
|
self.network.ensure_bridge()
|
|
self.assertEqual(self.runner.calls, [
|
|
("ip", "link", "show", "uvm0"), (*LINK_INVENTORY, "dev", "uvm0"),
|
|
])
|
|
|
|
def test_prepare_tap_sets_owner_bridge_and_lldp_mask_without_activating_it(self) -> None:
|
|
unrelated = {"ifname": f"{self.vm.tap}-other", "linkinfo": {"info_kind": "dummy"}}
|
|
for bridge in (None, "uvm-retained"):
|
|
with self.subTest(bridge=bridge):
|
|
self.runner.links = {unrelated["ifname"]: deepcopy(unrelated)}
|
|
self.runner.calls.clear()
|
|
vm = replace(self.vm, bridge=bridge, tap_owned=False)
|
|
self.network.prepare_tap(vm)
|
|
|
|
link = self.runner.links[vm.tap]
|
|
self.assertIs(vm.tap_owned, True)
|
|
self.assertEqual(link["ifalias"], f"uvm:{vm.id}")
|
|
self.assertEqual(link["master"], bridge or "uvm0")
|
|
self.assertEqual(link["linkinfo"]["info_kind"], "tun")
|
|
self.assertEqual(link["linkinfo"]["info_data"]["type"], "tap")
|
|
self.assertEqual(link["linkinfo"]["info_slave_data"]["group_fwd_mask"], 16384)
|
|
self.assertNotIn("UP", link["flags"])
|
|
self.assertFalse(any(call[-1] == "up" for call in self.runner.calls))
|
|
self.assertEqual(self.runner.links[unrelated["ifname"]], unrelated)
|
|
self.assertEqual(self.runner.calls[0], LINK_INVENTORY)
|
|
commands = (
|
|
("ip", "tuntap", "add", "dev", vm.tap, "mode", "tap"),
|
|
("ip", "link", "set", "dev", vm.tap, "alias", f"uvm:{vm.id}"),
|
|
("ip", "link", "set", vm.tap, "master", bridge or "uvm0"),
|
|
("ip", "link", "set", "dev", vm.tap, "type", "bridge_slave", "group_fwd_mask", "16384"),
|
|
)
|
|
positions = [self.runner.calls.index(command) for command in commands]
|
|
self.assertEqual(positions, sorted(positions))
|
|
|
|
def test_owned_tap_can_be_activated_deactivated_and_deleted(self) -> None:
|
|
unrelated = {"ifname": "eth0", "flags": ["UP"], "linkinfo": {"info_kind": "veth"}}
|
|
for owned in (False, True):
|
|
with self.subTest(owned=owned):
|
|
vm = replace(self.vm, tap_owned=owned)
|
|
self.runner.links = {"eth0": deepcopy(unrelated), vm.tap: deepcopy(self.tap_link)}
|
|
self.runner.calls.clear()
|
|
self.assertTrue(self.network.has_tap(vm))
|
|
self.assertEqual(self.runner.calls, [LINK_INVENTORY])
|
|
|
|
self.network.activate_tap(vm)
|
|
self.assertIn("UP", self.runner.links[vm.tap]["flags"])
|
|
self.network.deactivate_tap(vm)
|
|
self.assertNotIn("UP", self.runner.links[vm.tap]["flags"])
|
|
self.assertIs(vm.tap_owned, owned)
|
|
self.network.delete_vm_tap(vm)
|
|
self.assertNotIn(vm.tap, self.runner.links)
|
|
self.assertIs(vm.tap_owned, False)
|
|
self.assertFalse(self.network.has_tap(vm))
|
|
self.assertEqual(self.runner.links, {"eth0": unrelated})
|
|
|
|
def test_missing_tap_cannot_be_activated_and_cleanup_is_idempotent(self) -> None:
|
|
self.vm.tap_owned = True
|
|
with self.assertRaisesRegex(UvmError, "TAP device missing"):
|
|
self.network.activate_tap(self.vm)
|
|
self.assertIs(self.vm.tap_owned, True)
|
|
self.network.deactivate_tap(self.vm)
|
|
self.assertFalse(self.network.has_tap(self.vm))
|
|
self.network.delete_vm_tap(self.vm)
|
|
self.network.delete_vm_tap(self.vm)
|
|
self.assertIs(self.vm.tap_owned, False)
|
|
self.assertTrue(all(call == LINK_INVENTORY for call in self.runner.calls))
|
|
|
|
def test_named_foreign_taps_are_never_activated_deactivated_deleted_or_claimed(self) -> None:
|
|
cases = (
|
|
((PortRule(22),), True, None), ((), True, ""), ((), False, None),
|
|
(None, False, None), (None, True, "uvm:another-vm"),
|
|
((), True, "uvm:another-vm"), ((), True, self.vm.id),
|
|
((), True, f"uvm:{self.vm.id}-suffix"),
|
|
)
|
|
for ports, owned, alias in cases:
|
|
for operation in ("activate_tap", "deactivate_tap", "delete_vm_tap", "has_tap"):
|
|
with self.subTest(ports=ports, owned=owned, alias=alias, operation=operation):
|
|
vm = replace(self.vm, ports=ports, tap_owned=owned)
|
|
link = {**deepcopy(self.tap_link), "ifalias": alias}
|
|
self.runner.links = {vm.tap: link}
|
|
before = deepcopy(self.runner.links)
|
|
self.runner.calls.clear()
|
|
with self.assertRaisesRegex(ConflictError, "unowned TAP"):
|
|
getattr(self.network, operation)(vm)
|
|
self.assertEqual(self.runner.calls, [LINK_INVENTORY])
|
|
self.assertEqual(self.runner.links, before)
|
|
self.assertIs(vm.tap_owned, owned)
|
|
|
|
def test_owned_legacy_unaliased_records_can_only_manage_real_tun_taps(self) -> None:
|
|
for alias in (None, ""):
|
|
with self.subTest(alias=alias):
|
|
vm = replace(self.vm, ports=None, tap_owned=True)
|
|
link = deepcopy(self.tap_link)
|
|
if alias is None:
|
|
link.pop("ifalias")
|
|
else:
|
|
link["ifalias"] = alias
|
|
self.runner.links = {vm.tap: link}
|
|
self.assertTrue(self.network.has_tap(vm))
|
|
self.network.activate_tap(vm)
|
|
self.assertIn("UP", link["flags"])
|
|
self.network.deactivate_tap(vm)
|
|
self.assertNotIn("UP", link["flags"])
|
|
self.network.delete_vm_tap(vm)
|
|
self.assertIs(vm.tap_owned, False)
|
|
self.assertNotIn(vm.tap, self.runner.links)
|
|
|
|
def test_ownership_markers_do_not_authorize_non_tap_interfaces(self) -> None:
|
|
for info in (
|
|
{}, {"info_kind": "dummy"}, {"info_kind": "bridge"}, {"info_kind": "veth"},
|
|
{"info_kind": "tap"}, {"info_kind": "tun"},
|
|
{"info_kind": "tun", "info_data": {"type": "tun"}},
|
|
):
|
|
for ports, alias in ((None, None), ((), f"uvm:{self.vm.id}")):
|
|
for operation in ("activate_tap", "deactivate_tap", "delete_vm_tap", "has_tap"):
|
|
with self.subTest(info=info, ports=ports, operation=operation):
|
|
vm = replace(self.vm, ports=ports, tap_owned=True)
|
|
link = {"ifname": vm.tap, "ifalias": alias, "linkinfo": info}
|
|
self.runner.links = {vm.tap: link}
|
|
self.runner.calls.clear()
|
|
with self.assertRaises(ConflictError):
|
|
getattr(self.network, operation)(vm)
|
|
self.assertEqual(self.runner.calls, [LINK_INVENTORY])
|
|
self.assertIs(vm.tap_owned, True)
|
|
|
|
def test_invalid_or_ambiguous_inventory_fails_closed(self) -> None:
|
|
for inventory in (
|
|
"invalid-json", "null", "{}", "[null]", "[[]]", "[1]",
|
|
json.dumps([self.tap_link, self.tap_link]),
|
|
):
|
|
for operation in ("prepare_tap", "activate_tap", "deactivate_tap", "delete_vm_tap", "has_tap"):
|
|
with self.subTest(inventory=inventory, operation=operation):
|
|
self.runner.inventory = inventory
|
|
self.runner.calls.clear()
|
|
self.vm.tap_owned = True
|
|
with self.assertRaisesRegex(UvmError, "cannot inspect network interfaces"):
|
|
getattr(self.network, operation)(self.vm)
|
|
self.assertEqual(self.runner.calls, [LINK_INVENTORY])
|
|
self.assertIs(self.vm.tap_owned, True)
|
|
|
|
def test_invalid_interface_names_cannot_authorize_creation_or_cleanup(self) -> None:
|
|
for link in ({}, {"ifname": None}, {"ifname": 123}, {"ifname": ""}):
|
|
for operation, owned in (("prepare_tap", False), ("delete_vm_tap", True)):
|
|
with self.subTest(link=link, operation=operation):
|
|
self.runner.inventory = json.dumps([link])
|
|
self.runner.links.clear()
|
|
self.runner.calls.clear()
|
|
self.vm.tap_owned = owned
|
|
with self.assertRaises(UvmError):
|
|
getattr(self.network, operation)(self.vm)
|
|
self.assertEqual(self.runner.calls, [LINK_INVENTORY])
|
|
self.assertIs(self.vm.tap_owned, owned)
|
|
|
|
def test_malformed_tap_metadata_is_a_controlled_error_without_mutation(self) -> None:
|
|
for info in (
|
|
None, [], {"info_kind": "tun", "info_data": None},
|
|
{"info_kind": "tun", "info_data": []},
|
|
):
|
|
for operation in ("activate_tap", "delete_vm_tap"):
|
|
with self.subTest(info=info, operation=operation):
|
|
self.runner.links = {self.vm.tap: {**self.tap_link, "linkinfo": info}}
|
|
self.runner.calls.clear()
|
|
self.vm.tap_owned = True
|
|
with self.assertRaises(ConflictError):
|
|
getattr(self.network, operation)(self.vm)
|
|
self.assertEqual(self.runner.calls, [LINK_INVENTORY])
|
|
self.assertIs(self.vm.tap_owned, True)
|
|
|
|
def test_inventory_command_failure_never_mutates_a_link_or_ownership(self) -> None:
|
|
failure = CommandError("inventory unavailable")
|
|
self.runner.failures[LINK_INVENTORY] = failure
|
|
for operation in ("prepare_tap", "activate_tap", "deactivate_tap", "delete_vm_tap", "has_tap"):
|
|
with self.subTest(operation=operation):
|
|
self.runner.calls.clear()
|
|
self.vm.tap_owned = True
|
|
with self.assertRaises(CommandError) as raised:
|
|
getattr(self.network, operation)(self.vm)
|
|
self.assertIs(raised.exception, failure)
|
|
self.assertEqual(self.runner.calls, [LINK_INVENTORY])
|
|
self.assertIs(self.vm.tap_owned, True)
|
|
|
|
def test_prepare_failure_cleans_created_taps_or_retains_the_persistable_owned_flag(self) -> None:
|
|
stages = (
|
|
("ip", "tuntap", "add", "dev", self.vm.tap, "mode", "tap"),
|
|
("ip", "link", "set", "dev", self.vm.tap, "alias", f"uvm:{self.vm.id}"),
|
|
("ip", "link", "set", self.vm.tap, "master", "uvm0"),
|
|
("ip", "link", "set", "dev", self.vm.tap, "type", "bridge_slave", "group_fwd_mask", "16384"),
|
|
)
|
|
delete = ("ip", "link", "del", self.vm.tap)
|
|
for index, stage in enumerate(stages):
|
|
for cleanup_fails in (False, True):
|
|
with self.subTest(stage=stage, cleanup_fails=cleanup_fails):
|
|
vm = replace(self.vm, tap_owned=False)
|
|
self.runner.links.clear()
|
|
self.runner.calls.clear()
|
|
failure = CommandError("preparation failed")
|
|
self.runner.failures = {stage: failure}
|
|
if cleanup_fails:
|
|
self.runner.failures[delete] = CommandError("deletion failed")
|
|
with self.assertRaises(CommandError) as raised:
|
|
self.network.prepare_tap(vm)
|
|
|
|
retained = index > 0 and cleanup_fails
|
|
self.assertIs(raised.exception, failure)
|
|
self.assertIs(vm.tap_owned, retained)
|
|
restored = VmRecord.from_dict(json.loads(json.dumps(vm.to_dict())))
|
|
self.assertIs(restored.tap_owned, retained)
|
|
self.assertEqual(vm.tap in self.runner.links, retained)
|
|
self.assertEqual(delete in self.runner.calls, index > 0)
|
|
self.assertFalse(any(call[-1] == "up" for call in self.runner.calls))
|
|
|
|
def test_interrupted_preparation_cleans_up_and_reraises_the_interrupt(self) -> None:
|
|
stage = ("ip", "link", "set", self.vm.tap, "master", "uvm0")
|
|
for failure in (KeyboardInterrupt(), SystemExit(2)):
|
|
with self.subTest(failure=type(failure).__name__):
|
|
self.runner.calls.clear()
|
|
self.runner.failures = {stage: failure}
|
|
with self.assertRaises(type(failure)) as raised:
|
|
self.network.prepare_tap(self.vm)
|
|
self.assertIs(raised.exception, failure)
|
|
self.assertNotIn(self.vm.tap, self.runner.links)
|
|
self.assertIs(self.vm.tap_owned, False)
|
|
self.assertIn(("ip", "link", "del", self.vm.tap), self.runner.calls)
|
|
|
|
def test_failed_deletion_keeps_ownership_until_a_successful_retry(self) -> None:
|
|
self.vm.tap_owned = True
|
|
self.runner.links[self.vm.tap] = deepcopy(self.tap_link)
|
|
delete = ("ip", "link", "del", self.vm.tap)
|
|
self.runner.failures[delete] = CommandError("device busy")
|
|
|
|
with self.assertRaisesRegex(UvmError, "could not delete TAP.*device busy"):
|
|
self.network.delete_vm_tap(self.vm)
|
|
self.assertIn(self.vm.tap, self.runner.links)
|
|
self.assertIs(self.vm.tap_owned, True)
|
|
self.assertIs(VmRecord.from_dict(self.vm.to_dict()).tap_owned, True)
|
|
|
|
self.runner.failures.clear()
|
|
self.network.delete_vm_tap(self.vm)
|
|
self.assertNotIn(self.vm.tap, self.runner.links)
|
|
self.assertIs(self.vm.tap_owned, False)
|