From 071ca050be982de064ced0180f67124aa56e26c5 Mon Sep 17 00:00:00 2001 From: Jonathan Coates Date: Thu, 1 Oct 2026 18:02:37 +0100 Subject: [PATCH] Write some tests for bandwidth/traffic limits Just to prove to myself that it works! Specifically, I have some concerns with HTTP streaming about how that will interact with bandwidth limits, so having this in place should make it easier to write further tests. 7c60ed109fdc27cca04b7b59a84c08e257c1b0bf already proving useful. As we no longer have any global state we can run these tests in parallel. --- .../core/ComputerTestDelegate.java | 9 +- .../core/apis/http/HttpServer.kt | 18 +- .../core/apis/http/TestHttpApi.kt | 174 ++++++++++++++---- .../test/resources/junit-platform.properties | 1 + .../test/core/computer/LuaTaskRunner.kt | 1 + 5 files changed, 164 insertions(+), 39 deletions(-) diff --git a/projects/core/src/test/java/dan200/computercraft/core/ComputerTestDelegate.java b/projects/core/src/test/java/dan200/computercraft/core/ComputerTestDelegate.java index 9521cbe7b..67cc3e4cb 100644 --- a/projects/core/src/test/java/dan200/computercraft/core/ComputerTestDelegate.java +++ b/projects/core/src/test/java/dan200/computercraft/core/ComputerTestDelegate.java @@ -24,6 +24,8 @@ import it.unimi.dsi.fastutil.ints.Int2IntArrayMap; import org.jspecify.annotations.Nullable; import org.junit.jupiter.api.*; import org.junit.jupiter.api.function.Executable; +import org.junit.jupiter.api.parallel.Execution; +import org.junit.jupiter.api.parallel.ExecutionMode; import org.opentest4j.AssertionFailedError; import org.opentest4j.TestAbortedException; import org.slf4j.Logger; @@ -62,6 +64,7 @@ import java.util.stream.Stream; *

