diff --git a/tests/e2e/README.md b/tests/e2e/README.md
new file mode 100644
index 000000000..1addb9d2f
--- /dev/null
+++ b/tests/e2e/README.md
@@ -0,0 +1,46 @@
+
+
+# Live end-to-end tests
+
+Tests under this directory run against a **real** server: real uBridge, real
+kernel interfaces (privileged), real emulator binaries and real images. They
+verify what unit tests cannot — that a datapath actually forwards traffic,
+that kernel bridges actually enslave anchors, that tc impairments actually
+shape packets. They are marked `e2e` and **excluded from normal test runs**
+(see `tests/pytest.ini`); run them explicitly:
+
+```bash
+pytest -m e2e tests/e2e -s # -s shows the progress prints
+pytest -m e2e tests/e2e -k dynamips # one scenario
+```
+
+Two run modes, chosen per test through `harness.live_server(kernel=...)`:
+
+* **Target an existing server** — set `GNS3_E2E_URL` (e.g.
+ `http://127.0.0.1:3080/v3`, plus `GNS3_E2E_USER`/`GNS3_E2E_PASSWORD`,
+ default admin/admin). The server must already be configured for the
+ datapath under test (kernel links need `enable_kernel_datapath = True`).
+* **Isolated instance** (the default) — the harness starts its own server
+ process on a free port with its own config directory (the controller DB
+ and every path follow the config), so nothing on the machine is touched.
+ It needs `ubridge`, `dynamips` and an images directory
+ (`GNS3_E2E_IMAGES`, default `~/GNS3/images`) available locally.
+
+Either way each test creates its own project and deletes it again. Set
+`GNS3_E2E_KEEP=1` to keep the project (and a started instance) on failure
+and print where to look at it.
+
+`harness.py` holds the shared pieces — REST client, server lifecycle, the
+IOS console driver, host-side kernel inspections (bridge membership, tap
+admin state, tc qdiscs) — so a new node type adds only its own scenario
+module.
+
+Current scenarios:
+
+* `test_dynamips_kernel_datapath.py` — two real c7200 routers: kernel link
+ (anchors, per-link bridge, netem delay, suspend, capture, delete/re-create,
+ stop/start) plus a serial link staying on the relay, and a negative
+ control on an isolated relay-configured instance.
diff --git a/tests/e2e/__init__.py b/tests/e2e/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/e2e/harness.py b/tests/e2e/harness.py
new file mode 100644
index 000000000..d1690f2d0
--- /dev/null
+++ b/tests/e2e/harness.py
@@ -0,0 +1,526 @@
+#
+# 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 .
+
+"""
+Shared harness for live end-to-end tests (pytest marker ``e2e``).
+
+These tests drive a **real** gns3-server process against real uBridge, real
+kernel interfaces and real emulator binaries — things a unit-test run cannot
+provide (privileges, images, a live kernel). They exist to verify what mocks
+cannot: that kernel bridges actually enslave anchors, that tc impairments
+actually shape traffic, that a datapath actually forwards. Run them with::
+
+ pytest -m e2e tests/e2e -s
+
+The harness targets an **already running server** when ``GNS3_E2E_URL`` is
+set (``GNS3_E2E_USER`` / ``GNS3_E2E_PASSWORD`` default to admin/admin), and
+otherwise starts an **isolated instance** — its own config dir, controller
+DB, project paths and free port, all under a temporary directory that is
+removed on exit. Either way every test creates its own project and deletes
+it again (``GNS3_E2E_KEEP=1`` keeps the project for inspection).
+
+Environment knobs:
+
+* ``GNS3_E2E_URL`` — target server, e.g. ``http://127.0.0.1:3080/v3``;
+ when unset, an isolated instance is started for each test.
+* ``GNS3_E2E_USER`` / ``GNS3_E2E_PASSWORD`` — credentials for the target.
+* ``GNS3_E2E_IMAGES`` — images directory the isolated instance uses
+ (default ``~/GNS3/images``); also where the dynamips test finds its image.
+* ``GNS3_E2E_KEEP=1`` — keep the project (and the isolated instance) when a
+ test fails, and print their ids, for inspection.
+"""
+
+import json
+import os
+import re
+import shutil
+import socket
+import subprocess
+import sys
+import tempfile
+import time
+import urllib.error
+import urllib.request
+
+import pytest
+
+REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+# tc lives in /usr/sbin on most distributions; subprocesses do not always
+# inherit a PATH that has it.
+TC = shutil.which("tc", path=os.environ.get("PATH", "") + ":/usr/sbin:/sbin")
+
+# IOS prompt as the last thing on the stream: any hostname/mode shape.
+PROMPT_RE = re.compile(r"[A-Za-z0-9\-_.()/]+[>#]\s*$")
+
+# Test-only default credentials (every real deployment overrides them via
+# GNS3_E2E_USER / GNS3_E2E_PASSWORD); a freshly started isolated instance
+# seeds exactly these.
+DEFAULT_USER = "admin"
+DEFAULT_PASSWORD = "admin"
+
+
+def keep_on_failure():
+ return os.environ.get("GNS3_E2E_KEEP") == "1"
+
+
+# ---------------------------------------------------------------------------
+# REST client
+# ---------------------------------------------------------------------------
+
+
+class Compute:
+ """REST client for one server (controller API), token-authenticated."""
+
+ def __init__(self, url, username=DEFAULT_USER, password=DEFAULT_PASSWORD):
+ self.url = url.rstrip("/")
+ self.username = username
+ self.password = password
+ self.token = None
+
+ def call(self, method, path, body=None, timeout=60):
+ # S310: the base URL comes from the operator (GNS3_E2E_URL or the
+ # isolated instance this harness just started), not untrusted input
+ data = json.dumps(body).encode() if body is not None else None
+ req = urllib.request.Request(self.url + path, data=data, method=method.upper()) # noqa: S310
+ req.add_header("Content-Type", "application/json")
+ if self.token:
+ req.add_header("Authorization", f"Bearer {self.token}")
+ try:
+ with urllib.request.urlopen(req, timeout=timeout) as resp: # noqa: S310
+ payload = resp.read()
+ return json.loads(payload) if payload else None
+ except urllib.error.HTTPError as e:
+ detail = e.read().decode(errors="replace")
+ raise AssertionError(f"{method} {path} -> {e.code}: {detail}") from None
+
+ def login(self):
+ body = self.call("POST", "/access/users/authenticate", {"username": self.username, "password": self.password})
+ self.token = body["access_token"]
+ return self
+
+ # -- common controller operations ------------------------------------
+
+ def create_project(self, name="e2e"):
+ return self.call("POST", "/projects", {"name": f"{name}-{int(time.time())}"})
+
+ def create_dynamips_router(self, project_id, name, image, slots, platform="c7200", ram=512):
+ properties = {"platform": platform, "image": image, "ram": ram}
+ properties.update(slots)
+ return self.call(
+ "POST",
+ f"/projects/{project_id}/nodes",
+ {"compute_id": "local", "name": name, "node_type": "dynamips", "properties": properties},
+ )
+
+ def create_link(self, project_id, a, b):
+ return self.call(
+ "POST",
+ f"/projects/{project_id}/links",
+ {
+ "nodes": [
+ {"node_id": a[0], "adapter_number": a[1], "port_number": a[2]},
+ {"node_id": b[0], "adapter_number": b[1], "port_number": b[2]},
+ ]
+ },
+ )
+
+ def delete_project(self, project_id):
+ self.call("DELETE", f"/projects/{project_id}")
+
+ def node(self, project_id, node_id):
+ return self.call("GET", f"/projects/{project_id}/nodes/{node_id}")
+
+ def capabilities(self):
+ return self.call("GET", "/computes/local").get("capabilities") or {}
+
+ def dynamips_images(self):
+ return [image["filename"] for image in self.call("GET", "/computes/local/dynamips/images")]
+
+
+# ---------------------------------------------------------------------------
+# Server lifecycle
+# ---------------------------------------------------------------------------
+
+ISOLATED_CONFIG = """\
+[Controller]
+jwt_secret_key = 00000000000000000000000000000000000000000000000000000000000000e2
+default_admin_username = admin
+default_admin_password = admin
+
+[Server]
+host = 127.0.0.1
+port = {port}
+ubridge_path = {ubridge}
+enable_kernel_datapath = {kernel}
+report_errors = False
+projects_path = {tmp}/projects
+configs_path = {tmp}/configs
+images_path = {images}
+appliances_path = {tmp}/appliances
+symbols_path = {tmp}/symbols
+
+[Dynamips]
+dynamips_path = {dynamips}
+images_path = {images}
+allocate_aux_console_ports = False
+mmap_support = False
+sparse_memory_support = False
+ghost_ios_support = False
+"""
+
+
+def _free_port():
+ with socket.socket() as sock:
+ sock.bind(("127.0.0.1", 0))
+ return sock.getsockname()[1]
+
+
+class Server:
+ """A server under test: an external one, or a freshly started isolated
+ instance (own config dir — the controller DB and every path follow it —
+ under a temporary directory removed on stop)."""
+
+ def __init__(self, compute, process=None, tmpdir=None):
+ self.compute = compute
+ self.process = process
+ self.tmpdir = tmpdir
+
+ @property
+ def isolated(self):
+ return self.process is not None
+
+ def log_tail(self, lines=40):
+ if not self.tmpdir:
+ return ""
+ path = os.path.join(self.tmpdir, "server.log")
+ if not os.path.exists(path):
+ return ""
+ with open(path) as f:
+ return "".join(f.readlines()[-lines:])
+
+ def stop(self):
+ if self.process and self.process.poll() is None:
+ self.process.terminate()
+ try:
+ self.process.wait(timeout=15)
+ except subprocess.TimeoutExpired:
+ self.process.kill()
+ self.process.wait(timeout=10)
+ if self.tmpdir:
+ shutil.rmtree(self.tmpdir, ignore_errors=True)
+
+
+def _start_isolated(kernel):
+ ubridge = shutil.which("ubridge")
+ dynamips = shutil.which("dynamips")
+ if not ubridge:
+ pytest.skip("ubridge is not in PATH (needed to start an isolated instance)")
+ if not dynamips:
+ pytest.skip("dynamips is not in PATH (needed to start an isolated instance)")
+ images = os.environ.get("GNS3_E2E_IMAGES", os.path.expanduser("~/GNS3/images"))
+ if not os.path.isdir(images):
+ pytest.skip(f"images directory not found: {images} (set GNS3_E2E_IMAGES)")
+
+ tmpdir = tempfile.mkdtemp(prefix="gns3-e2e-")
+ os.makedirs(os.path.join(tmpdir, "projects"), exist_ok=True)
+ port = _free_port()
+ config = ISOLATED_CONFIG.format(
+ port=port, ubridge=ubridge, dynamips=dynamips, images=images, tmp=tmpdir, kernel=kernel
+ )
+ config_path = os.path.join(tmpdir, "gns3_server.conf")
+ with open(config_path, "w") as f:
+ f.write(config)
+
+ log = open(os.path.join(tmpdir, "server.log"), "w")
+ # S603: fixed argv, this interpreter, no shell involvement
+ process = subprocess.Popen( # noqa: S603
+ [sys.executable, "-m", "gns3server", "--config", config_path],
+ cwd=REPO_ROOT,
+ stdout=log,
+ stderr=subprocess.STDOUT,
+ )
+ process.log = log # keep the file object alive with the process handle
+ compute = Compute(f"http://127.0.0.1:{port}/v3")
+
+ deadline = time.time() + 60
+ while time.time() < deadline:
+ try:
+ # S310: the port is ours, the host loopback
+ with urllib.request.urlopen(f"http://127.0.0.1:{port}/v3/version", timeout=2):
+ break
+ except Exception:
+ if process.poll() is not None:
+ raise AssertionError(
+ f"isolated server exited early:\n{open(os.path.join(tmpdir, 'server.log')).read()[-2000:]}"
+ )
+ time.sleep(0.3)
+ else:
+ raise AssertionError("isolated server did not come up within 60s")
+
+ compute.login()
+ return Server(compute, process, tmpdir)
+
+
+def live_server(kernel=True):
+ """The server under test.
+
+ With ``GNS3_E2E_URL`` set, that server is used for kernel mode (it must
+ already be configured for it); everything else starts an isolated
+ instance. Relay-mode tests always need an isolated instance — the
+ datapath choice is a server configuration.
+ """
+ url = os.environ.get("GNS3_E2E_URL")
+ if url and kernel:
+ compute = Compute(
+ url,
+ os.environ.get("GNS3_E2E_USER", "admin"),
+ os.environ.get("GNS3_E2E_PASSWORD", "admin"),
+ ).login()
+ return Server(compute)
+ return _start_isolated(kernel)
+
+
+def release(server, project_id, failed=False):
+ """Delete the test project (unless it should be kept), then stop the
+ isolated instance if this harness started one."""
+ if failed and keep_on_failure():
+ print(f"\n!! kept project {project_id} on {server.compute.url} for inspection")
+ else:
+ try:
+ server.compute.delete_project(project_id)
+ except Exception as e: # keep the original failure visible
+ print(f"!! could not delete project {project_id}: {e}")
+ if server.isolated and not (failed and keep_on_failure()):
+ server.stop()
+
+
+# ---------------------------------------------------------------------------
+# IOS console (dynamips telnet console)
+# ---------------------------------------------------------------------------
+
+
+class IOSConsole:
+ """A deliberately dumb IOS console driver.
+
+ ``run`` sends one line and returns *everything* received until the prompt
+ is back and the stream has been quiet — no split between "consumed" and
+ "remaining" buffers, so a caller can always parse the whole exchange.
+ Pacing matters: the IOS console drops characters typed while it is busy
+ (a long boot, the tail of a previous command), so every line goes out
+ after the previous exchange has settled.
+ """
+
+ def __init__(self, host, port):
+ self.sock = socket.create_connection((host, port), timeout=10)
+ self.sock.settimeout(0.3)
+ self.buf = ""
+
+ def _read(self):
+ """Drain whatever arrived; True when at least one byte was read."""
+ got = False
+ while True:
+ try:
+ data = self.sock.recv(65536)
+ except TimeoutError:
+ return got
+ if not data:
+ return got
+ self.buf += data.decode(errors="replace")
+ got = True
+
+ def run(self, line, timeout=30, quiet=0.8):
+ """Send one line; return the full exchange once the prompt is back
+ and nothing new arrived for *quiet* seconds."""
+ self.buf = ""
+ self.sock.sendall((line + "\r").encode())
+ deadline = time.time() + timeout
+ last_growth = time.time()
+ while time.time() < deadline:
+ if self._read():
+ last_growth = time.time()
+ if PROMPT_RE.search(self.buf) and time.time() - last_growth > quiet:
+ break
+ time.sleep(0.05)
+ out, self.buf = self.buf, ""
+ return out
+
+ def boot_wait(self, timeout=360):
+ """Wait until the IOS prompt is ready, answering the initial config
+ dialog if it appears. The prompt only counts once the stream around
+ it goes quiet — a boot banner still printing is not a prompt."""
+ answered = False
+ deadline = time.time() + timeout
+ self.buf = ""
+ last_growth = time.time()
+ while time.time() < deadline:
+ if self._read():
+ last_growth = time.time()
+ if not answered and "initial configuration dialog" in self.buf:
+ self.sock.sendall(b"no\r")
+ answered = True
+ last_growth = time.time()
+ prompt_seen = ("Press RETURN" in self.buf or answered) and PROMPT_RE.search(self.buf)
+ if prompt_seen and time.time() - last_growth > 1.2:
+ self.buf = ""
+ return
+ if not prompt_seen and time.time() - last_growth > 1.5:
+ # nudge the "Press RETURN" prompt along; never nudge an
+ # already-visible prompt (the echo would restart the quiet
+ # timer forever)
+ self.sock.sendall(b"\r")
+ time.sleep(0.1)
+ raise AssertionError(f"IOS console did not reach a prompt; tail: {self.buf[-300:]!r}")
+
+ def close(self):
+ self.sock.close()
+
+
+def dynamips_console(server, project_id, node_id):
+ node = server.compute.node(project_id, node_id)
+ host = node["console_host"] if node["console_host"] not in ("0.0.0.0", "::") else "127.0.0.1"
+ return IOSConsole(host, node["console"])
+
+
+def configure_ios(console, hostname, eth_ip=None, serial_ip=None, clock=False, eth_if="f1/0", serial_if="s2/0"):
+ """Baseline config on a freshly booted router: hostname, no console spam,
+ optional interface addresses."""
+ console.run("enable")
+ console.run("conf t")
+ console.run(f"hostname {hostname}")
+ console.run("no ip domain-lookup")
+ # global form: console logging off in one step, no line sub-mode to lose
+ # characters in
+ console.run("no logging console")
+ if eth_ip:
+ console.run(f"interface {eth_if}")
+ console.run(f"ip address {eth_ip} 255.255.255.0")
+ console.run("no shutdown")
+ console.run("exit")
+ if serial_ip:
+ console.run(f"interface {serial_if}")
+ console.run(f"ip address {serial_ip} 255.255.255.0")
+ if clock:
+ console.run("clock rate 64000")
+ console.run("no shutdown")
+ console.run("exit")
+ console.run("end")
+
+
+def ping(console, target, repeat=3, timeout=1):
+ """Run an IOS ping; returns {success, avg, raw} (success in percent)."""
+ out = console.run(f"ping {target} repeat {repeat} timeout {timeout}", timeout=repeat * timeout + 20)
+ success = re.search(r"Success rate is (\d+) percent", out)
+ rtt = re.search(r"round-trip min/avg/max = (\d+)/(\d+)/(\d+)", out)
+ return {"success": int(success.group(1)) if success else 0, "avg": int(rtt.group(2)) if rtt else None, "raw": out}
+
+
+def wait_ping(console, target, attempts=4, repeat=3, timeout=1):
+ """ping until it fully succeeds (the first packets of a fresh adjacency
+ are legitimately lost while the line protocol settles or ARP resolves);
+ returns the last result either way."""
+ result = None
+ for _ in range(attempts):
+ result = ping(console, target, repeat=repeat, timeout=timeout)
+ if result["success"] == 100:
+ return result
+ time.sleep(1.5)
+ return result
+
+
+# ---------------------------------------------------------------------------
+# Host-side inspection (same machine as the compute)
+# ---------------------------------------------------------------------------
+
+
+def sysfs(path):
+ with open(path) as f:
+ return f.read().strip()
+
+
+def tap_exists(name):
+ return os.path.exists(f"/sys/class/net/{name}")
+
+
+def tap_up(name):
+ """The tap's administrative state (IFF_UP): what the kernel datapath
+ drives for carrier on attach/detach/suspend."""
+ return bool(int(sysfs(f"/sys/class/net/{name}/flags"), 16) & 1)
+
+
+def bridge_members(bridge):
+ """Interfaces enslaved to a kernel bridge, or None when the bridge does
+ not exist."""
+ if not os.path.exists(f"/sys/class/net/{bridge}/brif"):
+ return None
+ return sorted(os.listdir(f"/sys/class/net/{bridge}/brif"))
+
+
+def link_bridge_name(link_id):
+ """The per-link kernel bridge name the controller derives from a link
+ id (see UDPLink._prepare)."""
+ return "gns3" + link_id.replace("-", "")[:11]
+
+
+def anchor_name(node_id, adapter, port):
+ """The dynamips anchor TAP name for a slot/port (see Router._tap_name)."""
+ return "gd" + node_id.replace("-", "")[:8] + f"e{adapter}p{port}"
+
+
+def qdiscs(dev):
+ """The tc qdisc listing of a host interface ('' when tc is missing)."""
+ if not TC:
+ return ""
+ # S603: resolved binary, fixed argv, no shell
+ result = subprocess.run( # noqa: S603
+ [TC, "qdisc", "show", "dev", dev], capture_output=True, text=True, check=False
+ )
+ return result.stdout
+
+
+def wait_until(predicate, timeout=10, interval=0.3):
+ deadline = time.time() + timeout
+ while time.time() < deadline:
+ if predicate():
+ return True
+ time.sleep(interval)
+ return predicate()
+
+
+def pcap_records(path):
+ """Count the records of a pcap file and collect the ethertype of the
+ first few (hex, e.g. '0800') — enough to prove a capture is real."""
+ import struct
+
+ ethertypes = []
+ with open(path, "rb") as f:
+ header = f.read(24)
+ if len(header) < 24:
+ return 0, ethertypes
+ endian = "<" if header[:4] in (b"\xd4\xc3\xb2\xa1", b"\x4d\x3c\xb2\xa1") else ">"
+ count = 0
+ while True:
+ rec = f.read(16)
+ if len(rec) < 16:
+ break
+ _ts, _tus, caplen, _orig = struct.unpack(endian + "IIII", rec)
+ data = f.read(caplen)
+ if len(data) < caplen:
+ break
+ count += 1
+ if len(ethertypes) < 5 and caplen >= 14:
+ ethertypes.append(data[12:14].hex())
+ return count, ethertypes
diff --git a/tests/e2e/test_dynamips_kernel_datapath.py b/tests/e2e/test_dynamips_kernel_datapath.py
new file mode 100644
index 000000000..deb8305d4
--- /dev/null
+++ b/tests/e2e/test_dynamips_kernel_datapath.py
@@ -0,0 +1,253 @@
+#
+# 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 .
+
+"""
+Live end-to-end test for the Dynamips kernel datapath (pytest marker
+``e2e``): two real c7200 routers on a real IOS image, one Ethernet link on
+the kernel path and one serial link on the relay, driven through the
+controller REST API and the IOS consoles. Real ICMP proves the datapath:
+
+* the link reports ``kernel_datapath`` and the per-link kernel bridge on
+ this host has exactly the two anchor TAPs enslaved (kernel objects, not
+ server bookkeeping);
+* a ``delay 100`` filter shows up as netem on the anchors and the measured
+ RTT rises by ~2x100 ms; clearing it resets the qdiscs and the RTT;
+* suspend drives the anchor administratively down and the pings stop;
+* capture writes a pcap full of ICMP frames;
+* a serial link between the same routers stays on the relay and still
+ pings (the relay regression net);
+* deleting / re-creating the link, and stopping / starting a router, leave
+ the wiring consistent;
+* deleting the project leaves no anchors or bridges behind.
+
+``test_dynamips_relay_control`` is the negative control: the same topology
+on an isolated relay-configured instance wires nothing into the kernel — no
+per-link bridge, nothing enslaved — and still pings, so the kernel objects
+above can only come from the kernel datapath. (The anchors themselves exist
+on both: they are the adapter's port interfaces, created by node start
+whenever the tap module is available, exactly like QEMU's TAPs.)
+"""
+
+import re
+import time
+
+import pytest
+
+from tests.e2e import harness
+
+pytestmark = pytest.mark.e2e
+
+# c7200 slot 0 is the fixed IO slot; the Ethernet PA goes in slot 1 (two
+# FE ports) and the serial PA in slot 2 (four ports).
+SLOTS = {"slot1": "PA-2FE-TX", "slot2": "PA-4T+"}
+ETH = (1, 0)
+SERIAL = (2, 0)
+
+
+def _pick_image(server):
+ images = server.compute.dynamips_images()
+ image = next((name for name in images if name.startswith("c7200")), None)
+ if not image:
+ pytest.skip(f"no c7200 image available on this server (images: {images[:5]})")
+ return image
+
+
+def _boot(server, project_id, node_id, timeout=360):
+ console = harness.dynamips_console(server, project_id, node_id)
+ console.boot_wait(timeout=timeout)
+ return console
+
+
+def test_dynamips_kernel_datapath():
+ server = harness.live_server(kernel=True)
+ compute = server.compute
+ if compute.capabilities().get("ubridge_tap") is not True:
+ pytest.skip("this compute's uBridge cannot create persistent TAPs (ubridge_tap != True)")
+ image = _pick_image(server)
+
+ project = compute.create_project("dyn-e2e")
+ pid = project["project_id"]
+ t1 = t2 = None
+ try:
+ r1 = compute.create_dynamips_router(pid, "E2E-R1", image, SLOTS)
+ r2 = compute.create_dynamips_router(pid, "E2E-R2", image, SLOTS)
+ r1_id, r2_id = r1["node_id"], r2["node_id"]
+ t1 = harness.anchor_name(r1_id, *ETH)
+ t2 = harness.anchor_name(r2_id, *ETH)
+
+ # The link is created while both routers are stopped: the kernel NIO
+ # is bound with no anchors in existence yet and attached by node
+ # start (the deferred-wiring path).
+ link = compute.create_link(pid, (r1_id, *ETH), (r2_id, *ETH))
+ lid = link["link_id"]
+ bridge = harness.link_bridge_name(lid)
+ assert link["kernel_datapath"] is True, link
+ assert not harness.tap_exists(t1), "anchors must not exist before start"
+
+ print(".. starting routers and waiting for IOS boots")
+ compute.call("POST", f"/projects/{pid}/nodes/{r1_id}/start")
+ compute.call("POST", f"/projects/{pid}/nodes/{r2_id}/start")
+ c1 = _boot(server, pid, r1_id)
+ c2 = _boot(server, pid, r2_id)
+
+ # Kernel objects: anchors exist and the per-link bridge has exactly
+ # the two of them enslaved.
+ assert harness.tap_exists(t1) and harness.tap_exists(t2), (t1, t2)
+ assert harness.bridge_members(bridge) == sorted([t1, t2]), harness.bridge_members(bridge)
+
+ harness.configure_ios(c1, "R1", eth_ip="10.1.1.1")
+ harness.configure_ios(c2, "R2", eth_ip="10.1.1.2")
+ out = c1.run("show ip int brief")
+ assert re.search(r"FastEthernet1/0\s+10\.1\.1\.1\s+\S+\s+\S+\s+up\s+up", out), out
+ print(".. both routers configured, pinging over the kernel link")
+ baseline = harness.wait_ping(c1, "10.1.1.2")
+ assert baseline["success"] == 100, baseline["raw"]
+
+ # delay 100: netem on both anchors (each impairs one direction),
+ # one-way ~100 ms => RTT grows by ~200 ms
+ compute.call("PUT", f"/projects/{pid}/links/{lid}", {"filters": {"delay": [100]}})
+ assert "netem" in harness.qdiscs(t1) and "netem" in harness.qdiscs(t2), (
+ harness.qdiscs(t1),
+ harness.qdiscs(t2),
+ )
+ delayed = harness.wait_ping(c1, "10.1.1.2")
+ assert delayed["success"] == 100, delayed["raw"]
+ assert delayed["avg"] >= 150, (baseline, delayed)
+ compute.call("PUT", f"/projects/{pid}/links/{lid}", {"filters": {}})
+ assert "netem" not in harness.qdiscs(t1), harness.qdiscs(t1)
+ fast = harness.wait_ping(c1, "10.1.1.2")
+ assert fast["success"] == 100 and fast["avg"] < 50, fast
+
+ # suspend: anchor admin-down, traffic dead; resume restores it
+ compute.call("PUT", f"/projects/{pid}/links/{lid}", {"suspend": True})
+ assert not harness.tap_up(t1)
+ dead = harness.ping(c1, "10.1.1.2", repeat=3)
+ assert dead["success"] == 0, dead["raw"]
+ compute.call("PUT", f"/projects/{pid}/links/{lid}", {"suspend": False})
+ assert harness.tap_up(t1)
+ assert harness.wait_ping(c1, "10.1.1.2")["success"] == 100
+
+ # capture: AF_PACKET on the anchor writes a real pcap
+ capture = compute.call("POST", f"/projects/{pid}/links/{lid}/capture/start", {"data_link_type": "DLT_EN10MB"})
+ harness.ping(c1, "10.1.1.2")
+ time.sleep(1)
+ compute.call("POST", f"/projects/{pid}/links/{lid}/capture/stop")
+ path = capture["capture_file_path"]
+ count, ethertypes = harness.pcap_records(path)
+ assert count >= 4, f"{path}: {count} records"
+ assert "0800" in ethertypes, (path, ethertypes)
+
+ # serial link between the same routers: per-port exclusion keeps it
+ # on the relay, where the real traffic still crosses
+ serial = compute.create_link(pid, (r1_id, *SERIAL), (r2_id, *SERIAL))
+ assert serial["kernel_datapath"] is False, serial
+ harness.configure_ios(c1, "R1", serial_ip="10.2.2.1", clock=True)
+ harness.configure_ios(c2, "R2", serial_ip="10.2.2.2")
+ relay_ping = harness.wait_ping(c1, "10.2.2.2")
+ assert relay_ping["success"] == 100, relay_ping["raw"]
+ compute.call("DELETE", f"/projects/{pid}/links/{serial['link_id']}")
+
+ # delete / re-create the kernel link: anchors survive, the bridge
+ # goes and comes back
+ compute.call("DELETE", f"/projects/{pid}/links/{lid}")
+ assert harness.bridge_members(bridge) is None
+ assert harness.tap_exists(t1) and harness.tap_exists(t2)
+ link = compute.create_link(pid, (r1_id, *ETH), (r2_id, *ETH))
+ lid = link["link_id"]
+ bridge = harness.link_bridge_name(lid)
+ assert link["kernel_datapath"] is True, link
+ assert harness.wait_ping(c1, "10.1.1.2", attempts=5)["success"] == 100
+
+ # stop / start keeps the whole wiring (the Dynamips hypervisor
+ # outlives ``vm stop``); the restarted router just reboots and is
+ # reconfigured
+ print(".. stopping and restarting R1")
+ compute.call("POST", f"/projects/{pid}/nodes/{r1_id}/stop")
+ assert harness.tap_exists(t1)
+ assert harness.bridge_members(bridge) == sorted([t1, t2])
+ compute.call("POST", f"/projects/{pid}/nodes/{r1_id}/start")
+ c1.close()
+ c1 = _boot(server, pid, r1_id)
+ assert harness.bridge_members(bridge) == sorted([t1, t2])
+ harness.configure_ios(c1, "R1", eth_ip="10.1.1.1")
+ assert harness.wait_ping(c1, "10.1.1.2", attempts=5)["success"] == 100
+
+ c1.close()
+ c2.close()
+ except BaseException:
+ harness.release(server, pid, failed=True)
+ raise
+ harness.release(server, pid, failed=False)
+
+ # deleting the project closes the nodes: anchors and bridges go with them
+ assert harness.wait_until(lambda: not harness.tap_exists(t1) and not harness.tap_exists(t2), timeout=15), (
+ t1,
+ t2,
+ )
+ assert harness.wait_until(lambda: harness.bridge_members(bridge) is None, timeout=10), bridge
+
+
+def test_dynamips_relay_control():
+ """Negative control: the same topology with ``enable_kernel_datapath =
+ false`` (an isolated instance) wires nothing into the kernel — no
+ per-link bridge, no enslavement — and still pings over the relay.
+
+ Anchors DO exist here (a Dynamips router creates its port TAPs at start
+ whenever the uBridge tap module is available — they are the adapter's
+ interfaces, exactly like QEMU's); the server configuration only decides
+ whether links are attached to them."""
+ server = harness.live_server(kernel=False)
+ compute = server.compute
+ image = _pick_image(server)
+
+ project = compute.create_project("dyn-e2e-relay")
+ pid = project["project_id"]
+ t1 = None
+ try:
+ r1 = compute.create_dynamips_router(pid, "E2E-R1", image, SLOTS)
+ r2 = compute.create_dynamips_router(pid, "E2E-R2", image, SLOTS)
+ r1_id, r2_id = r1["node_id"], r2["node_id"]
+ t1 = harness.anchor_name(r1_id, *ETH)
+
+ link = compute.create_link(pid, (r1_id, *ETH), (r2_id, *ETH))
+ lid = link["link_id"]
+ assert link["kernel_datapath"] is False, link
+
+ print(".. starting routers and waiting for IOS boots (relay)")
+ compute.call("POST", f"/projects/{pid}/nodes/{r1_id}/start")
+ compute.call("POST", f"/projects/{pid}/nodes/{r2_id}/start")
+ c1 = _boot(server, pid, r1_id)
+ c2 = _boot(server, pid, r2_id)
+
+ # no per-link kernel bridge on the relay: nothing is enslaved, the
+ # anchor (if present) is just an idle port interface
+ assert harness.bridge_members(harness.link_bridge_name(lid)) is None
+ if harness.tap_exists(t1):
+ assert not harness.tap_up(t1), "an unattached relay anchor should sit down"
+
+ harness.configure_ios(c1, "R1", eth_ip="10.1.1.1")
+ harness.configure_ios(c2, "R2", eth_ip="10.1.1.2")
+ relay_ping = harness.wait_ping(c1, "10.1.1.2")
+ assert relay_ping["success"] == 100, relay_ping["raw"]
+
+ c1.close()
+ c2.close()
+ except BaseException:
+ harness.release(server, pid, failed=True)
+ raise
+ harness.release(server, pid, failed=False)
+
+ assert harness.wait_until(lambda: not harness.tap_exists(t1), timeout=15), t1
diff --git a/tests/pytest.ini b/tests/pytest.ini
index 88c111a51..8914391ec 100644
--- a/tests/pytest.ini
+++ b/tests/pytest.ini
@@ -1,2 +1,9 @@
[pytest]
asyncio_default_fixture_loop_scope=function
+
+; Live end-to-end tests (tests/e2e, see its README) need privileges, real
+; images and emulator binaries — they never run in a normal test run. Run
+; them explicitly with: pytest -m e2e tests/e2e
+markers =
+ e2e: live end-to-end tests (real server, kernel interfaces, emulators)
+addopts = -m "not e2e"