mirror of
https://github.com/GNS3/gns3-server.git
synced 2026-10-13 14:05:36 +03:00
96be37baa1
Both fixes from the code-review pass were carried by other tests' side effects: the move_to_ns failure path (the branch that used to skip the delete_veth) and the required-but-nullable bridge field on the NIO create routes.
1498 lines
54 KiB
Python
1498 lines
54 KiB
Python
#
|
|
# Copyright (C) 2026 GNS3 Technologies Inc.
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
#
|
|
|
|
"""
|
|
Kernel-datapath (NIOBridge) adapter wiring for Docker containers: veth pair
|
|
creation, per-link kernel bridge enslavement, carrier semantics, relay
|
|
fallback and cleanup.
|
|
"""
|
|
|
|
import os
|
|
import uuid
|
|
from unittest.mock import MagicMock, call, patch
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
|
|
from gns3server.compute.compute_error import ComputeError
|
|
from gns3server.compute.docker import Docker
|
|
from gns3server.compute.docker.docker_error import DockerError, DockerHttp404Error
|
|
from gns3server.compute.docker.docker_vm import DockerVM
|
|
from gns3server.compute.nios.nio_bridge import NIOBridge
|
|
from gns3server.compute.ubridge.ubridge_error import UbridgeError, UbridgeNamespaceError
|
|
from tests.utils import AsyncioMagicMock
|
|
|
|
BRIDGE = "gns3a1b2c3d4e5f"
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def manager(port_manager):
|
|
|
|
m = Docker.instance()
|
|
m.port_manager = port_manager
|
|
return m
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="function")
|
|
async def vm(compute_project, manager):
|
|
|
|
vm = DockerVM("test", str(uuid.uuid4()), compute_project, manager, "ubuntu:latest", aux_type="none")
|
|
vm._cid = "e90e34656842"
|
|
vm.mac_address = "02:42:3d:b7:93:00"
|
|
vm._start_interface_monitor = AsyncioMagicMock()
|
|
vm._stop_interface_monitor = AsyncioMagicMock()
|
|
return vm
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# NIO factory
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_create_nio_bridge(vm):
|
|
|
|
nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE})
|
|
assert isinstance(nio, NIOBridge)
|
|
assert nio.bridge == BRIDGE
|
|
assert nio.suspend is False
|
|
|
|
|
|
def test_create_nio_bridge_accepts_netem_filters(vm):
|
|
"""
|
|
Impairment filters with a tc netem equivalent ride the kernel NIO (they
|
|
become one netem qdisc on the veth host end); relay-only types are
|
|
rejected.
|
|
"""
|
|
|
|
nio = vm.manager.create_nio(
|
|
{"type": "nio_bridge", "bridge": BRIDGE, "filters": {"delay": [50, 5], "packet_loss": [10]}}
|
|
)
|
|
assert isinstance(nio, NIOBridge)
|
|
assert nio.filters == {"delay": [50, 5], "packet_loss": [10]}
|
|
|
|
|
|
def test_create_nio_bridge_accepts_bpf_filters(vm):
|
|
"""
|
|
bpf expressions run on the kernel datapath as cls_bpf match-drop
|
|
classifiers (uBridge tc bpf_drop); only frequency_drop has no kernel
|
|
equivalent yet.
|
|
"""
|
|
|
|
nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE, "filters": {"bpf": ["icmp"]}})
|
|
assert isinstance(nio, NIOBridge)
|
|
assert nio.filters == {"bpf": ["icmp"]}
|
|
|
|
|
|
def test_create_nio_bridge_accepts_frequency_drop(vm):
|
|
"""
|
|
frequency_drop runs on the kernel datapath as the eBPF classifier's
|
|
exact every-Nth mode — every filter type is accepted on a kernel NIO.
|
|
"""
|
|
|
|
nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE, "filters": {"frequency_drop": [10]}})
|
|
assert isinstance(nio, NIOBridge)
|
|
assert nio.filters == {"frequency_drop": [10]}
|
|
|
|
|
|
def test_create_nio_bridge_accepts_quota(vm):
|
|
|
|
nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE, "filters": {"quota": [1000000, 100]}})
|
|
assert isinstance(nio, NIOBridge)
|
|
assert nio.filters == {"quota": [1000000, 100]}
|
|
|
|
|
|
def test_create_nio_bridge_accepts_markers(vm):
|
|
"""
|
|
Markers ride the kernel NIO (attached to the veth host end via uBridge's
|
|
AF_PACKET marker module).
|
|
"""
|
|
|
|
nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE, "markers": {"m": {"bpf": "icmp"}}})
|
|
assert isinstance(nio, NIOBridge)
|
|
assert nio.markers == {"m": {"bpf": "icmp"}}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Adapter wiring (start path)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_ubridge_kernel_connection(vm):
|
|
|
|
vm._ubridge_hypervisor = MagicMock()
|
|
vm._namespace = 42
|
|
nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE})
|
|
|
|
await vm._add_ubridge_connection(nio, 0)
|
|
|
|
host_ifc, guest_ifc = vm._veth_names(0, 0)
|
|
calls = [
|
|
# stale-pair sweep (no-ops on a clean host)
|
|
call.send(f'docker delete_veth "{host_ifc}"'),
|
|
call.send(f'docker delete_veth "{guest_ifc}"'),
|
|
call.send(f'docker create_veth "{host_ifc}" "{guest_ifc}"'),
|
|
call.send(f'link set "{host_ifc}" down'),
|
|
call.send(f"docker set_mac_addr {guest_ifc} 02:42:3d:b7:93:00"),
|
|
call.send(f"docker move_to_ns {guest_ifc} 42 eth0"),
|
|
# link attach
|
|
call.send(f'brctl create "{BRIDGE}"'),
|
|
call.send(f'link set "{BRIDGE}" up'),
|
|
call.send(f'brctl addif "{BRIDGE}" "{host_ifc}"'),
|
|
call.send(f'link set "{host_ifc}" up'),
|
|
]
|
|
vm._ubridge_hypervisor.assert_has_calls(calls, any_order=True)
|
|
assert vm._kernel_veths[(0, 0)] == host_ifc
|
|
# no uBridge relay bridge on the kernel datapath
|
|
assert "bridge0" not in vm._bridges
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_ubridge_kernel_connection_applies_netem(vm):
|
|
"""
|
|
Filters carried by the NIO become a tc netem qdisc at start (node
|
|
restart restore), exactly like capture and markers.
|
|
"""
|
|
|
|
vm._ubridge_hypervisor = MagicMock()
|
|
vm._namespace = 42
|
|
nio = vm.manager.create_nio(
|
|
{"type": "nio_bridge", "bridge": BRIDGE, "filters": {"delay": [50, 5], "packet_loss": [10]}}
|
|
)
|
|
|
|
await vm._add_ubridge_connection(nio, 0)
|
|
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
netem_calls = [c for c in vm._ubridge_hypervisor.method_calls if "netem set" in str(c)]
|
|
assert call.send(f'tc netem set "{host_ifc}" delay 50 jitter 5 loss 10') in netem_calls
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_ubridge_connection_none_nio_creates_veth(vm):
|
|
"""
|
|
Unconnected adapters are born as veths too (the unified interface): the
|
|
interface exists inside the container with carrier off, and any link
|
|
type can be attached later — at runtime — without rebirthing it.
|
|
"""
|
|
|
|
vm._ubridge_hypervisor = MagicMock()
|
|
vm._namespace = 42
|
|
|
|
await vm._add_ubridge_connection(None, 0)
|
|
|
|
host_ifc, guest_ifc = vm._veth_names(0, 0)
|
|
vm._ubridge_hypervisor.assert_has_calls(
|
|
[
|
|
call.send(f'docker create_veth "{host_ifc}" "{guest_ifc}"'),
|
|
call.send(f'link set "{host_ifc}" down'),
|
|
call.send(f"docker move_to_ns {guest_ifc} 42 eth0"),
|
|
],
|
|
any_order=True,
|
|
)
|
|
assert vm._kernel_veths[(0, 0)] == host_ifc
|
|
# no relay bridge and no TAP for an unconnected adapter
|
|
assert "bridge0" not in vm._bridges
|
|
for c in vm._ubridge_hypervisor.method_calls:
|
|
assert "add_nio_tap" not in str(c)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_ubridge_connection_udp_relays_over_veth(vm):
|
|
"""
|
|
A relay NIO at start: the adapter is born as a veth and the uBridge
|
|
relay attaches to its host end via AF_PACKET (add_nio_ethernet) —
|
|
no TAP anywhere.
|
|
"""
|
|
|
|
vm._ubridge_hypervisor = MagicMock()
|
|
vm._namespace = 42
|
|
|
|
nio = vm.manager.create_nio({"type": "nio_udp", "lport": 4242, "rport": 4343, "rhost": "127.0.0.1"})
|
|
await vm._add_ubridge_connection(nio, 0)
|
|
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._ubridge_hypervisor.assert_has_calls(
|
|
[
|
|
call.send(f'docker create_veth "{host_ifc}" "{vm._veth_names(0, 0)[1]}"'),
|
|
call.send("bridge create bridge0"),
|
|
# the host end is born down; libpcap cannot attach to an
|
|
# admin-down interface, so the relay brings it up first
|
|
call.send(f'link set "{host_ifc}" up'),
|
|
call.send(f'bridge add_nio_ethernet bridge0 "{host_ifc}"'),
|
|
call.send("bridge add_nio_udp bridge0 4242 127.0.0.1 4343"),
|
|
call.send("bridge start bridge0"),
|
|
],
|
|
any_order=True,
|
|
)
|
|
assert "bridge0" in vm._bridges
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_ubridge_kernel_connection_cleans_up_on_failure(vm):
|
|
|
|
vm._ubridge_hypervisor = MagicMock()
|
|
vm._namespace = 42
|
|
|
|
async def failing_send(command):
|
|
if "create_veth" in command and "delete" not in command:
|
|
raise UbridgeError("could not complete netlink transaction")
|
|
|
|
vm._ubridge_send = AsyncioMagicMock(side_effect=failing_send)
|
|
|
|
with pytest.raises(UbridgeError):
|
|
await vm._add_ubridge_connection(vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE}), 0)
|
|
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._ubridge_send.assert_any_call(f'docker delete_veth "{host_ifc}"')
|
|
assert (0, 0) not in vm._kernel_veths
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_veth_cleans_up_when_the_namespace_move_fails(vm):
|
|
"""
|
|
``move_to_ns`` failing (the container died between create and move)
|
|
raises UbridgeNamespaceError — and must take the same cleanup path as
|
|
any other failure: the old branch re-raised before the delete_veth,
|
|
leaving the half-created pair behind until the next start's sweep.
|
|
"""
|
|
|
|
vm._ubridge_hypervisor = MagicMock()
|
|
vm._namespace = 42
|
|
|
|
async def failing_send(command):
|
|
if command.startswith("docker move_to_ns"):
|
|
raise UbridgeError("Could not move interface: No such process")
|
|
|
|
vm._ubridge_send = AsyncioMagicMock(side_effect=failing_send)
|
|
|
|
with pytest.raises(UbridgeNamespaceError):
|
|
await vm._create_veth(0)
|
|
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._ubridge_send.assert_any_call(f'docker delete_veth "{host_ifc}"')
|
|
assert vm._kernel_veths == {}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Link attach / detach
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connect_nio_kernel_tolerates_bridge_create_race(vm):
|
|
|
|
# Both endpoints create the per-link bridge concurrently; the loser must
|
|
# verify (brctl show) instead of failing.
|
|
async def send(command):
|
|
if "brctl create" in command:
|
|
raise UbridgeError("Could not create bridge: File exists")
|
|
|
|
vm._ubridge_send = AsyncioMagicMock(side_effect=send)
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc
|
|
|
|
nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE})
|
|
await vm._connect_nio(0, nio)
|
|
|
|
vm._ubridge_send.assert_any_call(f'brctl show "{BRIDGE}"')
|
|
vm._ubridge_send.assert_any_call(f'brctl addif "{BRIDGE}" "{host_ifc}"')
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connect_nio_kernel_eexist_still_reapplies_everything(vm):
|
|
"""
|
|
EEXIST is a *common* path now: a bridge survives a single-sided node
|
|
restart (the peer's still-enslaved port makes brctl delete answer EBUSY,
|
|
which is suppressed), so the restarted side finds its bridge already
|
|
there. The create-or-verify step must not short-circuit the rest of the
|
|
attach — the restarted side has a brand-new anchor and needs its capture,
|
|
markers and filters re-applied on it.
|
|
"""
|
|
|
|
async def send(command):
|
|
if "brctl create" in command:
|
|
raise UbridgeError("Could not create bridge gns3a1b2c3d4e5f: File exists (EEXIST)")
|
|
|
|
vm._ubridge_send = AsyncioMagicMock(side_effect=send)
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc
|
|
nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE})
|
|
nio.filters = {"delay": [10]}
|
|
|
|
await vm._connect_nio(0, nio)
|
|
|
|
vm._ubridge_send.assert_any_call(f'brctl show "{BRIDGE}"')
|
|
vm._ubridge_send.assert_any_call(f'brctl addif "{BRIDGE}" "{host_ifc}"')
|
|
# ... and the port-level configuration of the fresh anchor goes on top of
|
|
# it, unconditionally (never inside a "the bridge was created" branch).
|
|
vm._ubridge_send.assert_any_call(f'tc reset "{host_ifc}"')
|
|
vm._ubridge_send.assert_any_call(f'tc netem set "{host_ifc}" delay 10')
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connect_nio_kernel_passive_anchor_skips_impairments(vm):
|
|
"""
|
|
A NIOBridge with no bridge of its own (bridge is None) is the absorbed
|
|
end of an Ethernet switch link: the anchor belongs to the switch, which
|
|
owns its tc state and applies the link's filters there (the controller
|
|
routes them to the switch end, so this NIO always carries {}). The
|
|
passive end must not reconcile impairments — an empty filter set is a
|
|
tc reset, which would silently wipe the switch's qdisc off the shared
|
|
anchor at every node start.
|
|
"""
|
|
|
|
vm._ubridge_hypervisor = MagicMock()
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc
|
|
nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": None})
|
|
|
|
await vm._connect_nio(0, nio)
|
|
|
|
tc_calls = [c.args[0] for c in vm._ubridge_send.call_args_list if str(c.args[0]).startswith("tc ")]
|
|
assert tc_calls == [], tc_calls
|
|
# The membership is the switch's as well: no per-link bridge work here.
|
|
assert not any("brctl" in c.args[0] for c in vm._ubridge_send.call_args_list)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connect_nio_kernel_create_failure_propagates(vm):
|
|
|
|
async def send(command):
|
|
if "brctl create" in command or "brctl show" in command:
|
|
raise UbridgeError("Could not create bridge: No such device")
|
|
|
|
vm._ubridge_send = AsyncioMagicMock(side_effect=send)
|
|
vm._kernel_veths[(0, 0)] = vm._veth_names(0, 0)[0]
|
|
|
|
nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE})
|
|
with pytest.raises(UbridgeError):
|
|
await vm._connect_nio(0, nio)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connect_nio_kernel_without_veth_raises(vm):
|
|
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE})
|
|
with pytest.raises(DockerError, match="restart the node"):
|
|
await vm._connect_nio(0, nio)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_remove_kernel_nio(vm):
|
|
|
|
vm._ubridge_hypervisor = MagicMock()
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
vm.status = "started"
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc
|
|
vm._ethernet_adapters[0].add_nio(0, vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE}))
|
|
|
|
await vm.adapter_remove_nio_binding(0)
|
|
|
|
vm._ubridge_send.assert_has_calls(
|
|
[
|
|
call(f'tc reset "{host_ifc}"'),
|
|
call(f'link set "{host_ifc}" down'),
|
|
call(f'brctl delif "{BRIDGE}" "{host_ifc}"'),
|
|
call(f'brctl delete "{BRIDGE}"'),
|
|
],
|
|
any_order=True,
|
|
)
|
|
assert vm._ethernet_adapters[0].get_nio(0) is None
|
|
# the veth itself stays: it is the adapter interface, not the link
|
|
assert vm._kernel_veths[(0, 0)] == host_ifc
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_remove_kernel_nio_tolerates_peer_winning_bridge_delete(vm):
|
|
|
|
# Deleting a bridge that still holds the peer's port fails with EBUSY;
|
|
# a bridge already deleted by the peer fails with ENOENT — both expected.
|
|
vm._ubridge_hypervisor = MagicMock()
|
|
|
|
async def send(command):
|
|
if "brctl delif" in command or "brctl delete" in command:
|
|
raise UbridgeError("busy")
|
|
|
|
vm._ubridge_send = AsyncioMagicMock(side_effect=send)
|
|
vm._kernel_veths[(0, 0)] = vm._veth_names(0, 0)[0]
|
|
vm._ethernet_adapters[0].add_nio(0, vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE}))
|
|
|
|
await vm.adapter_remove_nio_binding(0) # must not raise
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Carrier / suspend
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_adapter_carrier_kernel(vm):
|
|
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc
|
|
|
|
await vm._set_adapter_carrier(0, True)
|
|
await vm._set_adapter_carrier(0, False)
|
|
|
|
vm._ubridge_send.assert_has_calls(
|
|
[
|
|
call(f'link set "{host_ifc}" up'),
|
|
call(f'link set "{host_ifc}" down'),
|
|
]
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# UDP fallback on a veth adapter (kernel link deleted, relay link created
|
|
# on a running node)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connect_nio_udp_relays_over_veth(vm):
|
|
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc
|
|
|
|
nio = vm.manager.create_nio({"type": "nio_udp", "lport": 4242, "rport": 4343, "rhost": "127.0.0.1"})
|
|
await vm._connect_nio(0, nio)
|
|
|
|
vm._ubridge_send.assert_has_calls(
|
|
[
|
|
call("bridge create bridge0"),
|
|
call(f'bridge add_nio_ethernet bridge0 "{host_ifc}"'),
|
|
call("bridge add_nio_udp bridge0 4242 127.0.0.1 4343"),
|
|
call("bridge start bridge0"),
|
|
],
|
|
any_order=True,
|
|
)
|
|
assert "bridge0" in vm._bridges
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connect_nio_udp_on_veth_is_idempotent(vm):
|
|
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc
|
|
vm._bridges.add("bridge0") # relay already exists from a previous link
|
|
|
|
nio = vm.manager.create_nio({"type": "nio_udp", "lport": 4242, "rport": 4343, "rhost": "127.0.0.1"})
|
|
await vm._connect_nio(0, nio)
|
|
|
|
for c in vm._ubridge_send.call_args_list:
|
|
command = c.args[0]
|
|
assert "bridge create" not in command
|
|
assert "add_nio_ethernet" not in command
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Guards and cleanup
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_nio_kernel_applies_ebpf_drops(vm):
|
|
"""
|
|
frequency_drop becomes the eBPF every-Nth mode and quota the byte-cap
|
|
mode, applied after the netem/bpf pass on the same NIO update.
|
|
"""
|
|
|
|
vm._ubridge_hypervisor = _caps_hypervisor(ebpf="1")
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
vm.status = "started"
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc
|
|
nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE})
|
|
vm._ethernet_adapters[0].add_nio(0, nio)
|
|
nio.filters = {"frequency_drop": [3], "quota": [1000000, 50]}
|
|
|
|
await vm.adapter_update_nio_binding(0, nio)
|
|
vm._ubridge_send.assert_any_call(f'tc nth_drop "{host_ifc}" 3')
|
|
vm._ubridge_send.assert_any_call(f'tc quota_drop "{host_ifc}" 1000000 50')
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_nio_kernel_ebpf_drop_everything(vm):
|
|
"""
|
|
frequency_drop -1 (drop everything) maps to every 1st packet.
|
|
"""
|
|
|
|
vm._ubridge_hypervisor = _caps_hypervisor(ebpf="1")
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
vm.status = "started"
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc
|
|
nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE})
|
|
vm._ethernet_adapters[0].add_nio(0, nio)
|
|
nio.filters = {"frequency_drop": [-1]}
|
|
|
|
await vm.adapter_update_nio_binding(0, nio)
|
|
vm._ubridge_send.assert_any_call(f'tc nth_drop "{host_ifc}" 1')
|
|
# the quota mode is explicitly off (not left to chance)
|
|
vm._ubridge_send.assert_any_call(f'tc quota_drop "{host_ifc}" off')
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_nio_kernel_turns_ebpf_modes_off(vm):
|
|
"""
|
|
A re-apply with the stateful filters removed must turn both modes off
|
|
(the netem reset already detached the program; the offs are idempotent).
|
|
"""
|
|
|
|
vm._ubridge_hypervisor = _caps_hypervisor(ebpf="1")
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
vm.status = "started"
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc
|
|
nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE})
|
|
vm._ethernet_adapters[0].add_nio(0, nio)
|
|
nio.filters = {"frequency_drop": [3]}
|
|
|
|
await vm.adapter_update_nio_binding(0, nio)
|
|
vm._ubridge_send.reset_mock()
|
|
nio.filters = {"delay": [50]}
|
|
await vm.adapter_update_nio_binding(0, nio)
|
|
|
|
vm._ubridge_send.assert_any_call(f'tc nth_drop "{host_ifc}" off')
|
|
vm._ubridge_send.assert_any_call(f'tc quota_drop "{host_ifc}" off')
|
|
vm._ubridge_send.assert_any_call(f'tc window_drop "{host_ifc}" off')
|
|
for c in vm._ubridge_send.call_args_list:
|
|
assert "nth_drop" not in c.args[0] or "off" in c.args[0]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_nio_kernel_passive_anchor_skips_impairments(vm):
|
|
"""
|
|
The link-update reconcile on the absorbed (passive) end must leave the
|
|
anchor's tc state to the switch: no tc reset / netem / drop classifiers
|
|
from this end, or the reset would wipe what the switch applied moments
|
|
earlier in the same link update (switch-first links PUT this side last).
|
|
The carrier (suspend) is still driven here, and markers still reconcile
|
|
(this end may host some).
|
|
"""
|
|
|
|
vm._ubridge_hypervisor = MagicMock()
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
vm.status = "started"
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc
|
|
nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": None})
|
|
vm._ethernet_adapters[0].add_nio(0, nio)
|
|
|
|
await vm.adapter_update_nio_binding(0, nio)
|
|
|
|
tc_calls = [c.args[0] for c in vm._ubridge_send.call_args_list if str(c.args[0]).startswith("tc ")]
|
|
assert tc_calls == [], tc_calls
|
|
# The carrier still follows the suspend flag on the shared anchor.
|
|
vm._ubridge_send.assert_any_call(f'link set "{host_ifc}" up')
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_ebpf_drops_requires_ebpf_capability(vm):
|
|
"""
|
|
An old or non-capped uBridge (ebpf=0) gets a clear upgrade error naming
|
|
the setcap fix.
|
|
"""
|
|
|
|
hyp = MagicMock()
|
|
hyp.is_running.return_value = True
|
|
hyp.send = AsyncioMagicMock(return_value=["netem=delay;ebpf=0;cbpf=1"])
|
|
vm._ubridge_hypervisor = hyp
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
|
|
with pytest.raises(DockerError, match="setcap cap_bpf"):
|
|
await vm._ubridge_apply_ebpf_drops(host_ifc, {"frequency_drop": [3]})
|
|
vm._ubridge_send.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_nio_kernel_applies_window_drop(vm):
|
|
"""
|
|
window_drop becomes the eBPF time-window mode: the positional
|
|
[start, outage, chance, period, jitter] maps 1:1 onto the tc command
|
|
(trailing optionals omitted), and an absent window is turned off.
|
|
"""
|
|
|
|
vm._ubridge_hypervisor = _caps_hypervisor(ebpf="1")
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
vm.status = "started"
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc
|
|
nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE})
|
|
vm._ethernet_adapters[0].add_nio(0, nio)
|
|
|
|
nio.filters = {"window_drop": [0, 2000, 100]}
|
|
await vm.adapter_update_nio_binding(0, nio)
|
|
vm._ubridge_send.assert_any_call(f'tc window_drop "{host_ifc}" 0 2000 100')
|
|
|
|
vm._ubridge_send.reset_mock()
|
|
nio.filters = {"window_drop": [500, 800, 100, 2400, 300]}
|
|
await vm.adapter_update_nio_binding(0, nio)
|
|
vm._ubridge_send.assert_any_call(f'tc window_drop "{host_ifc}" 500 800 100 2400 300')
|
|
|
|
# only the window mode present: nth/quota are explicitly off
|
|
vm._ubridge_send.assert_any_call(f'tc nth_drop "{host_ifc}" off')
|
|
vm._ubridge_send.assert_any_call(f'tc quota_drop "{host_ifc}" off')
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_window_drop_requires_ebpf_capability(vm):
|
|
"""
|
|
The window mode shares the eBPF capability gate (error names setcap).
|
|
"""
|
|
|
|
hyp = MagicMock()
|
|
hyp.is_running.return_value = True
|
|
hyp.send = AsyncioMagicMock(return_value=["netem=delay;ebpf=0;cbpf=1"])
|
|
vm._ubridge_hypervisor = hyp
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
|
|
with pytest.raises(DockerError, match="setcap cap_bpf"):
|
|
await vm._ubridge_apply_ebpf_drops(host_ifc, {"window_drop": [0, 2000, 100]})
|
|
vm._ubridge_send.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_window_drop_requires_mode_token(vm):
|
|
"""
|
|
ebpf=1 is not enough: a build that does not declare the window mode
|
|
token (pre-correction feature/tc-precision semantics) must get a clear
|
|
upgrade error instead of silently installing back-to-back windows.
|
|
"""
|
|
|
|
vm._ubridge_hypervisor = _caps_hypervisor(ebpf="1", modes="nth,quota")
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
|
|
with pytest.raises(DockerError, match="ebpf_modes=nth,quota"):
|
|
await vm._ubridge_apply_ebpf_drops(host_ifc, {"window_drop": [0, 2000, 100]})
|
|
vm._ubridge_send.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_ebpf_drops_legacy_build_keeps_shipped_modes(vm):
|
|
"""
|
|
A build predating the ebpf_modes field (feature/tc-precision) keeps the
|
|
modes that shipped with it: nth/quota still apply, window does not.
|
|
"""
|
|
|
|
vm._ubridge_hypervisor = _caps_hypervisor(ebpf="1", modes=None)
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
|
|
await vm._ubridge_apply_ebpf_drops(host_ifc, {"frequency_drop": [3]})
|
|
vm._ubridge_send.assert_any_call(f'tc nth_drop "{host_ifc}" 3')
|
|
vm._ubridge_send.assert_any_call(f'tc quota_drop "{host_ifc}" off')
|
|
|
|
vm._ubridge_send.reset_mock()
|
|
with pytest.raises(DockerError, match="newer uBridge"):
|
|
await vm._ubridge_apply_ebpf_drops(host_ifc, {"window_drop": [0, 2000, 100]})
|
|
vm._ubridge_send.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mode_off_skips_undeclared_modes(vm):
|
|
"""
|
|
The cleanup path only sends "off" for modes this build declares — an
|
|
undeclared mode's command would be an unknown command on that build.
|
|
"""
|
|
|
|
vm._ubridge_hypervisor = _caps_hypervisor(ebpf="1", modes="nth")
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
|
|
# A first apply with a filter forces the capability probe; the second
|
|
# (filters removed) takes the cached caps down the off path.
|
|
await vm._ubridge_apply_ebpf_drops(host_ifc, {"frequency_drop": [3]})
|
|
vm._ubridge_send.reset_mock()
|
|
await vm._ubridge_apply_ebpf_drops(host_ifc, {})
|
|
|
|
vm._ubridge_send.assert_any_call(f'tc nth_drop "{host_ifc}" off')
|
|
sent = [call.args[0] for call in vm._ubridge_send.call_args_list]
|
|
assert not any("quota_drop" in command or "window_drop" in command for command in sent)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_ebpf_drops_skips_old_ubridge_without_filters(vm):
|
|
"""
|
|
No stateful filters and no capability probe yet: nothing was ever
|
|
installed on an old uBridge, so it must not be touched at all.
|
|
"""
|
|
|
|
vm._ubridge_hypervisor = MagicMock()
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
|
|
await vm._ubridge_apply_ebpf_drops(host_ifc, {})
|
|
vm._ubridge_send.assert_not_called()
|
|
vm._ubridge_hypervisor.send.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_ebpf_drops_cleanup_only_probes_with_ebpf(vm):
|
|
"""
|
|
The filter-removal path (modes off) only talks to the eBPF commands
|
|
when the capability probe reported them — a netem-ext uBridge without
|
|
the eBPF module must not see an unknown command.
|
|
"""
|
|
|
|
hyp = MagicMock()
|
|
hyp.is_running.return_value = True
|
|
hyp.send = AsyncioMagicMock(return_value=["netem=delay,rate;ebpf=0;cbpf=1"])
|
|
vm._ubridge_hypervisor = hyp
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._ubridge_tc_caps = None # force the probe (something was set before)
|
|
|
|
await vm._ubridge_apply_ebpf_drops(host_ifc, {})
|
|
vm._ubridge_send.assert_not_called()
|
|
|
|
|
|
def _caps_hypervisor(cbpf="1", ebpf="0", modes="nth,quota,window"):
|
|
"""
|
|
A hypervisor mock whose direct send() answers `tc capabilities`.
|
|
modes=None emulates a build predating the ebpf_modes field.
|
|
"""
|
|
|
|
reply = f"netem=delay;ebpf={ebpf};cbpf={cbpf}"
|
|
if modes is not None:
|
|
reply += f";ebpf_modes={modes}"
|
|
hyp = MagicMock()
|
|
hyp.is_running.return_value = True
|
|
hyp.send = AsyncioMagicMock(return_value=[reply])
|
|
return hyp
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_nio_kernel_applies_bpf_drops(vm):
|
|
"""
|
|
bpf lines reconcile as flush + one bpf_drop add per line at successive
|
|
priorities (any line matching drops — OR semantics like the relay).
|
|
"""
|
|
|
|
vm._ubridge_hypervisor = _caps_hypervisor()
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
vm.status = "started"
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc
|
|
nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE})
|
|
vm._ethernet_adapters[0].add_nio(0, nio)
|
|
nio.filters = {"bpf": ["icmp\nudp port 53"]}
|
|
|
|
await vm.adapter_update_nio_binding(0, nio)
|
|
vm._ubridge_send.assert_any_call(f'tc bpf_drop flush "{host_ifc}"')
|
|
vm._ubridge_send.assert_any_call(f'tc bpf_drop add "{host_ifc}" 10 "icmp"')
|
|
vm._ubridge_send.assert_any_call(f'tc bpf_drop add "{host_ifc}" 11 "udp port 53"')
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_nio_kernel_flushes_bpf_drops_when_removed(vm):
|
|
|
|
vm._ubridge_hypervisor = _caps_hypervisor()
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
vm.status = "started"
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc
|
|
nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE})
|
|
vm._ethernet_adapters[0].add_nio(0, nio)
|
|
nio.filters = {"bpf": ["icmp"]}
|
|
|
|
await vm.adapter_update_nio_binding(0, nio)
|
|
vm._ubridge_send.reset_mock()
|
|
nio.filters = {}
|
|
await vm.adapter_update_nio_binding(0, nio)
|
|
vm._ubridge_send.assert_any_call(f'tc bpf_drop flush "{host_ifc}"')
|
|
for c in vm._ubridge_send.call_args_list:
|
|
assert "bpf_drop add" not in c.args[0]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_bpf_drops_requires_cbpf_capability(vm):
|
|
|
|
vm._ubridge_hypervisor = _caps_hypervisor(cbpf="0")
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
|
|
with pytest.raises(DockerError, match="upgrade uBridge"):
|
|
await vm._ubridge_apply_bpf_drops(host_ifc, {"bpf": ["icmp"]})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_bpf_drops_compile_failure_warns(vm):
|
|
"""
|
|
A line that compiles for the controller's tcpdump but not for this
|
|
uBridge's libpcap is skipped with a warning — mirrors the relay path.
|
|
"""
|
|
|
|
vm._ubridge_hypervisor = _caps_hypervisor()
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
|
|
async def send(command):
|
|
if "bpf_drop add" in command:
|
|
raise UbridgeError("209-Cannot compile filter 'icmp': can't parse filter expression: syntax error")
|
|
|
|
vm._ubridge_send = AsyncioMagicMock(side_effect=send)
|
|
with patch("gns3server.compute.kernel_datapath.log.warning") as mock_log:
|
|
await vm._ubridge_apply_bpf_drops(host_ifc, {"bpf": ["icmp"]}) # must not raise
|
|
assert mock_log.called
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_bpf_drops_skips_old_ubridge_without_lines(vm):
|
|
"""
|
|
No bpf lines and no capability probe yet: nothing was ever installed on
|
|
an old uBridge, so it must not be touched at all (no unknown-module error).
|
|
"""
|
|
|
|
vm._ubridge_hypervisor = MagicMock()
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
|
|
await vm._ubridge_apply_bpf_drops(host_ifc, {})
|
|
vm._ubridge_send.assert_not_called()
|
|
vm._ubridge_hypervisor.send.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ubridge_tc_capabilities_probed_once(vm):
|
|
|
|
vm._ubridge_hypervisor = _caps_hypervisor()
|
|
caps = await vm._ubridge_tc_capabilities()
|
|
assert caps == {"netem": "delay", "ebpf": "0", "cbpf": "1", "ebpf_modes": "nth,quota,window"}
|
|
await vm._ubridge_tc_capabilities()
|
|
assert vm._ubridge_hypervisor.send.call_count == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_nio_kernel_applies_netem(vm):
|
|
|
|
vm._ubridge_hypervisor = MagicMock()
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
vm.status = "started"
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc
|
|
nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE})
|
|
vm._ethernet_adapters[0].add_nio(0, nio)
|
|
nio.filters = {"delay": [50, 5], "packet_loss": [10], "corrupt": [2]}
|
|
|
|
await vm.adapter_update_nio_binding(0, nio)
|
|
vm._ubridge_send.assert_any_call(f'tc netem set "{host_ifc}" delay 50 jitter 5 loss 10 corrupt 2')
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_nio_kernel_clears_netem(vm):
|
|
"""
|
|
Filters dropped to empty must detach the qdisc (the veth survives link
|
|
changes); ENOENT (no qdisc was attached) is tolerated.
|
|
"""
|
|
|
|
vm._ubridge_hypervisor = MagicMock()
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
vm.status = "started"
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc
|
|
nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE})
|
|
vm._ethernet_adapters[0].add_nio(0, nio)
|
|
nio.filters = {}
|
|
|
|
await vm.adapter_update_nio_binding(0, nio)
|
|
vm._ubridge_send.assert_any_call(f'tc reset "{host_ifc}"')
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_netem_reset_tolerates_missing_qdisc(vm):
|
|
|
|
vm._ubridge_hypervisor = MagicMock()
|
|
|
|
async def refuse(command):
|
|
raise UbridgeError("207-Could not reset qdisc on gv0: No such file or directory")
|
|
|
|
vm._ubridge_send = refuse
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
# ENOENT means "already clean" — must not raise
|
|
await vm._ubridge_apply_netem(host_ifc, {})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_netem_reset_reraises_real_errors(vm):
|
|
|
|
vm._ubridge_hypervisor = MagicMock()
|
|
|
|
async def refuse(command):
|
|
raise UbridgeError("206-Could not reset qdisc on gv0: Operation not permitted")
|
|
|
|
vm._ubridge_send = refuse
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
with pytest.raises(UbridgeError):
|
|
await vm._ubridge_apply_netem(host_ifc, {})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_nio_kernel_suspend_toggles_carrier(vm):
|
|
|
|
vm._ubridge_hypervisor = MagicMock()
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
vm.status = "started"
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc
|
|
nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE})
|
|
vm._ethernet_adapters[0].add_nio(0, nio)
|
|
|
|
nio.suspend = True
|
|
await vm.adapter_update_nio_binding(0, nio)
|
|
vm._ubridge_send.assert_any_call(f'link set "{host_ifc}" down')
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_capture_kernel_uses_af_packet(vm):
|
|
|
|
vm._ubridge_hypervisor = MagicMock()
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
vm.status = "started"
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc
|
|
nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE})
|
|
vm._ethernet_adapters[0].add_nio(0, nio)
|
|
|
|
await vm.start_capture(0, "/tmp/capture.pcap")
|
|
|
|
assert nio.capturing is True
|
|
vm._ubridge_send.assert_any_call(f'capture start_kernel {host_ifc} "/tmp/capture.pcap"')
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_capture_kernel(vm):
|
|
|
|
vm._ubridge_hypervisor = MagicMock()
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
vm.status = "started"
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc
|
|
nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE})
|
|
vm._ethernet_adapters[0].add_nio(0, nio)
|
|
await vm.start_capture(0, "/tmp/capture.pcap")
|
|
|
|
await vm.stop_capture(0)
|
|
|
|
assert nio.capturing is False
|
|
vm._ubridge_send.assert_any_call("capture stop_kernel")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_capture_second_concurrent_fails(vm):
|
|
"""
|
|
uBridge's kernel capture is a singleton per process: a second concurrent
|
|
capture returns EALREADY and must surface to the caller.
|
|
"""
|
|
|
|
async def send(command):
|
|
if "capture start_kernel" in command:
|
|
raise UbridgeError("Could not start kernel capture: Operation already in progress")
|
|
|
|
vm._ubridge_hypervisor = MagicMock()
|
|
vm._ubridge_send = AsyncioMagicMock(side_effect=send)
|
|
vm.status = "started"
|
|
vm._kernel_veths[(0, 0)] = vm._veth_names(0, 0)[0]
|
|
nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE})
|
|
vm._ethernet_adapters[0].add_nio(0, nio)
|
|
|
|
with pytest.raises(UbridgeError, match="already in progress"):
|
|
await vm.start_capture(0, "/tmp/capture.pcap")
|
|
|
|
# the failed start must not leave the port marked as capturing: a stuck
|
|
# flag refuses the retry and lets this port's stop kill the winner
|
|
assert nio.capturing is False
|
|
assert vm._kernel_capture_ifc is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_capture_second_port_on_the_same_node_is_refused(vm):
|
|
"""
|
|
uBridge serves one kernel capture per process while the server models
|
|
captures per port: a second port's capture must be refused with a clear
|
|
error and its flag rolled back — otherwise its stop would issue the
|
|
process-wide, argument-less ``capture stop_kernel`` and kill the first
|
|
port's capture while its own flag still says capturing.
|
|
"""
|
|
|
|
vm._ubridge_hypervisor = MagicMock()
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
vm.status = "started"
|
|
vm._ethernet_adapters.append(type(vm._ethernet_adapters[0])())
|
|
host_ifc1, _ = vm._veth_names(0, 0)
|
|
host_ifc2, _ = vm._veth_names(1, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc1
|
|
vm._kernel_veths[(1, 0)] = host_ifc2
|
|
nio1 = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE})
|
|
nio2 = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE})
|
|
vm._ethernet_adapters[0].add_nio(0, nio1)
|
|
vm._ethernet_adapters[1].add_nio(0, nio2)
|
|
|
|
await vm.start_capture(0, "/tmp/capture.pcap")
|
|
|
|
with pytest.raises(DockerError, match="one kernel capture per node"):
|
|
await vm.start_capture(1, "/tmp/capture2.pcap")
|
|
|
|
assert nio1.capturing is True
|
|
assert nio2.capturing is False
|
|
vm._ubridge_send.assert_any_call(f'capture start_kernel {host_ifc1} "/tmp/capture.pcap"')
|
|
assert not any("stop_kernel" in str(c) for c in vm._ubridge_send.call_args_list)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_capture_on_a_port_that_does_not_own_the_slot_skips_stop_kernel(vm):
|
|
"""
|
|
``capture stop_kernel`` is process-wide and takes no interface: a stop on
|
|
a port whose flag says capturing while another port owns the slot must
|
|
not reach uBridge — it would stop the owner's capture.
|
|
"""
|
|
|
|
vm._ubridge_hypervisor = MagicMock()
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
vm.status = "started"
|
|
vm._ethernet_adapters.append(type(vm._ethernet_adapters[0])())
|
|
host_ifc1, _ = vm._veth_names(0, 0)
|
|
host_ifc2, _ = vm._veth_names(1, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc1
|
|
vm._kernel_veths[(1, 0)] = host_ifc2
|
|
nio1 = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE})
|
|
nio2 = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE})
|
|
vm._ethernet_adapters[0].add_nio(0, nio1)
|
|
vm._ethernet_adapters[1].add_nio(0, nio2)
|
|
|
|
await vm.start_capture(0, "/tmp/capture.pcap") # port 0 owns the slot
|
|
nio2.start_packet_capture("/tmp/capture2.pcap") # a stale second flag
|
|
|
|
await vm.stop_capture(1)
|
|
|
|
assert nio2.capturing is False
|
|
assert not any("stop_kernel" in str(c) for c in vm._ubridge_send.call_args_list)
|
|
|
|
# the owner's capture is untouched and still stoppable
|
|
await vm.stop_capture(0)
|
|
vm._ubridge_send.assert_any_call("capture stop_kernel")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connect_nio_kernel_restarts_capture(vm):
|
|
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc
|
|
|
|
nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE})
|
|
nio.start_packet_capture("/tmp/capture.pcap")
|
|
await vm._connect_nio(0, nio)
|
|
|
|
vm._ubridge_send.assert_any_call(f'capture start_kernel {host_ifc} "/tmp/capture.pcap"')
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Markers (AF_PACKET taps on the veth host end)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _marker_spec(bpf="icmp", tag=1, enabled=True, direction=None, link_id="l1"):
|
|
return {
|
|
"bpf": bpf,
|
|
"tag": tag,
|
|
"enabled": enabled,
|
|
"direction": direction,
|
|
"data_link_type": "DLT_EN10MB",
|
|
"link_id": link_id,
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connect_nio_kernel_applies_markers(vm):
|
|
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc
|
|
|
|
nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE, "markers": {"icmp-m": _marker_spec()}})
|
|
await vm._connect_nio(0, nio)
|
|
|
|
pcap = os.path.join(vm.project.markers_working_directory(), f"{vm.id}_l1_icmp-m.pcap")
|
|
vm._ubridge_send.assert_any_call(f'marker add_kernel icmp-m {host_ifc} "icmp" tag 1 link l1 pcap "{pcap}"')
|
|
assert vm._marker_filter_bridges[("icmp-m", "l1")] == host_ifc
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connect_nio_kernel_marker_disabled_at_apply(vm):
|
|
"""
|
|
A disabled marker is installed but silenced (marker enable_kernel off) so
|
|
the UI can flip it back on instantly — same contract as the relay path.
|
|
"""
|
|
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc
|
|
|
|
nio = vm.manager.create_nio(
|
|
{"type": "nio_bridge", "bridge": BRIDGE, "markers": {"icmp-m": _marker_spec(enabled=False)}}
|
|
)
|
|
await vm._connect_nio(0, nio)
|
|
|
|
vm._ubridge_send.assert_any_call(f"marker enable_kernel {host_ifc} icmp-m off")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_nio_kernel_applies_markers(vm):
|
|
|
|
vm._ubridge_hypervisor = MagicMock()
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
vm.status = "started"
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc
|
|
nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE})
|
|
vm._ethernet_adapters[0].add_nio(0, nio)
|
|
|
|
nio.markers = {"tcp-m": _marker_spec(bpf="tcp", tag=2, link_id="l2")}
|
|
await vm.adapter_update_nio_binding(0, nio)
|
|
|
|
pcap = os.path.join(vm.project.markers_working_directory(), f"{vm.id}_l2_tcp-m.pcap")
|
|
vm._ubridge_send.assert_any_call(f'marker add_kernel tcp-m {host_ifc} "tcp" tag 2 link l2 pcap "{pcap}"')
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_marker_toggle_kernel_uses_enable_kernel(vm):
|
|
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc
|
|
vm._marker_filter_bridges[("icmp-m", "l1")] = host_ifc
|
|
|
|
await vm._ubridge_set_marker_filter_state("icmp-m", False)
|
|
vm._ubridge_send.assert_any_call(f"marker enable_kernel {host_ifc} icmp-m off")
|
|
await vm._ubridge_set_marker_filter_state("icmp-m", True)
|
|
vm._ubridge_send.assert_any_call(f"marker enable_kernel {host_ifc} icmp-m on")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_remove_kernel_nio_deletes_markers(vm):
|
|
"""
|
|
The veth survives link deletion (it is the adapter interface), so kernel
|
|
markers must be torn down explicitly — otherwise a deleted link's markers
|
|
would keep sniffing and signaling.
|
|
"""
|
|
|
|
vm._ubridge_hypervisor = MagicMock()
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
vm.status = "started"
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc
|
|
vm._marker_filter_bridges[("icmp-m", "l1")] = host_ifc
|
|
vm._marker_specs[("icmp-m", "l1")] = _marker_spec()
|
|
vm._ethernet_adapters[0].add_nio(0, vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE}))
|
|
|
|
await vm.adapter_remove_nio_binding(0)
|
|
|
|
vm._ubridge_send.assert_any_call(f"marker delete_kernel {host_ifc} icmp-m")
|
|
assert ("icmp-m", "l1") not in vm._marker_filter_bridges
|
|
assert ("icmp-m", "l1") not in vm._marker_specs
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rebuild_marker_filter_kernel(vm):
|
|
"""
|
|
The fine-grained rebuild path (bpf/tag/direction change) goes through the
|
|
overridden primitives: delete_kernel + add_kernel with the new expression.
|
|
"""
|
|
|
|
vm._ubridge_hypervisor = MagicMock()
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc
|
|
vm._marker_filter_bridges[("icmp-m", "l1")] = host_ifc
|
|
|
|
await vm.rebuild_marker_filter("icmp-m", "l1", "tcp", tag=9)
|
|
|
|
pcap = os.path.join(vm.project.markers_working_directory(), f"{vm.id}_l1_icmp-m.pcap")
|
|
vm._ubridge_send.assert_any_call(f"marker delete_kernel {host_ifc} icmp-m")
|
|
vm._ubridge_send.assert_any_call(f'marker add_kernel icmp-m {host_ifc} "tcp" tag 9 link l1 pcap "{pcap}"')
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_remove_kernel_veths(vm):
|
|
|
|
vm._ubridge_hypervisor = MagicMock()
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc
|
|
|
|
await vm._remove_kernel_veths()
|
|
|
|
vm._ubridge_send.assert_called_once_with(f'docker delete_veth "{host_ifc}"')
|
|
assert vm._kernel_veths == {}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_remove_kernel_veths_deletes_per_link_bridge(vm):
|
|
"""
|
|
The per-link kernel bridge dies with the node too: deleting a project
|
|
never runs the link-teardown path, so without this the bridge stays
|
|
behind as an empty orphan (EBUSY from the peer's port is suppressed —
|
|
the last endpoint wins).
|
|
"""
|
|
|
|
vm._ubridge_hypervisor = MagicMock()
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc
|
|
nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE})
|
|
vm._ethernet_adapters[0].add_nio(0, nio)
|
|
# a relay NIO's bridge is not this node's to delete (uBridge dies with it)
|
|
relay = vm.manager.create_nio({"type": "nio_udp", "lport": 10000, "rhost": "127.0.0.1", "rport": 10001})
|
|
vm._ethernet_adapters[0].add_nio(1, relay)
|
|
|
|
await vm._remove_kernel_veths()
|
|
|
|
vm._ubridge_send.assert_any_call(f'docker delete_veth "{host_ifc}"')
|
|
# exactly the kernel bridge — the relay NIO's uBridge bridge dies with
|
|
# the uBridge process and must not be touched here
|
|
deletes = [call.args[0] for call in vm._ubridge_send.call_args_list if "brctl delete" in call.args[0]]
|
|
assert deletes == [f'brctl delete "{BRIDGE}"']
|
|
|
|
|
|
def test_veth_names_deterministic_and_bounded(vm):
|
|
|
|
host_ifc, guest_ifc = vm._veth_names(3, 2)
|
|
assert host_ifc == f"gv{vm.id.replace('-', '')[:8]}e3p2"
|
|
assert guest_ifc == f"gc{vm.id.replace('-', '')[:8]}e3p2"
|
|
assert host_ifc != guest_ifc
|
|
# multi-port adapters can reach two-digit adapter/port numbers
|
|
long_host, long_guest = vm._veth_names(12, 8)
|
|
assert len(long_host) <= 15 and len(long_guest) <= 15
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# P6a: netem extensions (rate / reorder / gemodel / duplicate / seed / limit /
|
|
# distribution / correlation) — translation and capability gating
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _caps_hypervisor_netem(tokens="delay,jitter,loss,dup,corrupt,rate,reorder,gemodel,dist,seed,limit"):
|
|
"""A hypervisor mock whose direct send() answers `tc capabilities` with
|
|
the full netem-extension surface."""
|
|
hyp = MagicMock()
|
|
hyp.is_running.return_value = True
|
|
hyp.send = AsyncioMagicMock(return_value=[f"netem={tokens};ebpf=1;cbpf=1"])
|
|
return hyp
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_nio_kernel_applies_netem_extensions(vm):
|
|
"""
|
|
The netem-extension filters merge into the single netem qdisc command,
|
|
in the frozen grammar order.
|
|
"""
|
|
|
|
vm._ubridge_hypervisor = _caps_hypervisor_netem()
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
vm.status = "started"
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc
|
|
nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE})
|
|
vm._ethernet_adapters[0].add_nio(0, nio)
|
|
nio.filters = {
|
|
"delay": [100, 20, "normal"],
|
|
"packet_loss": [5, 25],
|
|
"duplicate": [10, 50],
|
|
"corrupt": [2],
|
|
"reorder": [25, 50, 5],
|
|
"rate": ["512kbit"],
|
|
"limit": [5000],
|
|
"seed": [42],
|
|
}
|
|
|
|
await vm.adapter_update_nio_binding(0, nio)
|
|
vm._ubridge_send.assert_any_call(
|
|
f'tc netem set "{host_ifc}" delay 100 jitter 20 loss 5 correl 25 dup 10 correl 50 corrupt 2 '
|
|
"reorder 25 correl 50 gap 5 rate 512kbit limit 5000 distribution normal seed 42"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_nio_kernel_applies_gemodel(vm):
|
|
"""
|
|
gemodel replaces the plain loss keyword and is mutually exclusive with
|
|
packet_loss (the controller validates; the translation just prefers
|
|
gemodel when both somehow arrive).
|
|
"""
|
|
|
|
vm._ubridge_hypervisor = _caps_hypervisor_netem()
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
vm.status = "started"
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
vm._kernel_veths[(0, 0)] = host_ifc
|
|
nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE})
|
|
vm._ethernet_adapters[0].add_nio(0, nio)
|
|
nio.filters = {"delay": [50, 5], "gemodel": [100, 0, 30]}
|
|
|
|
await vm.adapter_update_nio_binding(0, nio)
|
|
vm._ubridge_send.assert_any_call(f'tc netem set "{host_ifc}" delay 50 jitter 5 loss gemodel 100 0 30')
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_netem_extension_needs_capability(vm):
|
|
"""
|
|
Extension keywords on an old uBridge (no tc capabilities / no netem-ext
|
|
tokens) raise a clear upgrade error instead of a raw parse failure.
|
|
"""
|
|
|
|
# old uBridge: no tc module at all -> empty capabilities
|
|
vm._ubridge_hypervisor = MagicMock()
|
|
vm._ubridge_hypervisor.is_running.return_value = True
|
|
vm._ubridge_hypervisor.send = AsyncioMagicMock(side_effect=UbridgeError("201-Unknown command"))
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
|
|
with pytest.raises(DockerError, match="netem extensions"):
|
|
await vm._ubridge_apply_netem(host_ifc, {"rate": ["512kbit"]})
|
|
vm._ubridge_send.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_netem_extension_partial_capability(vm):
|
|
"""
|
|
A uBridge that reports some but not all needed tokens fails on the
|
|
missing ones (per-token check, not just build detection).
|
|
"""
|
|
|
|
vm._ubridge_hypervisor = _caps_hypervisor_netem(tokens="delay,jitter,loss,dup,corrupt")
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
|
|
with pytest.raises(DockerError, match="rate, reorder"):
|
|
await vm._ubridge_apply_netem(host_ifc, {"delay": [50], "reorder": [25], "rate": ["1mbit"]})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_netem_plain_surface_never_probes(vm):
|
|
"""
|
|
Original-surface filters (delay/loss/corrupt, and plain dup) must not
|
|
touch the capability probe at all — an old uBridge serves them as-is.
|
|
The reset-before-set is unconditional (kernel netem replace merges
|
|
optional attrs, so a removed parameter would otherwise leak).
|
|
"""
|
|
|
|
vm._ubridge_hypervisor = MagicMock()
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
|
|
await vm._ubridge_apply_netem(host_ifc, {"delay": [50, 5], "packet_loss": [10], "duplicate": [10], "corrupt": [2]})
|
|
assert vm._ubridge_send.call_args_list == [
|
|
call(f'tc reset "{host_ifc}"'),
|
|
call(f'tc netem set "{host_ifc}" delay 50 jitter 5 loss 10 dup 10 corrupt 2'),
|
|
]
|
|
vm._ubridge_hypervisor.send.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_netem_reset_leak_on_removal(vm):
|
|
"""
|
|
A re-apply with a parameter REMOVED (rate gone, delay kept) must reset
|
|
first: the kernel's netem change merges optional attributes, so a bare
|
|
replace would keep the old rate shaping forever.
|
|
"""
|
|
|
|
vm._ubridge_hypervisor = _caps_hypervisor_netem()
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
|
|
await vm._ubridge_apply_netem(host_ifc, {"delay": [50], "rate": ["1mbit"]})
|
|
vm._ubridge_send.reset_mock()
|
|
await vm._ubridge_apply_netem(host_ifc, {"delay": [50]})
|
|
|
|
assert vm._ubridge_send.call_args_list == [
|
|
call(f'tc reset "{host_ifc}"'),
|
|
call(f'tc netem set "{host_ifc}" delay 50'),
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_netem_correl_gates_on_extension_build(vm):
|
|
"""
|
|
Loss/dup correlation is an extension (the capabilities string has no
|
|
correl token) — it gates on the netem-extension build marker "rate".
|
|
"""
|
|
|
|
vm._ubridge_hypervisor = _caps_hypervisor_netem(tokens="delay,jitter,loss,dup,corrupt")
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
host_ifc, _ = vm._veth_names(0, 0)
|
|
|
|
with pytest.raises(DockerError, match="netem extensions"):
|
|
await vm._ubridge_apply_netem(host_ifc, {"packet_loss": [5, 25]})
|
|
|
|
|
|
def test_create_nio_bridge_accepts_netem_extension_filters(vm):
|
|
|
|
nio = vm.manager.create_nio(
|
|
{"type": "nio_bridge", "bridge": BRIDGE, "filters": {"rate": ["512kbit"], "reorder": [25, 0, 5]}}
|
|
)
|
|
assert isinstance(nio, NIOBridge)
|
|
assert nio.filters == {"rate": ["512kbit"], "reorder": [25, 0, 5]}
|
|
|
|
|
|
def test_create_nio_udp_rejects_kernel_only_filters(vm):
|
|
"""
|
|
The relay NIO rejects netem-extension filter types (second guard behind
|
|
the controller).
|
|
"""
|
|
|
|
with pytest.raises(ComputeError, match="kernel-datapath"):
|
|
vm.manager.create_nio(
|
|
{
|
|
"type": "nio_udp",
|
|
"lport": 4242,
|
|
"rport": 4343,
|
|
"rhost": "127.0.0.1",
|
|
"filters": {"gemodel": [100, 0, 30]},
|
|
}
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_clears_relay_bridge_registry(vm):
|
|
"""
|
|
The uBridge process dies with every bridge it hosted — stop() must drop
|
|
the relay bridge names, or the next start skips "bridge create" (the
|
|
gate checks the name's presence) and fails on add_nio_udp.
|
|
"""
|
|
|
|
vm._ubridge_hypervisor = MagicMock()
|
|
vm._ubridge_send = AsyncioMagicMock()
|
|
vm._bridges.add("bridge1")
|
|
vm._ubridge_tc_caps = {"netem": "delay"}
|
|
|
|
async def query(method, path, *args, **kwargs):
|
|
raise DockerHttp404Error("no such container")
|
|
|
|
vm.manager.query = AsyncioMagicMock(side_effect=query)
|
|
await vm.stop()
|
|
assert vm._bridges == set()
|
|
assert vm._ubridge_tc_caps is None
|