* Once all tests are done, we invoke {@code cct_test.finish} in order to mark everything as complete. */ +@Execution(ExecutionMode.SAME_THREAD) public class ComputerTestDelegate { private static final Path REPORT_PATH = TestFiles.get("luacov.report.out"); @@ -361,7 +364,7 @@ public class ComputerTestDelegate { currentTest = name; // Tell the computer to run it - LOG.info("Starting '{}'", formatName(name)); + LOG.debug("Starting '{}'", formatName(name)); computer.queueEvent("cct_test_run", new Object[]{ name }); var remaining = TIMEOUT; @@ -373,7 +376,7 @@ public class ComputerTestDelegate { remaining -= TICK_TIME; } - LOG.info("Finished '{}'", formatName(name)); + LOG.debug("Finished '{}'", formatName(name)); if (remaining <= 0) { throw new IllegalStateException("Timed out waiting for test"); @@ -423,7 +426,7 @@ public class ComputerTestDelegate { throw new RuntimeException(e); } try { - LOG.info("'{}' finished with {}", formatName(name), status); + LOG.debug("'{}' finished with {}", formatName(name), status); // Skip if a test mismatch if (!name.equals(currentTest)) { diff --git a/projects/core/src/test/kotlin/dan200/computercraft/core/apis/http/HttpServer.kt b/projects/core/src/test/kotlin/dan200/computercraft/core/apis/http/HttpServer.kt index af8cb8e76..ed6a0d739 100644 --- a/projects/core/src/test/kotlin/dan200/computercraft/core/apis/http/HttpServer.kt +++ b/projects/core/src/test/kotlin/dan200/computercraft/core/apis/http/HttpServer.kt @@ -15,10 +15,15 @@ import io.netty.handler.codec.http.* import io.netty.handler.codec.http.websocketx.* import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler.HandshakeComplete import io.netty.handler.codec.http.websocketx.extensions.compression.WebSocketServerCompressionHandler +import io.netty.util.AttributeKey import java.net.InetSocketAddress import java.nio.charset.StandardCharsets -class HttpRequest(val headers: HttpHeaders) +class HttpRequest(val headers: HttpHeaders, var lastMessageTime: Long) { + companion object { + internal val KEY: AttributeKey = AttributeKey.newInstance("cct.request") + } +} private class HttpState(var lastRequest: HttpRequest? = null) @@ -98,7 +103,10 @@ private class HttpServerHandler(private val state: HttpState) : SimpleChannelInb } override fun channelRead0(ctx: ChannelHandlerContext, request: FullHttpRequest) { - state.lastRequest = HttpRequest(request.headers()) + val requestInfo = HttpRequest(request.headers(), System.nanoTime()) + state.lastRequest = requestInfo + ctx.channel().attr(HttpRequest.KEY).set(requestInfo) + when (request.uri()) { "/", "/index.html" -> handleIndex(ctx, request) "/ws" -> handleWebsocket(ctx, request) @@ -143,6 +151,8 @@ private class HttpServerHandler(private val state: HttpState) : SimpleChannelInb */ private class WebSocketFrameHandler(private val activeConnections: MutableSet) : SimpleChannelInboundHandler() { override fun channelRead0(ctx: ChannelHandlerContext, frame: WebSocketFrame) { + ctx.channel().attr(HttpRequest.KEY).get().lastMessageTime = System.nanoTime() + when (frame) { is TextWebSocketFrame -> { // Send the uppercase string back. @@ -150,6 +160,10 @@ private class WebSocketFrameHandler(private val activeConnections: MutableSet { + ctx.channel().writeAndFlush(BinaryWebSocketFrame(frame.content().retainedDuplicate())) + } + is CloseWebSocketFrame -> {} else -> throw UnsupportedOperationException("unsupported frame type: ${frame.javaClass.name}") diff --git a/projects/core/src/test/kotlin/dan200/computercraft/core/apis/http/TestHttpApi.kt b/projects/core/src/test/kotlin/dan200/computercraft/core/apis/http/TestHttpApi.kt index a75d638bc..d6eb9ca30 100644 --- a/projects/core/src/test/kotlin/dan200/computercraft/core/apis/http/TestHttpApi.kt +++ b/projects/core/src/test/kotlin/dan200/computercraft/core/apis/http/TestHttpApi.kt @@ -12,6 +12,8 @@ import dan200.computercraft.core.apis.HTTPAPI import dan200.computercraft.core.apis.IAPIEnvironment import dan200.computercraft.core.apis.handles.ReadHandle import dan200.computercraft.core.apis.http.HttpServer.Companion.runServer +import dan200.computercraft.core.apis.http.TestHttpApi.Companion.INSTANT_DURATION +import dan200.computercraft.core.apis.http.TestHttpApi.Companion.LIMITED_DURATION import dan200.computercraft.core.apis.http.options.Action import dan200.computercraft.core.apis.http.options.AddressRule import dan200.computercraft.core.apis.http.request.HttpResponseHandle @@ -26,18 +28,37 @@ import org.junit.jupiter.api.Assertions.assertFalse import org.junit.jupiter.api.Test import org.junit.jupiter.api.assertThrows import java.lang.ref.Reference +import java.nio.ByteBuffer +import java.time.Duration import java.util.* +import java.util.concurrent.TimeUnit +import java.util.random.RandomGenerator +import java.util.random.RandomGeneratorFactory import kotlin.time.Duration.Companion.seconds class TestHttpApi { /** Create a [HTTPAPI] which is only permitted to access the current server. */ - private fun createHttpApi(environment: IAPIEnvironment, port: Int): HTTPAPI { + private fun createHttpApi(environment: IAPIEnvironment, port: Int, config: NettyHttp.Config = NettyHttp.DEFAULT_CONFIG): HTTPAPI { val rules = listOf( AddressRule.parse("127.0.0.1", OptionalInt.of(port), Action.ALLOW.toPartial()), ) - return HTTPAPI(environment, NettyHttp(NettyHttp.DEFAULT_CONFIG.withAddressRules(rules)).create(environment)) + return HTTPAPI(environment, NettyHttp(config.withAddressRules(rules)).create(environment)) } + /** Create a custom [NettyHttp.Config]. */ + private fun createHttpConfig( + uploadBandwidth: Int = NettyHttp.DEFAULT_CONFIG.uploadBandwidth, + downloadBandwidth: Int = NettyHttp.DEFAULT_CONFIG.downloadBandwidth, + ) = NettyHttp.Config( + NettyHttp.DEFAULT_CONFIG.enabled, + NettyHttp.DEFAULT_CONFIG.addressRules, + NettyHttp.DEFAULT_CONFIG.maxRequests, + NettyHttp.DEFAULT_CONFIG.maxWebsockets, + downloadBandwidth, + uploadBandwidth, + NettyHttp.DEFAULT_CONFIG.proxy, + ) + @Test fun `Connects to a HTTP server`() { runServer { server -> @@ -56,18 +77,23 @@ class TestHttpApi { } } + suspend fun LuaTaskRunner.websocket(server: HttpServer, httpApi: HTTPAPI): Pair { + val url = "ws://127.0.0.1:${server.port}/ws" + assertThat("http.websocket succeeded", httpApi.websocket(ObjectArguments(url)), array(equalTo(true))) + + val connectEvent = pullEvent() + assertThat(connectEvent, array(equalTo("websocket_success"), equalTo(url), isA(WebsocketHandle::class.java))) + + return Pair(url, connectEvent[2] as WebsocketHandle) + } + @Test fun `Connects to websocket`() { runServer { server -> LuaTaskRunner.runTest { - val url = "ws://127.0.0.1:${server.port}/ws" val httpApi = addApi(createHttpApi(environment, server.port)) - assertThat("http.websocket succeeded", httpApi.websocket(ObjectArguments(url)), array(equalTo(true))) + val (url, websocket) = websocket(server, httpApi) - val connectEvent = pullEvent() - assertThat(connectEvent, array(equalTo("websocket_success"), equalTo(url), isA(WebsocketHandle::class.java))) - - val websocket = connectEvent[2] as WebsocketHandle websocket.send(Coerced(LuaValues.encode("Hello")), Optional.of(false)) val message = websocket.receive(Optional.empty()).await() @@ -89,14 +115,9 @@ class TestHttpApi { fun `Errors if too many websocket messages are sent`() { runServer { server -> LuaTaskRunner.runTest { - val url = "ws://127.0.0.1:${server.port}/ws" val httpApi = addApi(createHttpApi(environment, server.port)) - assertThat("http.websocket succeeded", httpApi.websocket(ObjectArguments(url)), array(equalTo(true))) + val (_, websocket) = websocket(server, httpApi) - val connectEvent = pullEvent() - assertThat(connectEvent, array(equalTo("websocket_success"), equalTo(url), isA(WebsocketHandle::class.java))) - - val websocket = connectEvent[2] as WebsocketHandle val error = assertThrows { for (i in 0 until 10_000) { websocket.send(Coerced(LuaValues.encode("Hello")), Optional.of(false)) @@ -114,14 +135,8 @@ class TestHttpApi { fun `Closes if a websocket message is too large`() { runServer { server -> LuaTaskRunner.runTest { - val url = "ws://127.0.0.1:${server.port}/ws" val httpApi = addApi(createHttpApi(environment, server.port)) - assertThat("http.websocket succeeded", httpApi.websocket(ObjectArguments(url)), array(equalTo(true))) - - val connectEvent = pullEvent() - assertThat(connectEvent, array(equalTo("websocket_success"), equalTo(url), isA(WebsocketHandle::class.java))) - - val websocket = connectEvent[2] as WebsocketHandle + val (url, websocket) = websocket(server, httpApi) val out = ByteArray(AddressRule.WEBSOCKET_MESSAGE + 1) Random(0xDEADBEEF).nextBytes(out) @@ -139,14 +154,8 @@ class TestHttpApi { fun `Queues an event when the socket is externally closed`() { runServer { server -> LuaTaskRunner.runTest { - val url = "ws://127.0.0.1:${server.port}/ws" val httpApi = addApi(createHttpApi(environment, server.port)) - assertThat("http.websocket succeeded", httpApi.websocket(ObjectArguments(url)), array(equalTo(true))) - - val connectEvent = pullEvent() - assertThat(connectEvent, array(equalTo("websocket_success"), equalTo(url), isA(WebsocketHandle::class.java))) - - val websocket = connectEvent[2] as WebsocketHandle + val (url, websocket) = websocket(server, httpApi) server.stop() @@ -170,13 +179,8 @@ class TestHttpApi { fun `Closes websocket if the server doesn't respond`() { runServer(closeWebsocket = false) { server -> LuaTaskRunner.runTest(timeout = 10.seconds) { - val url = "ws://127.0.0.1:${server.port}/ws" val httpApi = addApi(createHttpApi(environment, server.port)) - assertThat("http.websocket succeeded", httpApi.websocket(ObjectArguments(url)), array(equalTo(true))) - - val connectEvent = pullEvent() - assertThat(connectEvent, array(equalTo("websocket_success"), equalTo(url), isA(WebsocketHandle::class.java))) - val websocket = connectEvent[2] as WebsocketHandle + val (url, websocket) = websocket(server, httpApi) websocket.close() @@ -187,4 +191,106 @@ class TestHttpApi { } } } + + private data class DelayStats(val send: Duration, val receive: Duration) + + /** + * Send a sequence of random messages on a websocket, and time how long each message takes to send and receive. + * + * As bandwidth limits are somewhat non-deterministic (dependent on IO and thread scheduling), this function sends + * multiple messages (as many as it can for 2 seconds), and takes the average delay over all of them. + */ + private fun measureWebsocketDelay(config: NettyHttp.Config): DelayStats { + var count = 0 + var send = Duration.ZERO + var receive = Duration.ZERO + + val random = RandomGeneratorFactory.of("L32X64MixRandom").create(-8780085030276499495L) + + /** Create a random message. This creates a new message every time to ensure messages do not compress. */ + fun randomMessage(): ByteBuffer { + val message = ByteArray(BANDWIDTH_MESSAGE_SIZE) + random.nextBytes(message) + return ByteBuffer.wrap(message).asReadOnlyBuffer() + } + + runServer { server -> + LuaTaskRunner.runTest(timeout = 10.seconds) { + val httpApi = addApi(createHttpApi(environment, server.port, config)) + val (_, websocket) = websocket(server, httpApi) + + // Send/receive one message before timing. Read bandwidth limits get applied retroactively (we disable + // auto-read after hitting the limit), so we need to ensure that happens first. + websocket.send(Coerced(randomMessage()), Optional.of(true)) + websocket.receive(Optional.empty()).await() + + val deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(2) + do { + val sent = System.nanoTime() + websocket.send(Coerced(randomMessage()), Optional.of(true)) + websocket.receive(Optional.empty()).await() + val clientReceived = System.nanoTime() + val serverReceived = server.lastRequest.lastMessageTime + + count += 1 + send += Duration.ofNanos(serverReceived - sent) + receive += Duration.ofNanos(clientReceived - serverReceived) + } while (sent < deadline) + + websocket.close() + } + } + + assertThat("Ran multiple tests", count, greaterThan(1)) + + return DelayStats(send = send.dividedBy(count.toLong()), receive = receive.dividedBy(count.toLong())) + } + + @Test + fun `Messages are sent and received with no delay`() { + val delay = measureWebsocketDelay(createHttpConfig()) + // When sending a message which does not hit the bandwidth cap, then we shouldn't see any delay on the message. + assertThat("Send delay", delay.send, lessThan(INSTANT_DURATION)) + assertThat("Receive delay", delay.receive, lessThan(INSTANT_DURATION)) + } + + @Test + fun `Limits upload bandwidth`() { + val delay = measureWebsocketDelay(createHttpConfig(uploadBandwidth = BANDWIDTH_MESSAGE_SIZE * BANDWIDTH_SCALE)) + // When sending a message which does hit the bandwidth cap, then we should expect a delay in sending the message, + // but not in receiving the reply. + assertThat("Send delay", delay.send, greaterThan(LIMITED_DURATION)) + assertThat("Receive delay", delay.receive, lessThan(INSTANT_DURATION)) + } + + @Test + fun `Limits download bandwidth`() { + val delay = + measureWebsocketDelay(createHttpConfig(downloadBandwidth = BANDWIDTH_MESSAGE_SIZE * BANDWIDTH_SCALE)) + // When sending a message which does hit the bandwidth cap, then we should expect a delay in sending the message, + // but not in receiving the reply. + assertThat("Send delay", delay.send, lessThan(INSTANT_DURATION)) + assertThat("Receive delay", delay.receive, greaterThan(LIMITED_DURATION)) + } + + companion object { + /** + * The length of a random websocket message used by our bandwidth tests + * + * @see measureWebsocketDelay + */ + private const val BANDWIDTH_MESSAGE_SIZE: Int = 1024 + + /** + * A scaler on our bandwidth. Increasing this allows us to send more messages within [measureWebsocketDelay], + * increasing the number of samples, but reducing the granularity of [INSTANT_DURATION] and [LIMITED_DURATION]. + */ + private const val BANDWIDTH_SCALE: Int = 2 + + /** The duration when we've not hit the bandwidth limit and the message is sent "instantly". */ + private val INSTANT_DURATION: Duration = Duration.ofMillis(50) + + /** The duration when we've hit the bandwidth limit. */ + private val LIMITED_DURATION: Duration = Duration.ofMillis(800).dividedBy(BANDWIDTH_SCALE.toLong()) + } } diff --git a/projects/core/src/test/resources/junit-platform.properties b/projects/core/src/test/resources/junit-platform.properties index ee4487e53..527dde601 100644 --- a/projects/core/src/test/resources/junit-platform.properties +++ b/projects/core/src/test/resources/junit-platform.properties @@ -2,3 +2,4 @@ # # SPDX-License-Identifier: CC0-1.0 junit.jupiter.execution.parallel.enabled=true +junit.jupiter.execution.parallel.mode.default=concurrent diff --git a/projects/core/src/testFixtures/kotlin/dan200/computercraft/test/core/computer/LuaTaskRunner.kt b/projects/core/src/testFixtures/kotlin/dan200/computercraft/test/core/computer/LuaTaskRunner.kt index d289de872..a587e37c8 100644 --- a/projects/core/src/testFixtures/kotlin/dan200/computercraft/test/core/computer/LuaTaskRunner.kt +++ b/projects/core/src/testFixtures/kotlin/dan200/computercraft/test/core/computer/LuaTaskRunner.kt @@ -33,6 +33,7 @@ class LuaTaskRunner : AbstractLuaTaskContext() { override fun close() { super.close() + for (api in apis) api.shutdown() environment.shutdown() }