#
# 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 socket
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.error import NodeError
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(f"capture stop_kernel {host_ifc}")


@pytest.mark.asyncio
async def test_start_capture_ubridge_ealready_rolls_the_flag_back(vm):
    """
    uBridge answers EALREADY when its interface is already being captured
    (the server's per-port flag should have refused earlier; this is the
    desync net): the error must surface and the port must not stay marked
    as capturing.
    """

    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
    assert nio.capturing is False


@pytest.mark.asyncio
async def test_capture_two_ports_at_once_and_stop_them_independently(vm):
    """
    uBridge serves one kernel capture per interface: two ports of the same
    node capture concurrently, and each port's stop names only its own
    anchor — a sibling port's capture is never touched.
    """

    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")
    await vm.start_capture(1, "/tmp/capture2.pcap")

    assert nio1.capturing is True
    assert nio2.capturing is True
    vm._ubridge_send.assert_any_call(f'capture start_kernel {host_ifc1} "/tmp/capture.pcap"')
    vm._ubridge_send.assert_any_call(f'capture start_kernel {host_ifc2} "/tmp/capture2.pcap"')

    await vm.stop_capture(1)

    assert nio1.capturing is True
    assert nio2.capturing is False
    vm._ubridge_send.assert_any_call(f"capture stop_kernel {host_ifc2}")
    assert not any(f"stop_kernel {host_ifc1}" in str(c) for c in vm._ubridge_send.call_args_list)

    await vm.stop_capture(0)
    vm._ubridge_send.assert_any_call(f"capture stop_kernel {host_ifc1}")


@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


# ---------------------------------------------------------------------------
# Cross-compute VXLAN tunnel
# ---------------------------------------------------------------------------

VXLAN_SPEC = {"name": "gvl01234567abcd", "vni": 1048577, "remote": "192.168.1.2", "dstport": 14789}


def test_create_nio_bridge_carries_vxlan_spec(vm):
    """
    The vxlan sub-spec of a cross-compute NIO survives the factory and the
    asdict round-trip (the node replays the NIO from asdict on restart).
    """

    nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE, "vxlan": dict(VXLAN_SPEC)})
    assert isinstance(nio, NIOBridge)
    assert nio.vxlan == VXLAN_SPEC
    assert nio.asdict()["vxlan"] == VXLAN_SPEC


@pytest.mark.asyncio
async def test_add_ubridge_kernel_connection_with_vxlan(vm):
    """
    A vxlan-carrying NIO creates this end's tunnel device (lower device
    resolved locally from the peer address) and enslaves it into the link's
    kernel bridge next to the anchor.
    """

    async def send(command):
        if command.startswith("vxlan show"):
            raise UbridgeError("VXLAN does not exist")

    vm._ubridge_hypervisor = MagicMock()
    vm._ubridge_hypervisor.send = AsyncioMagicMock(side_effect=send)
    vm._namespace = 42
    nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE, "vxlan": dict(VXLAN_SPEC)})

    with patch.object(DockerVM, "_vxlan_lowerdev", return_value="eth0"):
        await vm._add_ubridge_connection(nio, 0)

    host_ifc, _ = vm._veth_names(0, 0)
    sends = [c.args[0] for c in vm._ubridge_hypervisor.send.call_args_list]
    show = f'vxlan show "{VXLAN_SPEC["name"]}"'
    create = (
        f'vxlan create "{VXLAN_SPEC["name"]}" {VXLAN_SPEC["vni"]} '
        f"remote={VXLAN_SPEC['remote']} dstport={VXLAN_SPEC['dstport']} dev=eth0"
    )
    assert sends.index(show) < sends.index(create)
    # The tunnel device is enslaved after the node's anchor, and the anchor
    # still goes up.
    assert sends.index(f'brctl addif "{BRIDGE}" "{host_ifc}"') < sends.index(
        f'brctl addif "{BRIDGE}" "{VXLAN_SPEC["name"]}"'
    )
    assert f'link set "{host_ifc}" up' in sends
    # A failed show (device absent) must not trigger a delete.
    assert f'vxlan delete "{VXLAN_SPEC["name"]}"' not in sends


@pytest.mark.asyncio
async def test_add_ubridge_kernel_connection_replaces_vxlan_leftover(vm):
    """
    A device found under this link's deterministic name is this link's own
    leftover (a crashed run, a node restart): deleted and re-created, which
    makes wiring idempotent.
    """

    vm._ubridge_hypervisor = MagicMock()
    vm._namespace = 42
    nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE, "vxlan": dict(VXLAN_SPEC)})

    with patch.object(DockerVM, "_vxlan_lowerdev", return_value="eth0"):
        await vm._add_ubridge_connection(nio, 0)

    sends = [c.args[0] for c in vm._ubridge_hypervisor.send.call_args_list]
    show = f'vxlan show "{VXLAN_SPEC["name"]}"'
    delete = f'vxlan delete "{VXLAN_SPEC["name"]}"'
    create = (
        f'vxlan create "{VXLAN_SPEC["name"]}" {VXLAN_SPEC["vni"]} '
        f"remote={VXLAN_SPEC['remote']} dstport={VXLAN_SPEC['dstport']} dev=eth0"
    )
    assert sends.index(show) < sends.index(delete) < sends.index(create)


@pytest.mark.asyncio
async def test_remove_kernel_nio_deletes_vxlan_device(vm):
    """
    Link teardown deletes this end's tunnel device (the kernel un-enrolls it
    from the bridge on unregister) before the bridge delete.
    """

    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, "vxlan": dict(VXLAN_SPEC)})
    )

    await vm.adapter_remove_nio_binding(0)

    vm._ubridge_send.assert_has_calls(
        [
            call(f'brctl delif "{BRIDGE}" "{host_ifc}"'),
            call(f'vxlan delete "{VXLAN_SPEC["name"]}"'),
            call(f'brctl delete "{BRIDGE}"'),
        ],
        any_order=False,
    )


@pytest.mark.asyncio
async def test_vxlan_lowerdev_unresolvable_peer_raises(vm):
    """
    A peer that cannot be resolved (or routed to) is a wiring error of this
    link — the controller retries and then degrades it to the relay.
    """

    with patch("gns3server.compute.kernel_datapath.socket.getaddrinfo", side_effect=socket.gaierror("nope")):
        with pytest.raises(NodeError):
            await vm._vxlan_lowerdev("192.0.2.1")


def test_interface_with_address_matches_the_owning_interface():
    """
    The route's chosen source address is matched back to its interface,
    scope suffixes stripped (link-local IPv6).
    """

    class _Addr:
        def __init__(self, address):
            self.address = address

    with patch(
        "gns3server.compute.kernel_datapath.psutil.net_if_addrs",
        return_value={
            "eth0": [_Addr("192.168.1.1")],
            "eth1": [_Addr("fe80::1%eth1")],
        },
    ):
        assert DockerVM._interface_with_address("192.168.1.1") == "eth0"
        assert DockerVM._interface_with_address("fe80::1%eth1") == "eth1"
        assert DockerVM._interface_with_address("10.0.0.1") is None