From 06252759649f582222de4983403a788723bae1f2 Mon Sep 17 00:00:00 2001 From: pabloFuente Date: Thu, 24 Sep 2026 14:51:07 +0200 Subject: [PATCH] openvidu-test-e2e: add iceCandidateTypeTest --- .../openvidu/test/e2e/NetworkConditioner.java | 150 +++++-- .../test/e2e/OpenViduTestAppE2eTest.java | 385 +++++++++++++++--- 2 files changed, 434 insertions(+), 101 deletions(-) diff --git a/openvidu-test-e2e/src/main/java/io/openvidu/test/e2e/NetworkConditioner.java b/openvidu-test-e2e/src/main/java/io/openvidu/test/e2e/NetworkConditioner.java index d44c09af4..6c376db72 100644 --- a/openvidu-test-e2e/src/main/java/io/openvidu/test/e2e/NetworkConditioner.java +++ b/openvidu-test-e2e/src/main/java/io/openvidu/test/e2e/NetworkConditioner.java @@ -19,10 +19,19 @@ public class NetworkConditioner { private static final String NETTOOLS_IMAGE = "ghcr.io/alexei-led/pumba-alpine-nettools:sha-19e0a46"; /** - * TCP port of the SFU's ICE-TCP candidates (livekit.yaml {@code rtc.tcp_port}). - * Every OUTBOUND impairment meant to cut a publisher's media must cover it too. - * (clients can reconnect to this port with an ICE restart when the UDP media - * path is dead). + * Port range of the SFU's host candidates (livekit.yaml + * {@code rtc.port_range_start} / {@code rtc.port_range_end}): their UDP ports + * with both engines, and the TCP ports of their ICE-TCP candidates too with + * mediasoup, where each transport listens for ICE-TCP on a port of its own. + */ + public static final String SFU_RTC_PORT_RANGE = "7900-7999"; + + /** + * TCP port of the SFU's ICE-TCP candidates with pion (livekit.yaml + * {@code rtc.tcp_port}). Every OUTBOUND impairment meant to cut a publisher's + * media must cover it too, and TCP over {@link #SFU_RTC_PORT_RANGE} for + * mediasoup (clients can reconnect over ICE-TCP with an ICE restart when the + * UDP media path is dead). */ public static final String SFU_ICE_TCP_PORT = "7881"; @@ -50,13 +59,14 @@ public class NetworkConditioner { // 7900-7999) is 100 ports. This value covers it. private static final int MAX_EXPANDED_PORTS = 1024; - // Name of the currently-running Pumba container to clear it. - private static String currentPumbaContainerName; + // Names of the running Pumba containers, to stop them all on clear(). A test + // may run several at once: each Pumba iptables command drops a single L4 + // protocol, see blockInboundPackets(). + private static final List pumbaContainerNames = new ArrayList<>(); - // Container that currently has an OUTBOUND blackout rule (a single iptables - // OUTPUT DROP added - // directly, not via Pumba). Tracked so clear() can flush it. See - // blackoutOutbound(). + // Container that currently has OUTBOUND DROP rules (iptables OUTPUT rules + // added directly, not via Pumba). Tracked so clear() can flush them. See + // blockOutboundPackets(). private static String blackoutContainer; // Both Pumba images are pinned to an immutable tag: pulling them once per JVM @@ -115,6 +125,53 @@ public class NetworkConditioner { durationSec); } + /** + * Drop every packet entering {@code targetContainer} from {@code remotePorts} + * over {@code protocol}, and wait until the drop is actually installed: Pumba + * runs detached and adds its iptables rules from a sidecar container of its + * own, which can take a few seconds. Each call starts one more Pumba container, + * so that several protocols can be blocked at once. {@link #clear()} stops them + * all. + */ + public static void blockInboundPackets(String targetContainer, Protocol protocol, String remotePorts, + int durationSec) { + final String ports = expandPorts(remotePorts); + if (ports == null) { + throw new IllegalArgumentException("blockInboundPackets needs the source ports to block"); + } + applyLossToInboundPackets(targetContainer, protocol, remotePorts, 100, durationSec); + final String proto = "-p " + protocol.name().toLowerCase(Locale.US) + " "; + final List missingPorts = new ArrayList<>(List.of(ports.split(","))); + final long deadline = System.currentTimeMillis() + 60000; + while (true) { + // Split on the rule prefix: the command output comes with its lines joined + String[] rules = nettools(targetContainer, "iptables", "-S INPUT").split("-A INPUT"); + missingPorts.removeIf(port -> { + for (String rule : rules) { + if (rule.contains(proto) && rule.contains("--sport " + port + " ") && rule.contains("DROP")) { + return true; + } + } + return false; + }); + if (missingPorts.isEmpty()) { + log.info("Pumba is dropping every {} packet entering container {} from port {}", protocol, + targetContainer, remotePorts); + return; + } + if (System.currentTimeMillis() > deadline) { + throw new IllegalStateException("Pumba did not install its INPUT DROP rules on container " + + targetContainer + " for " + protocol + " source ports " + missingPorts); + } + try { + Thread.sleep(1000); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(e); + } + } + } + /** * Update the OUTBOUND packet-loss percentage in place (no Pumba restart). * {@code tc qdisc change} re-applies the netem parameters of the qdisc Pumba @@ -260,30 +317,33 @@ public class NetworkConditioner { * (iptables matches {@code low:high} directly) reliably blocks all media, * unlike a flaky 100-filter netem * - * Stops the running Pumba first (avoids conflicting root qdiscs), then installs - * an iptables OUTPUT DROP rule over the whole UDP {@code mediaPortRange} (e.g. - * "7900-7999") plus one over the SFU's ICE-TCP port - * ({@link #SFU_ICE_TCP_PORT}) and one over its TURN listener - * ({@link #SFU_TURN_UDP_PORT}), so the browser's reconnect cannot fail over to - * TCP or to a relay candidate and escape the blackout. + * Stops the running Pumba first (avoids conflicting root qdiscs), then drops + * UDP and TCP over the whole {@code mediaPortRange} (e.g. "7900-7999"), TCP to + * the SFU's ICE-TCP port ({@link #SFU_ICE_TCP_PORT}) and UDP to its TURN + * listener ({@link #SFU_TURN_UDP_PORT}), so the browser's reconnect cannot fail + * over to TCP or to a relay candidate and escape the blackout. */ public static void blackoutOutbound(String targetContainer, String mediaPortRange, int durationSec) { clear(); - final String iptablesRange = mediaPortRange.replace('-', ':'); // iptables ranges are low:high - log.info("Total OUTBOUND blackout (100% loss) on container {} across SFU media port range {}, " - + "ICE-TCP port {} and TURN port {} (iptables OUTPUT DROP)", targetContainer, mediaPortRange, - SFU_ICE_TCP_PORT, SFU_TURN_UDP_PORT); - String out = nettools(targetContainer, "iptables", - "-A OUTPUT -o eth0 -p udp --dport " + iptablesRange + " -j DROP"); - log.info("blackout iptables -A OUTPUT (udp {}) result: {}", iptablesRange, out); - // clear() flushes the whole OUTPUT chain, so this rule goes away with the other - // one - String outTcp = nettools(targetContainer, "iptables", - "-A OUTPUT -o eth0 -p tcp --dport " + SFU_ICE_TCP_PORT + " -j DROP"); - log.info("blackout iptables -A OUTPUT (tcp {}) result: {}", SFU_ICE_TCP_PORT, outTcp); - String outTurn = nettools(targetContainer, "iptables", - "-A OUTPUT -o eth0 -p udp --dport " + SFU_TURN_UDP_PORT + " -j DROP"); - log.info("blackout iptables -A OUTPUT (udp {}) result: {}", SFU_TURN_UDP_PORT, outTurn); + blockOutboundPackets(targetContainer, Protocol.UDP, mediaPortRange); + blockOutboundPackets(targetContainer, Protocol.TCP, mediaPortRange); + blockOutboundPackets(targetContainer, Protocol.TCP, SFU_ICE_TCP_PORT); + blockOutboundPackets(targetContainer, Protocol.UDP, SFU_TURN_UDP_PORT); + } + + /** + * Drop every packet leaving {@code targetContainer} towards {@code remotePorts} + * (a single port or a range) over {@code protocol}, with an iptables OUTPUT + * rule added directly: Pumba cannot filter the egress by protocol. + * {@link #clear()} flushes it. + */ + public static void blockOutboundPackets(String targetContainer, Protocol protocol, String remotePorts) { + final String proto = protocol.name().toLowerCase(Locale.US); + final String ports = remotePorts.replace('-', ':'); // iptables ranges are low:high + String out = nettools(targetContainer, "iptables", "-A OUTPUT -o eth0 -p " + proto + " --dport " + ports + + " -j DROP"); + log.info("Dropping every {} packet leaving container {} towards port {} (iptables -A OUTPUT): {}", protocol, + targetContainer, remotePorts, out.isBlank() ? "done" : out.trim()); blackoutContainer = targetContainer; } @@ -338,25 +398,27 @@ public class NetworkConditioner { } /** - * Remove every impairment: stop the current Pumba container (SIGTERM, so it - * reverts the netem qdisc / iptables rule itself) and then scrub the target's + * Remove every impairment: stop the running Pumba containers (SIGTERM, so they + * revert the netem qdisc / iptables rules themselves) and then scrub the target's * network namespace anyway, so that the impairment is gone whether or not * Pumba got to revert it. */ public static void clear() { final String target = impairedContainer; impairedContainer = null; - if (currentPumbaContainerName != null) { - log.info("Clearing network impairment (stopping Pumba container {})", currentPumbaContainerName); - String out = commandLine.executeCommand( - "docker stop -t " + PUMBA_STOP_GRACE_SEC + " " + currentPumbaContainerName + " 2>&1", + if (!pumbaContainerNames.isEmpty()) { + // A single docker stop stops all of them in parallel + final String names = String.join(" ", pumbaContainerNames); + pumbaContainerNames.clear(); + log.info("Clearing network impairment (stopping Pumba containers {})", names); + String out = commandLine.executeCommand("docker stop -t " + PUMBA_STOP_GRACE_SEC + " " + names + " 2>&1", PUMBA_STOP_GRACE_SEC + 30); - log.info("docker stop {} result: {}", currentPumbaContainerName, out); - currentPumbaContainerName = null; + log.info("docker stop {} result: {}", names, out); } if (blackoutContainer != null) { - log.info("Clearing OUTBOUND blackout (flushing iptables OUTPUT) on container {}", blackoutContainer); - // Flush the OUTPUT chain: the blackout DROP is the only rule we ever add there + log.info("Clearing OUTBOUND drops (flushing iptables OUTPUT) on container {}", blackoutContainer); + // Flush the OUTPUT chain: the DROP rules of blockOutboundPackets() are the only + // ones we ever add there nettools(blackoutContainer, "iptables", "-F OUTPUT"); blackoutContainer = null; } @@ -444,9 +506,9 @@ public class NetworkConditioner { } private static String pumbaRun() { - currentPumbaContainerName = "pumba-netem-" + System.currentTimeMillis(); - return "docker run -d --name " + currentPumbaContainerName + " --rm -v " + DOCKER_SOCK + ":" + DOCKER_SOCK + " " - + PUMBA_IMAGE; + final String name = "pumba-netem-" + System.currentTimeMillis() + "-" + pumbaContainerNames.size(); + pumbaContainerNames.add(name); + return "docker run -d --name " + name + " --rm -v " + DOCKER_SOCK + ":" + DOCKER_SOCK + " " + PUMBA_IMAGE; } private static void runPumba(String cmd) { diff --git a/openvidu-test-e2e/src/test/java/io/openvidu/test/e2e/OpenViduTestAppE2eTest.java b/openvidu-test-e2e/src/test/java/io/openvidu/test/e2e/OpenViduTestAppE2eTest.java index e4d1dbdad..0f6e5436d 100644 --- a/openvidu-test-e2e/src/test/java/io/openvidu/test/e2e/OpenViduTestAppE2eTest.java +++ b/openvidu-test-e2e/src/test/java/io/openvidu/test/e2e/OpenViduTestAppE2eTest.java @@ -20,6 +20,8 @@ package io.openvidu.test.e2e; import io.openvidu.test.e2e.annotations.OnlyMediasoup; import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; import java.security.InvalidKeyException; import java.security.NoSuchAlgorithmException; import java.time.Duration; @@ -54,6 +56,8 @@ import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Tag; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; import org.openqa.selenium.By; import org.openqa.selenium.JavascriptExecutor; import org.openqa.selenium.Keys; @@ -962,36 +966,9 @@ public class OpenViduTestAppE2eTest extends AbstractOpenViduTestappE2eTest { final Integer width = videoOptimizations ? 1280 : null; final Integer height = videoOptimizations ? 720 : null; - // Connect to LiveKit server from a secure URL - String secureLivekitUrlFromOpenViduLocalDeployment = getLivekitWssUrlFromReadyCheckContainer(); + final String secureLivekitUrlFromOpenViduLocalDeployment = prepareNetemBrowsers(); - Assertions.assertNotNull(secureLivekitUrlFromOpenViduLocalDeployment, - "Could not obtain the LiveKit wss:// URL from the 'ready-check' container log. Is openvidu-local-deployment running? "); - log.info("Using LiveKit URL: {}", secureLivekitUrlFromOpenViduLocalDeployment); - - // Both browsers below are bridged into their own Docker network and reach the - // SFU through this public wildcard name: pin its address so that they never - // depend on the runner's DNS to open the signaling WebSocket - pinHostForNetemBrowser(secureLivekitUrlFromOpenViduLocalDeployment); - - NetworkConditioner.pullImages(); - - // Connect to the openvidu-testapp through "host.docker.internal" - final String secureAppUrl = APP_URL.replace("localhost", "host.docker.internal"); - - OpenViduTestappUser punchbagUser = new OpenViduTestappUser(setupBrowser("chromeNetwork")); - this.testappUsers.add(punchbagUser); - punchbagUser.getDriver().get(secureAppUrl); - WebElement urlInput = punchbagUser.getDriver().findElement(By.id("livekit-url")); - urlInput.clear(); - urlInput.sendKeys(secureLivekitUrlFromOpenViduLocalDeployment); - WebElement keyInput = punchbagUser.getDriver().findElement(By.id("livekit-api-key")); - keyInput.clear(); - keyInput.sendKeys(LIVEKIT_API_KEY); - WebElement secretInput = punchbagUser.getDriver().findElement(By.id("livekit-api-secret")); - secretInput.clear(); - secretInput.sendKeys(LIVEKIT_API_SECRET); - punchbagUser.getEventManager().startPolling(); + OpenViduTestappUser punchbagUser = setupNetemBrowserUser(secureLivekitUrlFromOpenViduLocalDeployment); if (!isPublisher && !isSubscriber) { throw new IllegalArgumentException("At least one of isPublisher or isSubscriber must be true"); @@ -1010,10 +987,7 @@ public class OpenViduTestAppE2eTest extends AbstractOpenViduTestappE2eTest { false, null, null, null); } - WebElement participantNameInput = punchbagUser.getDriver().findElement(By.id("participant-name-input-0")); - participantNameInput.clear(); - participantNameInput.sendKeys("PunchbagUser"); - setConnectionQualityRoomName(punchbagUser, roomName); + setParticipantAndRoomName(punchbagUser, "PunchbagUser", roomName); punchbagUser.getDriver().findElement(By.cssSelector(".connect-btn")).sendKeys(Keys.ENTER); try { @@ -1033,19 +1007,7 @@ public class OpenViduTestAppE2eTest extends AbstractOpenViduTestappE2eTest { waitUntilConnectionQuality(punchbagUser, 0, "PunchbagUser", q -> q.contains("excellent"), 20, "Expected baseline connection quality to be EXCELLENT before impairment"); - OpenViduTestappUser regularUser = new OpenViduTestappUser(setupBrowser("chromeNetwork")); - this.testappUsers.add(regularUser); - regularUser.getDriver().get(secureAppUrl); - urlInput = regularUser.getDriver().findElement(By.id("livekit-url")); - urlInput.clear(); - urlInput.sendKeys(secureLivekitUrlFromOpenViduLocalDeployment); - keyInput = regularUser.getDriver().findElement(By.id("livekit-api-key")); - keyInput.clear(); - keyInput.sendKeys(LIVEKIT_API_KEY); - secretInput = regularUser.getDriver().findElement(By.id("livekit-api-secret")); - secretInput.clear(); - secretInput.sendKeys(LIVEKIT_API_SECRET); - regularUser.getEventManager().startPolling(); + OpenViduTestappUser regularUser = setupNetemBrowserUser(secureLivekitUrlFromOpenViduLocalDeployment); if (isSubscriber) { this.addPublisher(regularUser, true, videoOptimizations, videoOptimizations, videoOptimizations, true, true, @@ -1053,10 +1015,7 @@ public class OpenViduTestAppE2eTest extends AbstractOpenViduTestappE2eTest { } else { this.addSubscriber(regularUser, videoOptimizations); } - participantNameInput = regularUser.getDriver().findElement(By.id("participant-name-input-0")); - participantNameInput.clear(); - participantNameInput.sendKeys("RegularUser"); - setConnectionQualityRoomName(regularUser, roomName); + setParticipantAndRoomName(regularUser, "RegularUser", roomName); regularUser.getDriver().findElement(By.cssSelector(".connect-btn")).sendKeys(Keys.ENTER); if (isPublisher) { @@ -1095,10 +1054,14 @@ public class OpenViduTestAppE2eTest extends AbstractOpenViduTestappE2eTest { regularUser.getEventManager().clearAllCurrentEvents(); if (outboundPacketLoss != null) { - // Drop packets the client sends through the publisher PC. The SFU's ICE-TCP - // port is impaired alike to avoid TCP fallbacks from the client side + // Drop packets the client sends through the publisher PC. pion's ICE-TCP port + // is impaired alike to avoid TCP fallbacks from the client side. mediasoup's + // are + // ports of the RTC range, one per transport, so they are cut altogether NetworkConditioner.applyLossToOutboundPackets(getNetemContainerName(punchbagUser), publisherPortInSfu + "," + NetworkConditioner.SFU_ICE_TCP_PORT, outboundPacketLoss, 10000); + NetworkConditioner.blockOutboundPackets(getNetemContainerName(punchbagUser), + NetworkConditioner.Protocol.TCP, NetworkConditioner.SFU_RTC_PORT_RANGE); } if (inboundPacketLoss != null) { // Drop packets the client receives through the subscriber PC (or publisher PC @@ -1111,7 +1074,52 @@ public class OpenViduTestAppE2eTest extends AbstractOpenViduTestappE2eTest { return new ImmutablePair<>(punchbagUser, regularUser); } - private void setConnectionQualityRoomName(OpenViduTestappUser user, String roomName) { + /** + * Get ready to open bridged ("chromeNetwork") browsers, the only ones that + * {@link NetworkConditioner} can impair, and return the secure LiveKit URL they + * must connect to. + */ + private String prepareNetemBrowsers() { + String secureLivekitUrl = getLivekitWssUrlFromReadyCheckContainer(); + Assertions.assertNotNull(secureLivekitUrl, + "Could not obtain the LiveKit wss:// URL from the 'ready-check' container log. Is openvidu-local-deployment running? "); + log.info("Using LiveKit URL: {}", secureLivekitUrl); + + // Bridged browsers live in their own Docker network and reach the SFU through + // this public wildcard name: pin its address so that they never depend on the + // runner's DNS to open the signaling WebSocket + pinHostForNetemBrowser(secureLivekitUrl); + + NetworkConditioner.pullImages(); + return secureLivekitUrl; + } + + /** + * Open the openvidu-testapp in a new bridged ("chromeNetwork") browser, with + * the connection settings filled in and its events being polled. + */ + private OpenViduTestappUser setupNetemBrowserUser(String secureLivekitUrl) throws Exception { + OpenViduTestappUser user = new OpenViduTestappUser(setupBrowser("chromeNetwork")); + this.testappUsers.add(user); + // Connect to the openvidu-testapp through "host.docker.internal" + user.getDriver().get(APP_URL.replace("localhost", "host.docker.internal")); + WebElement urlInput = user.getDriver().findElement(By.id("livekit-url")); + urlInput.clear(); + urlInput.sendKeys(secureLivekitUrl); + WebElement keyInput = user.getDriver().findElement(By.id("livekit-api-key")); + keyInput.clear(); + keyInput.sendKeys(LIVEKIT_API_KEY); + WebElement secretInput = user.getDriver().findElement(By.id("livekit-api-secret")); + secretInput.clear(); + secretInput.sendKeys(LIVEKIT_API_SECRET); + user.getEventManager().startPolling(); + return user; + } + + private void setParticipantAndRoomName(OpenViduTestappUser user, String participantName, String roomName) { + WebElement participantNameInput = user.getDriver().findElement(By.id("participant-name-input-0")); + participantNameInput.clear(); + participantNameInput.sendKeys(participantName); WebElement roomNameInput = user.getDriver().findElement(By.id("room-name-input-0")); roomNameInput.clear(); roomNameInput.sendKeys(roomName); @@ -1197,7 +1205,8 @@ public class OpenViduTestAppE2eTest extends AbstractOpenViduTestappE2eTest { OpenViduTestappUser punchbagUser = users.getLeft(); OpenViduTestappUser regularUser = users.getRight(); - NetworkConditioner.blackoutOutbound(getNetemContainerName(punchbagUser), "7900-7999", 120); + NetworkConditioner.blackoutOutbound(getNetemContainerName(punchbagUser), + NetworkConditioner.SFU_RTC_PORT_RANGE, 120); punchbagUser.getEventManager().waitUntilEventReaches(0, "connectionQualityChanged", "RoomEvent", 1); regularUser.getEventManager().waitUntilEventReaches(0, "connectionQualityChanged", "RoomEvent", 1); @@ -1250,7 +1259,8 @@ public class OpenViduTestAppE2eTest extends AbstractOpenViduTestappE2eTest { // Cut the publisher's media uplink at once while its quality is EXCELLENT, // keeping its signaling up: a sudden network outage on the publisher's side - NetworkConditioner.blackoutOutbound(getNetemContainerName(punchbagUser), "7900-7999", 120); + NetworkConditioner.blackoutOutbound(getNetemContainerName(punchbagUser), + NetworkConditioner.SFU_RTC_PORT_RANGE, 120); try { waitUntilConnectionQuality(regularUser, 0, "PunchbagUser", q -> q.contains("lost"), 45, "Expected the publisher's connection quality to reach LOST"); @@ -1299,7 +1309,8 @@ public class OpenViduTestAppE2eTest extends AbstractOpenViduTestappE2eTest { // short network outage on the publisher's side. Its quality may well go LOST, // but once the network is back its session must go on (resumed, not left and // joined again) - NetworkConditioner.blackoutOutbound(getNetemContainerName(punchbagUser), "7900-7999", OUTAGE_SECONDS); + NetworkConditioner.blackoutOutbound(getNetemContainerName(punchbagUser), + NetworkConditioner.SFU_RTC_PORT_RANGE, OUTAGE_SECONDS); try { Thread.sleep(OUTAGE_SECONDS * 1000L); } finally { @@ -1437,7 +1448,7 @@ public class OpenViduTestAppE2eTest extends AbstractOpenViduTestappE2eTest { // Final step: a TOTAL blackout (100% loss) across the WHOLE SFU media port // range (7900-7999) final int BLACKOUT_PCT = 100; - NetworkConditioner.blackoutOutbound(container, "7900-7999", 120); + NetworkConditioner.blackoutOutbound(container, NetworkConditioner.SFU_RTC_PORT_RANGE, 120); ConnectionQuality pubBlackout = latestConnectionQuality(punchbagUser, 0, "PunchbagUser"); long lostDeadline = System.currentTimeMillis() + 45000L; while (pubBlackout != ConnectionQuality.LOST && System.currentTimeMillis() < lostDeadline) { @@ -1489,6 +1500,266 @@ public class OpenViduTestAppE2eTest extends AbstractOpenViduTestappE2eTest { gracefullyLeaveParticipants(punchbagUser, 1); } + /** + * Every way a client can reach the SFU of openvidu-local-deployment: a host + * candidate of the SFU over UDP ({@code rtc.port_range_*}) or over TCP + * ({@code rtc.tcp_port} with pion, a port of the RTC range per transport with + * mediasoup), or a relay candidate of the client allocated from the SFU's + * embedded TURN server over UDP ({@code turn.udp_port}, the only TURN listener + * it has). Each one knows the inbound packets to drop so that it is the only + * way left. + * + * The type of the client's selected local candidate is not enough to tell the + * paths apart: it is peer-reflexive on all of them whenever there is a NAT + * between the client and the SFU, like here, where both run in Docker bridge + * networks and the SFU gets the client's packets, directly or from the TURN + * relay, from the docker0 gateway through the published ports. A relayed + * candidate still carries the {@code relayProtocol} and {@code url} of the + * relay candidate it is based on. + */ + enum IceCandidateType { + + HOST_UDP(false, "udp"), HOST_TCP(false, "tcp"), RELAY_UDP(true, "udp"); + + // The ICE-TCP ports of both engines + static final String SFU_TCP_PORTS = NetworkConditioner.SFU_ICE_TCP_PORT + "," + + NetworkConditioner.SFU_RTC_PORT_RANGE; + + // Of the local candidate the client must end up connected through + final boolean relay; + final String protocol; + + IceCandidateType(boolean relay, String protocol) { + this.relay = relay; + this.protocol = protocol; + } + + /** + * Drop every answer of the SFU on the other paths, so that the connectivity + * checks of their candidate pairs fail. The Pumba duration outlasts the test: + * {@link NetworkConditioner#clear()} removes the drops at its end. + */ + void blockTheOtherPaths(String container) { + final int durationSec = 600; + switch (this) { + case HOST_UDP -> { + NetworkConditioner.blockInboundPackets(container, NetworkConditioner.Protocol.TCP, SFU_TCP_PORTS, + durationSec); + NetworkConditioner.blockInboundPackets(container, NetworkConditioner.Protocol.UDP, + NetworkConditioner.SFU_TURN_UDP_PORT, durationSec); + } + case HOST_TCP -> NetworkConditioner.blockInboundPackets(container, NetworkConditioner.Protocol.UDP, + NetworkConditioner.SFU_RTC_PORT_RANGE + "," + NetworkConditioner.SFU_TURN_UDP_PORT, + durationSec); + case RELAY_UDP -> { + NetworkConditioner.blockInboundPackets(container, NetworkConditioner.Protocol.UDP, + NetworkConditioner.SFU_RTC_PORT_RANGE, durationSec); + NetworkConditioner.blockInboundPackets(container, NetworkConditioner.Protocol.TCP, SFU_TCP_PORTS, + durationSec); + } + } + } + } + + @ParameterizedTest(name = "ICE candidate type {0}") + @EnumSource(IceCandidateType.class) + void iceCandidateTypeTest(IceCandidateType iceCandidateType) throws Exception { + + log.info("ICE candidate type {}", iceCandidateType); + + // mediasoup transports listen for ICE-TCP on the RTC port range, which the + // deployment only publishes over UDP + final boolean publishTcpPortRange = iceCandidateType.protocol.equals("tcp"); + if (publishTcpPortRange) { + setSfuTcpPortRangePublished(true); + } + try { + final String secureLivekitUrl = prepareNetemBrowsers(); + // A room of its own, like connectionQualityTest: a failed ICE connection makes + // LiveKit prefer TCP for the same room name and identity for a while + final String roomName = "IceCandidateType" + System.currentTimeMillis(); + + OpenViduTestappUser punchbagUser = setupNetemBrowserUser(secureLivekitUrl); + try { + // Before connecting: ICE checks all candidate pairs at once and keeps the + // first one that works + iceCandidateType.blockTheOtherPaths(getNetemContainerName(punchbagUser)); + + // PunchbagUser publishes audio and video and subscribes to RegularUser's, so + // that media goes both ways through the forced candidate + this.addPublisher(punchbagUser, true, false, false, false, true, true, null, null, null); + setParticipantAndRoomName(punchbagUser, "PunchbagUser", roomName); + punchbagUser.getDriver().findElement(By.cssSelector(".connect-btn")).sendKeys(Keys.ENTER); + try { + punchbagUser.getEventManager().waitUntilEventReaches("connected", "RoomEvent", 1); + } catch (Exception e) { + Assertions.fail("PunchbagUser could not connect with only its " + iceCandidateType + + " candidates able to reach the SFU", e); + } + punchbagUser.getEventManager().waitUntilEventReaches("localTrackPublished", "RoomEvent", 2); + + OpenViduTestappUser regularUser = setupNetemBrowserUser(secureLivekitUrl); + this.addPublisher(regularUser, true, false, false, false, true, true, null, null, null); + setParticipantAndRoomName(regularUser, "RegularUser", roomName); + regularUser.getDriver().findElement(By.cssSelector(".connect-btn")).sendKeys(Keys.ENTER); + regularUser.getEventManager().waitUntilEventReaches("connected", "RoomEvent", 1); + punchbagUser.getEventManager().waitUntilEventReaches("trackSubscribed", "RoomEvent", 2); + regularUser.getEventManager().waitUntilEventReaches("trackSubscribed", "RoomEvent", 2); + + // Media plays both ways... + waitUntilSubscriberFramesDecodedIncrease(punchbagUser, + punchbagUser.getDriver().findElement(By.cssSelector("#openvidu-instance-0 video.remote"))); + waitUntilSubscriberFramesDecodedIncrease(regularUser, + regularUser.getDriver().findElement(By.cssSelector("#openvidu-instance-0 video.remote"))); + Assertions.assertTrue(assertAllElementsHaveTracks(punchbagUser, "audio.remote", true, false), + "PunchbagUser's HTMLAudioElements were expected to have only one audio track"); + // ...with nothing lost on the way + waitUntilConnectionQuality(punchbagUser, 0, "PunchbagUser", q -> q.contains("excellent"), 20, + "Expected PunchbagUser's own connection quality to be EXCELLENT through " + iceCandidateType); + waitUntilConnectionQuality(regularUser, 0, "PunchbagUser", q -> q.contains("excellent"), 20, + "Expected PunchbagUser's connection quality seen by RegularUser to be EXCELLENT through " + + iceCandidateType); + + // ...and through the forced candidate + assertConnectedThroughIceCandidateType(readPcTransportsInfoJson(punchbagUser, 0), iceCandidateType); + Assertions.assertEquals(0, + punchbagUser.getEventManager().getNumEvents("reconnecting-RoomEvent").get(), + "PunchbagUser was expected to connect through " + iceCandidateType + " at the first attempt"); + } finally { + NetworkConditioner.clear(); + } + + gracefullyLeaveParticipants(punchbagUser, 1); + } finally { + if (publishTcpPortRange) { + setSfuTcpPortRangePublished(false); + } + } + } + + /** + * Add/Remove the TCP port range from the "openvidu" container in the + * openvidu-local-deployment. It replicates the declared UDP port range. + */ + private void setSfuTcpPortRangePublished(boolean published) throws Exception { + String[] labels = commandLine.executeCommand("docker inspect -f '" + + "{{index .Config.Labels \"com.docker.compose.project\"}}|" + + "{{index .Config.Labels \"com.docker.compose.project.working_dir\"}}|" + + "{{index .Config.Labels \"com.docker.compose.project.config_files\"}}' openvidu", 30).trim() + .split("\\|"); + Assertions.assertEquals(3, labels.length, "The openvidu container was not started by Docker Compose"); + final String project = labels[0]; + final String workingDir = labels[1]; + final List configFiles = List.of(labels[2].split(",")); + + final String range = NetworkConditioner.SFU_RTC_PORT_RANGE; + final String udpMapping = range + ":" + range + "/udp"; + final String tcpMapping = range + ":" + range + "/tcp"; + Path composeFile = null; + for (String configFile : configFiles) { + if (Files.readString(Path.of(configFile)).contains(udpMapping)) { + composeFile = Path.of(configFile); + } + } + Assertions.assertNotNull(composeFile, "No mapping " + udpMapping + " in " + configFiles); + + final String original = Files.readString(composeFile) + .replaceAll("(?m)^[ \\t]*- " + java.util.regex.Pattern.quote(tcpMapping) + "\\R", ""); + String updated = original; + if (published) { + java.util.regex.Matcher udpLine = java.util.regex.Pattern + .compile("(?m)^([ \\t]*- )" + java.util.regex.Pattern.quote(udpMapping) + "(\\R)") + .matcher(original); + Assertions.assertTrue(udpLine.find(), "No mapping " + udpMapping + " in " + composeFile); + updated = new StringBuilder(original) + .insert(udpLine.end(), udpLine.group(1) + tcpMapping + udpLine.group(2)) + .toString(); + } + log.info("{} port mapping {} in {} and recreating the openvidu service", published ? "Adding" : "Removing", + tcpMapping, composeFile); + Files.writeString(composeFile, updated); + + StringBuilder compose = new StringBuilder("docker compose -p ").append(project).append(" --project-directory ") + .append(workingDir); + configFiles.forEach(configFile -> compose.append(" -f ").append(configFile)); + // Compose only recreates the service if its configuration changed + log.info("docker compose up result: {}", + commandLine.executeCommand(compose + " up -d --no-deps openvidu 2>&1", 180)); + + // The SFU is ready once its API answers again + final long deadline = System.currentTimeMillis() + 120000; + while (true) { + try { + if (LK.listRooms().execute().isSuccessful()) { + return; + } + } catch (IOException e) { + // Not listening yet + } + if (System.currentTimeMillis() > deadline) { + throw new IllegalStateException("openvidu did not come back after being recreated"); + } + Thread.sleep(1000); + } + } + + /** + * Check in the peer-info JSON of {@link #readPcTransportsInfoJson} that every + * peer connection of the user is connected through a local candidate of + * {@code iceCandidateType}, paired with the SFU's host candidate of the same + * protocol. + */ + private void assertConnectedThroughIceCandidateType(String json, IceCandidateType iceCandidateType) { + Assertions.assertNotNull(json, "Could not read the peer-info dialog"); + log.info("Peer-info of the user forced to {}: {}", iceCandidateType, json); + JsonObject info = JsonParser.parseString(json).getAsJsonObject(); + JsonObject pcTransports = info.getAsJsonObject("PCTransports"); + JsonObject selectedLocalCandidates = info.getAsJsonObject("RTCIceCandidateStats"); + // Only the publisher PC exists in single peer connection mode + for (String pc : pcTransports.keySet()) { + final String context = " of the " + pc + " PC forced to " + iceCandidateType + ". Peer-info: " + json; + JsonObject transport = pcTransports.getAsJsonObject(pc); + Assertions.assertEquals("connected", transport.get("connectionState").getAsString(), + "Wrong connectionState" + context); + String iceConnectionState = transport.get("iceConnectionState").getAsString(); + Assertions.assertTrue(iceConnectionState.equals("connected") || iceConnectionState.equals("completed"), + "Wrong iceConnectionState " + iceConnectionState + context); + + JsonArray selected = selectedLocalCandidates.getAsJsonArray(pc); + Assertions.assertEquals(1, selected.size(), "Expected exactly one selected local candidate" + context); + JsonObject local = selected.get(0).getAsJsonObject(); + final String candidateType = local.get("candidateType").getAsString(); + Assertions.assertTrue( + iceCandidateType.relay ? List.of("relay", "prflx").contains(candidateType) + : List.of("host", "srflx", "prflx").contains(candidateType), + "Wrong type " + candidateType + " of the selected local candidate" + context); + Assertions.assertEquals(iceCandidateType.protocol, local.get("protocol").getAsString(), + "Wrong protocol of the selected local candidate" + context); + if (iceCandidateType.relay) { + Assertions.assertTrue(local.has("relayProtocol") && local.has("url"), + "The selected local candidate is not relayed" + context); + Assertions.assertEquals("udp", local.get("relayProtocol").getAsString(), + "Wrong protocol towards the TURN server" + context); + String url = local.get("url").getAsString(); + Assertions.assertTrue( + url.startsWith("turn:") && url.contains(":" + NetworkConditioner.SFU_TURN_UDP_PORT + "?"), + "The relay candidate was not allocated from the SFU's TURN server" + context); + } else { + Assertions.assertFalse(local.has("relayProtocol"), "The selected local candidate is relayed" + context); + } + + // The remote end is always a host candidate of the SFU: the relay path goes + // from the TURN server to its UDP one + String connectedAddress = transport.get("connectedAddress").getAsString(); + String remotePort = connectedAddress.substring(connectedAddress.lastIndexOf(':') + 1); + String sfuPorts = iceCandidateType.protocol.equals("tcp") ? IceCandidateType.SFU_TCP_PORTS + : NetworkConditioner.SFU_RTC_PORT_RANGE; + Assertions.assertTrue(List.of(NetworkConditioner.expandPorts(sfuPorts).split(",")).contains(remotePort), + "Not connected to a " + iceCandidateType.protocol + " port of the SFU (" + sfuPorts + ")" + + context); + } + } + @Test @DisplayName("Data Tracks publish, subscribe, send and receive") void dataTracksTest() throws Exception {