gns3-server/tests/compute/dynamips/test_dynamips_kernel_datapath.py
YueGuobin c343ce0b8f
feat(esw): wire peer anchors straight into the switch bridge
The Ethernet switch's kernel fast path: instead of a link crossing the
switch's per-port uBridge relay, the switch absorbs the peer's anchor into
its own kernel bridge. The peer's interface (Docker veth host end, QEMU
TAP, IOU port TAP, Dynamips port TAP) becomes the switch port, and frames
cross anchor -> switch bridge -> switch ports entirely in the kernel.

Wire format: the switch end receives nio_anchor ('join this interface to
my own bridge' — the mirror of nio_bridge), the peer end nio_bridge with a
null bridge ('your anchor is bridged elsewhere': carrier, markers and
impairments only — never bridge membership, and never a bridge delete, on
an interface this node does not own). Anchor names come from the shared
naming contract (utils.kernel_anchor), a pure function of node type, id,
adapter and port — the controller computes the peer's anchor while the
node may be stopped, which is what makes links to stopped nodes wireable.

Stopped peers defer the join; after any node of a link starts
(UDPLink.node_started) the controller re-pushes the switch NIO, and the
switch completes — or repairs, when a peer restart replaced its interface
under the same name — the join against the kernel's actual membership
instead of assuming it.

VLAN changes now edit membership in place: only the ports whose mode/VLAN
actually changed are touched, each losing exactly the previous mode's
entries the new mode does not want (the dot1q admits-all entry is one
range operation in both directions), and no port ever leaves the bridge to
change its VLANs. This replaces the old 'reset by re-enslaving' shortcut
(delif+addif on every wired port, added when the ports were the switch's
own TAPs): on absorbed anchors, re-enslaving an interface owned by another
node is where live testing watched links die — anchors occasionally lost
their bridge membership right after a ports_mapping update (later updates
then failed with delif EINVAL), reproducibly with qemu, IOU and Dynamips
peers and never with Docker's veths.

Eligibility: the switch side needs no capability of its own (its bridge
exists by construction); the peer must be able to anchor. Switch-to-switch
cascades stay on the relay — a kernel interface belongs to one bridge, and
joining two bridges needs a veth pair, deferred.

Verified live on a real server: qemu peers through four VLAN cycles with
both anchors enslaved throughout; a mixed switch (2 IOU + 2 Dynamips, four
kernel ports); Docker peers with real ICMP isolating and recovering (with
both anchors still enslaved); VPCS relay ports likewise in place.
2026-10-06 00:21:09 +08:00

677 lines
24 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 wiring for Dynamips routers: the persistent TAP every
Ethernet slot/port owns (Ethernet adapters and WIC-1ENET ports only — serial,
ATM and POS adapters never anchor), the hypervisor TAP NIO (nio create_tap)
that holds the anchor open, per-link enslavement, the deferred wiring of
kernel links bound while the node was stopped, and the stop order (the
hypervisor's fds released before the TAPs are unpersisted — QEMU's order,
since here it is Dynamips that holds them).
"""
import os
import shutil
import contextlib
import pytest
import pytest_asyncio
from unittest.mock import MagicMock, patch
from tests.utils import AsyncioMagicMock
from gns3server.compute.dynamips import Dynamips
from gns3server.compute.dynamips.nodes.router import Router
from gns3server.compute.dynamips.dynamips_error import DynamipsError
from gns3server.compute.dynamips.adapters.gt96100_fe import GT96100_FE
from gns3server.compute.dynamips.adapters.pa_4t import PA_4T
from gns3server.compute.dynamips.adapters.pa_fe_tx import PA_FE_TX
from gns3server.compute.dynamips.adapters.wic_1enet import WIC_1ENET
from gns3server.compute.ubridge.ubridge_error import UbridgeError
from gns3server.compute.nios.nio_bridge import NIOBridge
NODE_ID = "00010203-0405-0607-0809-0a0b0c0d0e0f"
BRIDGE = "gns3a1b2c3d4e5f"
TAP00 = "gd00010203e0p0" # the router fixture's node id, slot 0 port 0
TAP01 = "gd00010203e0p1"
TAP016 = "gd00010203e0p16" # WIC-1ENET port in WIC slot 0
class FakeDynamipsHypervisor:
"""Records the hypervisor protocol; answers status queries."""
def __init__(self, status_code=2):
self.commands = []
self._status_code = status_code
def is_running(self):
return True
async def send(self, command):
self.commands.append(command)
if command.startswith("vm get_status"):
return [str(self._status_code)]
return ["100-OK"]
@pytest_asyncio.fixture
async def manager(port_manager):
m = Dynamips.instance()
m.port_manager = port_manager
return m
@pytest_asyncio.fixture
async def router(compute_project, manager):
router = Router("R1", NODE_ID, compute_project, manager)
router._hypervisor = FakeDynamipsHypervisor()
router._ubridge_hypervisor = MagicMock()
router._ubridge_hypervisor.is_running.return_value = True
router._ubridge_send = AsyncioMagicMock()
# slot 0: a 2-port Ethernet motherboard with a WIC-1ENET in WIC slot 0
# (port 16); slot 1: a serial PA that must never anchor.
router._create_slots(2)
router._slots[0] = GT96100_FE()
router._slots[0]._wics[0] = WIC_1ENET()
router._slots[1] = PA_4T()
yield router
# compute_project is the shared, id-fixed project every compute test
# file reuses (ProjectManager caches by id); leave its dynamips module
# directory as this fixture found it (absent) — a later manager test
# creates it with os.makedirs and no exist_ok.
shutil.rmtree(router._working_directory, ignore_errors=True)
with contextlib.suppress(OSError):
os.rmdir(compute_project.module_working_path(manager.module_name.lower()))
async def _nio(router, **kwargs):
settings = {"type": "nio_bridge", "bridge": BRIDGE}
settings.update(kwargs)
return await router.manager.create_nio(router, settings)
def _hypervisor_commands(router):
return list(router._hypervisor.commands)
def _ubridge_commands(router):
return [str(c[0][0]) if c[0] else "" for c in router._ubridge_send.call_args_list]
# ---------------------------------------------------------------------------
# Anchor topology: which ports anchor at all
# ---------------------------------------------------------------------------
def test_ethernet_slot_ports_skip_serial_and_include_wics(router):
"""
Only Ethernet-carrying ports anchor: the PA's two ports plus the
WIC-1ENET port (numbered from 16 per WIC slot), never the serial PA.
"""
assert router._ethernet_slot_ports() == [(0, 0), (0, 1), (0, 16)]
def test_tap_name_fits_ifnamsiz(router):
"""
The anchor name has to fit IFNAMSIZ (15) even for WIC port numbers.
"""
assert len(router._tap_name(6, 48)) <= 15
def test_kernel_host_ifc_is_none_before_start(router):
assert router._kernel_host_ifc(0, 0) is None
# ---------------------------------------------------------------------------
# Anchor lifecycle
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_prepare_tap_datapath_creates_one_tap_per_ethernet_port(router):
"""
On a capable uBridge every Ethernet slot/port gets its persistent TAP —
swept of leftovers first, ownership handed to this user (so the
unprivileged hypervisor can open it), born down.
"""
router._ubridge_send.reset_mock()
await router._prepare_tap_datapath()
assert router._tap_datapath is True
assert router._kernel_taps == {(0, 0): TAP00, (0, 1): TAP01, (0, 16): TAP016}
router._ubridge_send.assert_any_call(f'tap delete "{TAP00}"')
router._ubridge_send.assert_any_call(f'tap create "{TAP00}"')
router._ubridge_send.assert_any_call(f"tap set_owner {TAP016} {os.getuid()}")
router._ubridge_send.assert_any_call(f'link set "{TAP016}" down')
@pytest.mark.asyncio
async def test_prepare_tap_datapath_falls_back_without_the_tap_module(router):
"""
A uBridge without the tap module keeps the router on the relay datapath —
it still starts, it just cannot carry kernel links.
"""
async def refuse_probe(command):
if "prob" in command:
raise UbridgeError("Unknown command")
return ["100-OK"]
router._ubridge_send.side_effect = refuse_probe
with patch.object(router.project, "emit") as emit:
await router._prepare_tap_datapath()
assert router._tap_datapath is False
assert router._kernel_taps == {}
assert emit.called
@pytest.mark.asyncio
async def test_prepare_tap_datapath_skips_ghost_routers(compute_project, manager):
"""
Ghost IOS images share RAM with a real router and never link: no uBridge,
no anchors.
"""
ghost = Router("Ghost", NODE_ID, compute_project, manager, ghost_flag=True)
ghost._start_ubridge = AsyncioMagicMock()
await ghost._prepare_tap_datapath()
assert ghost._tap_datapath is False
assert not ghost._start_ubridge.called
@pytest.mark.asyncio
async def test_prepare_tap_datapath_keeps_existing_anchors_across_a_restart(router):
"""
Nothing is undone at node stop, so a restart must not recreate an anchor
a surviving kernel link is still enslaved to.
"""
await router._prepare_tap_datapath()
router._ubridge_send.reset_mock()
await router._prepare_tap_datapath()
assert router._kernel_taps == {(0, 0): TAP00, (0, 1): TAP01, (0, 16): TAP016}
assert not any("tap create" in c or "tap delete" in c for c in _ubridge_commands(router))
@pytest.mark.asyncio
async def test_prepare_tap_datapath_wires_nios_bound_while_stopped(router):
"""
A kernel link bound to a stopped router could only be stored (no anchors
existed); the start wires it — the hypervisor opens the anchor and binds
the port, the anchor joins the per-link bridge.
"""
router.status = "stopped"
nio = await _nio(router)
router._slots[0].add_nio(0, nio)
router._ubridge_send.reset_mock()
router._hypervisor.commands.clear()
await router._prepare_tap_datapath()
hypervisor = _hypervisor_commands(router)
assert any(c.startswith("nio create_tap tap-") and c.endswith(f"{TAP00}") for c in hypervisor)
assert any(c.startswith(f"vm slot_add_nio_binding \"R1\" 0 0 tap-") for c in hypervisor)
router._ubridge_send.assert_any_call(f'brctl addif "{BRIDGE}" "{TAP00}"')
assert (0, 0) in router._tap_nios
# ---------------------------------------------------------------------------
# Link attach / detach
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_attach_kernel_nio_opens_the_anchor_in_the_hypervisor_and_enslaves_it(router):
"""
A kernel link opens the port's anchor TAP in the Dynamips hypervisor
(Dynamips holds the fd — QEMU's shape), binds it to the slot/port, then
enslaves the anchor into the per-link kernel bridge.
"""
router.status = "started"
router._kernel_taps[(0, 0)] = TAP00
nio = await _nio(router)
await router._attach_kernel_nio(0, 0, nio)
hypervisor = _hypervisor_commands(router)
assert any(c.startswith("nio create_tap tap-") and c.endswith(f"{TAP00}") for c in hypervisor)
assert any(c.startswith('vm slot_add_nio_binding "R1" 0 0 tap-') for c in hypervisor)
router._ubridge_send.assert_any_call(f'brctl create "{BRIDGE}"')
router._ubridge_send.assert_any_call(f'link set "{BRIDGE}" up')
router._ubridge_send.assert_any_call(f'brctl addif "{BRIDGE}" "{TAP00}"')
router._ubridge_send.assert_any_call(f'link set "{TAP00}" up')
@pytest.mark.asyncio
async def test_attach_kernel_nio_without_anchor_on_a_running_node_raises(router):
"""
A kernel link on a serial port (or a router that never anchored) is an
error naming the relay datapath, not a silent fallback.
"""
router.status = "started"
nio = await _nio(router)
with pytest.raises(DynamipsError, match="relay datapath"):
await router._attach_kernel_nio(1, 0, nio)
@pytest.mark.asyncio
async def test_attach_kernel_nio_defers_on_a_stopped_node(router):
"""
On a stopped router there are no anchors yet: the NIO is stored and the
wiring happens at start (test above), so nothing is sent.
"""
router.status = "stopped"
nio = await _nio(router)
await router._attach_kernel_nio(1, 0, nio)
assert _hypervisor_commands(router) == []
assert not router._ubridge_send.called
@pytest.mark.asyncio
async def test_attach_kernel_nio_is_idempotent_across_a_restart(router):
"""
A stop tears nothing down, so a wired port must not re-bind on the next
start: a second nio create_tap would orphan the first NIO's fd and the
re-enslavement would collide on the bridge.
"""
router.status = "started"
router._kernel_taps[(0, 0)] = TAP00
nio = await _nio(router)
await router._attach_kernel_nio(0, 0, nio)
hypervisor_first = list(_hypervisor_commands(router))
router._ubridge_send.reset_mock()
await router._attach_kernel_nio(0, 0, nio)
assert _hypervisor_commands(router) == hypervisor_first
assert not router._ubridge_send.called
@pytest.mark.asyncio
async def test_add_nio_binding_relay_keeps_the_udp_tunnel(router):
"""
A relay link is wired exactly as before: the hypervisor NIO is a UDP NIO
tunneled through loopback into the node's uBridge relay.
"""
router.status = "started"
nio = await router.manager.create_nio(
router, {"type": "nio_udp", "lport": 20000, "rhost": "127.0.0.1", "rport": 20001}
)
await router.slot_add_nio_binding(0, 0, nio)
hypervisor = _hypervisor_commands(router)
assert any(c.startswith("nio create_udp udp-") for c in hypervisor)
assert any(c.startswith('vm slot_add_nio_binding "R1" 0 0 udp-') for c in hypervisor)
ubridge = _ubridge_commands(router)
assert any(c.startswith("bridge create DYNAMIPS-") for c in ubridge)
assert any(c.startswith("bridge add_nio_udp DYNAMIPS-") for c in ubridge)
assert any(c.startswith("bridge start DYNAMIPS-") for c in ubridge)
@pytest.mark.asyncio
async def test_add_nio_binding_kernel_binds_and_enslaves(router):
"""
The route-level add path dispatches a NIOBridge to the kernel wiring and
stores it in the slot adapter like any NIO.
"""
router.status = "started"
router._kernel_taps[(0, 0)] = TAP00
nio = await _nio(router)
await router.slot_add_nio_binding(0, 0, nio)
assert router._slots[0].get_nio(0) is nio
router._ubridge_send.assert_any_call(f'brctl addif "{BRIDGE}" "{TAP00}"')
@pytest.mark.asyncio
async def test_update_nio_binding_kernel_reapplies_on_the_anchor(router):
"""
A filter update on an attached kernel link reconciles on the anchor
(tc) without re-binding the port or re-enslaving the TAP.
"""
router._kernel_taps[(0, 0)] = TAP00
nio = await _nio(router, filters={"delay": [10]})
router._slots[0].add_nio(0, nio)
await router.slot_update_nio_binding(0, 0, nio)
router._ubridge_send.assert_any_call(f'tc reset "{TAP00}"')
router._ubridge_send.assert_any_call(f'tc netem set "{TAP00}" delay 10')
assert not any("slot_add_nio_binding" in c for c in _hypervisor_commands(router))
assert not any("brctl addif" in c for c in _ubridge_commands(router))
@pytest.mark.asyncio
async def test_update_nio_binding_kernel_suspend_flaps_the_anchor(router):
"""
Suspending a kernel link admin-downs the anchor — the native carrier,
which is why the route must copy suspend onto the NIO.
"""
router._kernel_taps[(0, 0)] = TAP00
nio = await _nio(router)
router._slots[0].add_nio(0, nio)
nio.suspend = True
await router.slot_update_nio_binding(0, 0, nio)
router._ubridge_send.assert_any_call(f'link set "{TAP00}" down')
@pytest.mark.asyncio
async def test_remove_nio_binding_kernel_releases_the_fd_and_detaches(router):
"""
Removing a kernel link detaches the anchor from the per-link bridge,
resets its qdisc, and unbinds the port plus deletes the hypervisor TAP
NIO — releasing Dynamips's fd while the control channels live.
"""
router.status = "started"
router._kernel_taps[(0, 0)] = TAP00
nio = await _nio(router)
router._slots[0].add_nio(0, nio)
await router._attach_kernel_nio(0, 0, nio)
router._ubridge_send.reset_mock()
router._hypervisor.commands.clear()
removed = await router.slot_remove_nio_binding(0, 0)
assert removed is nio
assert router._slots[0].get_nio(0) is None
router._ubridge_send.assert_any_call(f'tc reset "{TAP00}"')
router._ubridge_send.assert_any_call(f'brctl delif "{BRIDGE}" "{TAP00}"')
router._ubridge_send.assert_any_call(f'brctl delete "{BRIDGE}"')
hypervisor = _hypervisor_commands(router)
assert any(c.startswith('vm slot_remove_nio_binding "R1" 0 0') for c in hypervisor)
assert any(c.startswith("nio delete tap-") for c in hypervisor)
assert (0, 0) not in router._tap_nios
# ---------------------------------------------------------------------------
# Stop order (the hypervisor holds the anchor fds — QEMU's order)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_stop_ubridge_releases_hypervisor_fds_before_deleting_taps(router):
"""
The hypervisor's TAP NIOs are unbound and deleted first (closing every
anchor fd), then the TAPs are unpersisted, then the per-link bridges go —
all before uBridge itself stops.
"""
router.status = "started"
router._kernel_taps[(0, 0)] = TAP00
nio = await _nio(router)
router._slots[0].add_nio(0, nio)
await router._attach_kernel_nio(0, 0, nio)
router._ubridge_send.reset_mock()
router._hypervisor.commands.clear()
with patch(
"gns3server.compute.base_node.BaseNode._stop_ubridge", new=AsyncioMagicMock()
) as stop:
await router._stop_ubridge()
hypervisor = _hypervisor_commands(router)
ubridge = _ubridge_commands(router)
unbind = next(i for i, c in enumerate(hypervisor) if c.startswith('vm slot_remove_nio_binding "R1" 0 0'))
nio_delete = next(i for i, c in enumerate(hypervisor) if c.startswith("nio delete tap-"))
tap_delete = next(i for i, c in enumerate(ubridge) if c == f'tap delete "{TAP00}"')
brctl_delete = next(i for i, c in enumerate(ubridge) if c == f'brctl delete "{BRIDGE}"')
assert unbind < nio_delete
assert tap_delete < brctl_delete
assert stop.called
assert router._kernel_taps == {}
assert router._tap_nios == {}
assert router._ubridge_tc_caps is None
@pytest.mark.asyncio
async def test_remove_kernel_bridges_covers_wic_ports(router):
"""
The bridge sweep walks adapters derived from the anchor map, so a kernel
link on a WIC-1ENET port (inside a non-Ethernet motherboard) is found.
"""
router._kernel_taps[(0, 16)] = TAP016
wic_nio = await _nio(router)
router._slots[0].add_nio(16, wic_nio)
await router._remove_kernel_bridges()
router._ubridge_send.assert_any_call(f'brctl delete "{BRIDGE}"')
@pytest.mark.asyncio
async def test_slot_add_binding_hot_adds_anchors(router):
"""
An adapter inserted while the router runs (OIR) gets its anchors created
at once, so a kernel link can attach without a restart.
"""
router.status = "started"
router._tap_datapath = True
router._kernel_taps[(0, 0)] = TAP00
router._slots[1] = None # free the serial slot for the hot-insert
await router.slot_add_binding(1, PA_FE_TX())
assert router._kernel_taps[(1, 0)] == "gd00010203e1p0"
# ---------------------------------------------------------------------------
# Capture (kernel: the anchor; relay: the Dynamips NIO's own pcap)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_capture_on_a_kernel_link_uses_the_anchor(router):
router._kernel_taps[(0, 0)] = TAP00
nio = await _nio(router)
router._slots[0].add_nio(0, nio)
await router.start_capture(0, 0, "/tmp/test.pcap")
router._ubridge_send.assert_any_call(f'capture start_kernel {TAP00} "/tmp/test.pcap"')
assert not any("bind_filter" in c for c in _hypervisor_commands(router))
await router.stop_capture(0, 0)
router._ubridge_send.assert_any_call("capture stop_kernel")
@pytest.mark.asyncio
async def test_capture_on_a_relay_link_uses_the_dynamips_nio(router):
nio = await router.manager.create_nio(
router, {"type": "nio_udp", "lport": 20000, "rhost": "127.0.0.1", "rport": 20001}
)
router._slots[0].add_nio(0, nio)
await router.start_capture(0, 0, "/tmp/test.pcap")
assert not any("capture start_kernel" in c for c in _ubridge_commands(router))
hypervisor = _hypervisor_commands(router)
assert any(c.startswith("nio bind_filter udp-") and c.endswith(" capture") for c in hypervisor)
assert any(c.startswith("nio setup_filter udp-") for c in hypervisor)
# ---------------------------------------------------------------------------
# Markers (the polymorphic anchor: relay bridge name on relay, TAP on kernel)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_markers_on_a_kernel_link_attach_to_the_anchor(router):
"""
The marker reconcile is datapath-agnostic: with the anchor as the
location, the attach/toggle/delete primitives translate to
marker *_kernel on the TAP.
"""
router._kernel_taps[(0, 0)] = TAP00
nio = await _nio(router)
nio.markers = {
"m1": {"bpf": "vlan 100", "link_id": "lk1", "enabled": True, "data_link_type": "DLT_EN10MB"},
}
router._slots[0].add_nio(0, nio)
await router._ubridge_apply_markers(TAP00, nio)
router._ubridge_send.assert_any_call(
f'marker add_kernel m1 {TAP00} "vlan 100" link lk1 pcap "{router.project.markers_working_directory()}/'
f'{router.id}_lk1_m1.pcap"'
)
assert router._marker_filter_bridges[("m1", "lk1")] == TAP00
await router._ubridge_set_marker_filter_state("m1", False)
router._ubridge_send.assert_any_call(f"marker enable_kernel {TAP00} m1 off")
await router.delete_marker_capture("m1", "lk1", nio)
router._ubridge_send.assert_any_call(f"marker delete_kernel {TAP00} m1")
@pytest.mark.asyncio
async def test_markers_on_a_relay_link_keep_the_ubridge_shape(router):
"""
On the relay datapath the same reconcile talks to the uBridge relay
bridge behind the Dynamips UDP tunnel — unchanged behaviour.
"""
nio = await router.manager.create_nio(
router, {"type": "nio_udp", "lport": 20000, "rhost": "127.0.0.1", "rport": 20001}
)
nio.markers = {
"m1": {"bpf": "vlan 100", "link_id": "lk1", "enabled": True, "data_link_type": "DLT_EN10MB"},
}
router._slots[0].add_nio(0, nio)
tunnel_bridge = "DYNAMIPS-20000-20001"
await router._ubridge_apply_markers(tunnel_bridge, nio)
router._ubridge_send.assert_any_call(
f'bridge add_packet_filter {tunnel_bridge} m1 mark "vlan 100" link lk1 pcap '
f'"{router.project.markers_working_directory()}/{router.id}_lk1_m1.pcap"'
)
# ---------------------------------------------------------------------------
# Manager NIO construction
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_manager_create_nio_builds_the_bridge_nio(router):
"""
The Dynamips manager accepts nio_bridge settings (the kernel-datapath
NIO the controller POSTs) — without it the route 422s before the router
ever sees the link.
"""
nio = await router.manager.create_nio(
router, {"type": "nio_bridge", "bridge": BRIDGE, "filters": {"delay": [5]}, "suspend": True}
)
assert isinstance(nio, NIOBridge)
assert nio.bridge == BRIDGE
assert nio.filters == {"delay": [5]}
assert nio.suspend is True
assert _hypervisor_commands(router) == []
# ---------------------------------------------------------------------------
# Externally bridged anchors (an Ethernet switch absorbed them)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_external_bridge_nio_skips_bridge_membership(router):
"""
A NIOBridge with no bridge means an Ethernet switch owns the anchor's
bridge membership: the router still opens the anchor in the hypervisor
(the TAP needs its fd holder) but never touches any kernel bridge.
"""
router.status = "started"
router._kernel_taps[(0, 0)] = TAP00
nio = await _nio(router, bridge=None)
await router.slot_add_nio_binding(0, 0, nio)
hypervisor = _hypervisor_commands(router)
assert any(c.startswith("nio create_tap tap-") for c in hypervisor)
assert any(c.startswith('vm slot_add_nio_binding "R1" 0 0 tap-') for c in hypervisor)
assert not any("brctl" in c for c in _ubridge_commands(router))
@pytest.mark.asyncio
async def test_external_bridge_nio_remove_skips_bridge_ops(router):
"""
Detaching an externally bridged anchor releases the hypervisor's fd and
resets the link state, but must not delif or delete any bridge — the
switch owns them, and deleting the switch's bridge would take the whole
switch down.
"""
router.status = "started"
router._kernel_taps[(0, 0)] = TAP00
nio = await _nio(router, bridge=None)
await router.slot_add_nio_binding(0, 0, nio)
router._ubridge_send.reset_mock()
await router.slot_remove_nio_binding(0, 0)
commands = _ubridge_commands(router)
assert not any("brctl" in c for c in commands), commands
assert any(c == f'tc reset "{TAP00}"' for c in commands)
@pytest.mark.asyncio
async def test_remove_kernel_bridges_skips_external_nios(router):
"""
The bridge sweep never deletes a bridge this node does not own.
"""
router._kernel_taps[(0, 0)] = TAP00
external = await _nio(router, bridge=None)
router._slots[0].add_nio(0, external)
await router._remove_kernel_bridges()
assert not any("brctl" in c for c in _ubridge_commands(router))