marker: test enabled/pause/resume and instant toggle

- compute (test_base_node.py): set_marker_filter_state on/off command,
  marker pause/resume command, and apply issues enable_packet_filter off
  for a disabled marker (+ records the name->bridge map).
- controller (test_marker.py): _markers_for_node keeps disabled markers
  and carries enabled; update_marker enabled-only hits the toggle route
  (not NIO rebuild) while a bpf change still rebuilds; pause/resume fan
  out to capture nodes.

tests/compute/test_base_node.py

@@ -21,6 +21,7 @@ import pytest
 import pytest_asyncio
 
 from tests.utils import asyncio_patch, AsyncioMagicMock
+from unittest.mock import patch, MagicMock
 
 from gns3server.compute.vpcs.vpcs_vm import VPCSVM
 from gns3server.compute.docker.docker_vm import DockerVM
@@ -172,3 +173,57 @@ async def test_ubridge_apply_bpf_filters(node):
     node._ubridge_send.assert_any_call("bridge reset_packet_filters VPCS-10")
     node._ubridge_send.assert_any_call("bridge add_packet_filter VPCS-10 filter0 bpf \"icmp[icmptype] == 8\"")
     node._ubridge_send.assert_any_call("bridge add_packet_filter VPCS-10 filter1 bpf \"tcp src port 53\"")
+
+
+@pytest.mark.asyncio
+async def test_set_marker_filter_state_off(compute_project, manager):
+
+    node = VPCSVM("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", compute_project, manager)
+    node._ubridge_send = AsyncioMagicMock()
+    node._marker_filter_bridges["m"] = "VPCS-10"
+    await node._ubridge_set_marker_filter_state("m", False)
+    node._ubridge_send.assert_called_with("bridge enable_packet_filter VPCS-10 m off")
+
+
+@pytest.mark.asyncio
+async def test_set_marker_filter_state_on(compute_project, manager):
+
+    node = VPCSVM("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", compute_project, manager)
+    node._ubridge_send = AsyncioMagicMock()
+    node._marker_filter_bridges["m"] = "VPCS-10"
+    await node._ubridge_set_marker_filter_state("m", True)
+    node._ubridge_send.assert_called_with("bridge enable_packet_filter VPCS-10 m on")
+
+
+@pytest.mark.asyncio
+async def test_marker_pause_sends_command(compute_project, manager):
+
+    node = VPCSVM("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", compute_project, manager)
+    node._ubridge_hypervisor = AsyncioMagicMock()
+    await node._ubridge_marker_pause()
+    node._ubridge_hypervisor.send.assert_called_with("marker pause")
+
+
+@pytest.mark.asyncio
+async def test_marker_resume_sends_command(compute_project, manager):
+
+    node = VPCSVM("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", compute_project, manager)
+    node._ubridge_hypervisor = AsyncioMagicMock()
+    await node._ubridge_marker_resume()
+    node._ubridge_hypervisor.send.assert_called_with("marker resume")
+
+
+@pytest.mark.asyncio
+async def test_apply_markers_turns_disabled_filter_off(compute_project, manager):
+    # Part A: a disabled marker is installed (add_packet_filter) then turned off
+    # with enable_packet_filter … off, and its bridge is recorded for toggling.
+
+    node = VPCSVM("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", compute_project, manager)
+    node._ubridge_send = AsyncioMagicMock()
+    nio = NIOUDP(1234, "127.0.0.1", 4321)
+    nio.markers = {"m": {"bpf": "icmp", "tag": None, "link_id": "L1", "direction": None, "enabled": False}}
+    with patch("gns3server.compute.marker.marker_manager.MarkerManager") as mm:
+        mm.instance.return_value.register = MagicMock()
+        await node._ubridge_apply_markers("VPCS-10", nio)
+    node._ubridge_send.assert_any_call("bridge enable_packet_filter VPCS-10 m off")
+    assert node._marker_filter_bridges["m"] == "VPCS-10"

tests/controller/test_marker.py

@@ -461,3 +461,80 @@ async def test_start_marker_rejects_non_capable_capture_node(project):
         link._nodes.append({"node": nat, "adapter_number": 0, "port_number": 0})
         with pytest.raises(ControllerError):
             await link.start_marker("m", "icmp", capture_node_id=nat.id)
+
+
+# ---------------------------------------------------------------------------
+# Part A/B: enabled reaches uBridge + instant per-filter toggle
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_markers_for_node_keeps_disabled_and_carries_enabled(project):
+    # Part A: a disabled marker is NOT dropped from the NIO payload (so uBridge
+    # can install it then turn it off) and the spec carries `enabled`.
+    with _valid_bpf():
+        link = await _make_link(project)
+        node = link._nodes[0]["node"]
+        await link.start_marker("m", "icmp")
+        await link.update_marker("m", enabled=False)
+    spec = link._markers_for_node(node).get("m")
+    assert spec is not None
+    assert spec["enabled"] is False
+
+
+@pytest.mark.asyncio
+async def test_update_marker_enabled_only_hits_toggle_route(project):
+    # Part B: an enabled-only change routes to the per-marker toggle endpoint,
+    # not a full NIO reset+reapply.
+    with _valid_bpf():
+        link = await _make_link(project)
+        node = link._nodes[0]["node"]
+        await link.start_marker("m", "icmp")
+        compute = node.compute
+        compute.put.reset_mock()
+        await link.update_marker("m", enabled=False)
+    paths = [c.args[0] for c in compute.put.call_args_list]
+    assert any("/markers/m" in p for p in paths)
+    assert not any(p.endswith("/nio") for p in paths)
+
+
+@pytest.mark.asyncio
+async def test_update_marker_with_bpf_still_rebuilds_nio(project):
+    # A non-enabled-only change falls through to the NIO reset+reapply path.
+    with _valid_bpf():
+        link = await _make_link(project)
+        node = link._nodes[0]["node"]
+        await link.start_marker("m", "icmp")
+        compute = node.compute
+        compute.put.reset_mock()
+        await link.update_marker("m", bpf="tcp")
+    paths = [c.args[0] for c in compute.put.call_args_list]
+    assert any(p.endswith("/nio") for p in paths)
+
+
+# ---------------------------------------------------------------------------
+# Part C: project-level pause/resume fan-out
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_pause_all_markers_fans_out_to_capture_nodes(project):
+    # Part C: project-level pause hits each marker-hosting node once.
+    with _valid_bpf():
+        link = await _make_link(project)
+        await link.start_marker("m", "icmp")
+        node = link._nodes[0]["node"]
+        node.post = AsyncioMagicMock()
+        project.get_node = MagicMock(return_value=node)
+        await project.pause_all_markers()
+    node.post.assert_any_call("/markers/pause")
+
+
+@pytest.mark.asyncio
+async def test_resume_all_markers_fans_out(project):
+    with _valid_bpf():
+        link = await _make_link(project)
+        await link.start_marker("m", "icmp")
+        node = link._nodes[0]["node"]
+        node.post = AsyncioMagicMock()
+        project.get_node = MagicMock(return_value=node)
+        await project.resume_all_markers()
+    node.post.assert_any_call("/markers/resume")