Skip to content

Commit 8d53cb7

Browse files
whyolegwhyoleg
committed
Improve tests setup
* no parallel in tests * cleanup code in tests * use auto port in tests * simpler limiter implementation * implement logging for multiplexed transport
1 parent 91a07cf commit 8d53cb7

File tree

10 files changed

+91
-99
lines changed

10 files changed

+91
-99
lines changed

Diff for: ‎.github/workflows/run-tests.yml

+6-3
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@ jobs:
2727
fail-fast: false
2828
matrix:
2929
os: [ 'ubuntu-latest' ]
30-
target: [ 'jvmAll', 'js', 'native' ]
30+
target: [ 'jvm', 'jvm11', 'jvm17', 'jvm21', 'js', 'native' ]
3131
include:
3232
- os: 'macos-14'
3333
target: 'macos'
@@ -43,8 +43,11 @@ jobs:
4343
- uses: actions/checkout@v4
4444
- uses: ./.github/actions/setup-gradle
4545

46-
- run: ./gradlew ${{ matrix.target }}Test --continue
47-
timeout-minutes: 30
46+
- name: Build tests
47+
run: ./gradlew ${{ matrix.target }}Test --continue -Prsocketbuild.skipTests=true
48+
49+
- name: Run tests
50+
run: ./gradlew ${{ matrix.target }}Test --continue --no-configuration-cache --max-workers=1
4851

4952
- if: always() && !cancelled()
5053
uses: actions/upload-artifact@v4

Diff for: ‎rsocket-core/src/commonTest/kotlin/io/rsocket/kotlin/TestConnection.kt

+1-1
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,7 @@ class TestConnection : RSocketTransportSession.Sequential, RSocketClientTarget {
3939

4040
init {
4141
coroutineContext.job.invokeOnCompletion {
42-
sendChannel.close(it)
42+
sendChannel.cancelWithCause(it)
4343
receiveChannel.cancelWithCause(it)
4444
}
4545
}

Diff for: ‎rsocket-core/src/commonTest/kotlin/io/rsocket/kotlin/core/RSocketTest.kt

+14-10
Original file line numberDiff line numberDiff line change
@@ -51,8 +51,12 @@ class RSocketTest : SuspendTest, TestWithLeakCheck {
5151
}
5252
requestChannel { init, payloads ->
5353
init.close()
54-
payloads.onEach { it.close() }.launchIn(this)
55-
flow { repeat(10) { emitOrClose(payload("server got -> [$it]")) } }
54+
flow {
55+
coroutineScope {
56+
payloads.onEach { it.close() }.launchIn(this)
57+
repeat(10) { emitOrClose(payload("server got -> [$it]")) }
58+
}
59+
}
5660
}
5761
}
5862
}
@@ -251,10 +255,10 @@ class RSocketTest : SuspendTest, TestWithLeakCheck {
251255
}
252256
})
253257
val request = flow<Payload> { error("test") }
254-
//TODO
255-
kotlin.runCatching {
258+
// TODO: should requester fail if there was a failure in `request`?
259+
assertFailsWith(IllegalStateException::class) {
256260
requester.requestChannel(Payload.Empty, request).collect()
257-
}.also(::println)
261+
}
258262
val e = error.await()
259263
assertTrue(e is RSocketError.ApplicationError)
260264
assertEquals("test", e.message)
@@ -376,10 +380,10 @@ class RSocketTest : SuspendTest, TestWithLeakCheck {
376380
requesterReceiveChannel.cancel()
377381
delay(1000)
378382

379-
assertTrue(requesterSendChannel.isClosedForSend)
380-
assertTrue(responderSendChannel.isClosedForSend)
381-
assertTrue(requesterReceiveChannel.isClosedForReceive)
382-
assertTrue(responderReceiveChannel.isClosedForReceive)
383+
assertTrue(requesterSendChannel.isClosedForSend, "requesterSendChannel.isClosedForSend")
384+
assertTrue(responderSendChannel.isClosedForSend, "responderSendChannel.isClosedForSend")
385+
assertTrue(requesterReceiveChannel.isClosedForReceive, "requesterReceiveChannel.isClosedForReceive")
386+
assertTrue(responderReceiveChannel.isClosedForReceive, "responderReceiveChannel.isClosedForReceive")
383387
}
384388

385389
private suspend fun initRequestChannel(
@@ -413,7 +417,7 @@ class RSocketTest : SuspendTest, TestWithLeakCheck {
413417

414418
private suspend inline fun cancel(
415419
requesterChannel: SendChannel<Payload>,
416-
responderChannel: ReceiveChannel<Payload>
420+
responderChannel: ReceiveChannel<Payload>,
417421
) {
418422
responderChannel.cancel()
419423
delay(100)

Diff for: ‎rsocket-core/src/commonTest/kotlin/io/rsocket/kotlin/internal/RSocketRequesterTest.kt

+1-1
Original file line numberDiff line numberDiff line change
@@ -290,7 +290,7 @@ class RSocketRequesterTest : TestWithConnection(), TestWithLeakCheck {
290290
@Test
291291
fun testRequestReplyWithCancel() = test {
292292
connection.test {
293-
withTimeoutOrNull(100) { requester.requestResponse(Payload.Empty) }
293+
assertTrue(withTimeoutOrNull(100) { requester.requestResponse(Payload.Empty) } == null)
294294

295295
awaitFrame { assertTrue(it is RequestFrame) }
296296
awaitFrame { assertTrue(it is CancelFrame) }

Diff for: ‎rsocket-ktor/server/src/commonTest/kotlin/io/rsocket/kotlin/transport/ktor/websocket/server/WebSocketConnectionTest.kt

+2-4
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,6 @@ import io.rsocket.kotlin.ktor.client.*
2626
import io.rsocket.kotlin.ktor.server.*
2727
import io.rsocket.kotlin.payload.*
2828
import io.rsocket.kotlin.test.*
29-
import io.rsocket.kotlin.transport.tests.*
3029
import kotlinx.coroutines.*
3130
import kotlinx.coroutines.flow.*
3231
import kotlin.test.*
@@ -38,7 +37,6 @@ import io.rsocket.kotlin.ktor.client.RSocketSupport as ClientRSocketSupport
3837
import io.rsocket.kotlin.ktor.server.RSocketSupport as ServerRSocketSupport
3938

4039
class WebSocketConnectionTest : SuspendTest, TestWithLeakCheck {
41-
private val port = PortProvider.next()
4240
private val client = HttpClient(ClientCIO) {
4341
install(ClientWebSockets)
4442
install(ClientRSocketSupport) {
@@ -52,7 +50,7 @@ class WebSocketConnectionTest : SuspendTest, TestWithLeakCheck {
5250

5351
private var responderJob: Job? = null
5452

55-
private val server = embeddedServer(ServerCIO, port) {
53+
private val server = embeddedServer(ServerCIO, port = 0) {
5654
install(ServerWebSockets)
5755
install(ServerRSocketSupport) {
5856
server = TestServer()
@@ -87,7 +85,7 @@ class WebSocketConnectionTest : SuspendTest, TestWithLeakCheck {
8785

8886
@Test
8987
fun testWorks() = test {
90-
val rSocket = client.rSocket(port = port)
88+
val rSocket = client.rSocket(port = server.resolvedConnectors().single().port)
9189
val requesterJob = rSocket.coroutineContext.job
9290

9391
rSocket

Diff for: ‎rsocket-transport-tests/src/commonMain/kotlin/io/rsocket/kotlin/transport/tests/TransportTest.kt

+54-43
Original file line numberDiff line numberDiff line change
@@ -31,8 +31,9 @@ import kotlin.time.Duration.Companion.minutes
3131
import kotlin.time.Duration.Companion.seconds
3232

3333
//TODO: need to somehow rework those tests, as now they are super flaky
34+
// there is some issue in K/N tcp...
3435
abstract class TransportTest : SuspendTest, TestWithLeakCheck {
35-
override val testTimeout: Duration = 3.minutes
36+
override val testTimeout: Duration = 5.minutes
3637

3738
private val testJob = SupervisorJob()
3839
protected val testContext = testJob + TestExceptionHandler
@@ -60,25 +61,25 @@ abstract class TransportTest : SuspendTest, TestWithLeakCheck {
6061
@Test
6162
fun fireAndForget10() = test {
6263
(1..10).map { async { client.fireAndForget(payload(it)) } }.awaitAll()
63-
delay(1000) //TODO: leak check
64+
delay(100) // TODO: leak check
6465
}
6566

6667
@Test
6768
open fun largePayloadFireAndForget10() = test {
6869
(1..10).map { async { client.fireAndForget(requesterLargePayload) } }.awaitAll()
69-
delay(1000) //TODO: leak check
70+
delay(100) // TODO: leak check
7071
}
7172

7273
@Test
7374
fun metadataPush10() = test {
7475
(1..10).map { async { client.metadataPush(packet(requesterData)) } }.awaitAll()
75-
delay(1000) //TODO: leak check
76+
delay(100) // TODO: leak check
7677
}
7778

7879
@Test
7980
open fun largePayloadMetadataPush10() = test {
8081
(1..10).map { async { client.metadataPush(packet(requesterLargeData)) } }.awaitAll()
81-
delay(1000) //TODO: leak check
82+
delay(100) // TODO: leak check
8283
}
8384

8485
@Test
@@ -89,44 +90,50 @@ abstract class TransportTest : SuspendTest, TestWithLeakCheck {
8990

9091
@Test
9192
fun requestChannel1() = test(10.seconds) {
92-
val list = client.requestChannel(payload(0), flowOf(payload(0))).onEach { it.close() }.toList()
93-
assertEquals(1, list.size)
93+
val count =
94+
client.requestChannel(payload(0), flowOf(payload(0)))
95+
.onEach { it.close() }
96+
.count()
97+
assertEquals(1, count)
9498
}
9599

96100
@Test
97101
fun requestChannel3() = test {
98102
val request = flow {
99103
repeat(3) { emit(payload(it)) }
100104
}
101-
val list =
102-
client.requestChannel(payload(0), request).flowOn(PrefetchStrategy(3, 0)).onEach { it.close() }.toList()
103-
assertEquals(3, list.size)
105+
val count =
106+
client.requestChannel(payload(0), request)
107+
.flowOn(PrefetchStrategy(3, 0))
108+
.onEach { it.close() }
109+
.count()
110+
assertEquals(3, count)
104111
}
105112

106113
@Test
107114
open fun largePayloadRequestChannel200() = test {
108115
val request = flow {
109116
repeat(200) { emit(requesterLargePayload) }
110117
}
111-
val list =
118+
val count =
112119
client.requestChannel(requesterLargePayload, request)
113120
.flowOn(PrefetchStrategy(Int.MAX_VALUE, 0))
114121
.onEach { it.close() }
115-
.toList()
116-
assertEquals(200, list.size)
122+
.count()
123+
assertEquals(200, count)
117124
}
118125

119126
@Test
120-
@Ignore //flaky, ignore for now
127+
@IgnoreNative //flaky, ignore for now
121128
fun requestChannel20000() = test {
122129
val request = flow {
123130
repeat(20_000) { emit(payload(7)) }
124131
}
125-
val list = client.requestChannel(payload(7), request).flowOn(PrefetchStrategy(Int.MAX_VALUE, 0)).onEach {
132+
val count = client.requestChannel(payload(7), request).flowOn(PrefetchStrategy(Int.MAX_VALUE, 0)).onEach {
126133
assertEquals(requesterData, it.data.readText())
127134
assertEquals(requesterMetadata, it.metadata?.readText())
128-
}.toList()
129-
assertEquals(20_000, list.size)
135+
}.count()
136+
assertEquals(20_000, count)
130137
}
131138

132139
@Test
@@ -135,23 +142,26 @@ abstract class TransportTest : SuspendTest, TestWithLeakCheck {
135142
val request = flow {
136143
repeat(200_000) { emit(payload(it)) }
137144
}
138-
val list =
139-
client.requestChannel(payload(0), request).flowOn(PrefetchStrategy(10000, 0)).onEach { it.close() }.toList()
140-
assertEquals(200_000, list.size)
145+
val count =
146+
client.requestChannel(payload(0), request)
147+
.flowOn(PrefetchStrategy(10000, 0))
148+
.onEach { it.close() }
149+
.count()
150+
assertEquals(200_000, count)
141151
}
142152

143153
@Test
144-
@Ignore //flaky, ignore for now
154+
@IgnoreNative //flaky, ignore for now
145155
fun requestChannel16x256() = test {
146156
val request = flow {
147157
repeat(256) {
148158
emit(payload(it))
149159
}
150160
}
151161
(0..16).map {
152-
async(Dispatchers.Default) {
153-
val list = client.requestChannel(payload(0), request).onEach { it.close() }.toList()
154-
assertEquals(256, list.size)
162+
async {
163+
val count = client.requestChannel(payload(0), request).onEach { it.close() }.count()
164+
assertEquals(256, count)
155165
}
156166
}.awaitAll()
157167
}
@@ -165,30 +175,30 @@ abstract class TransportTest : SuspendTest, TestWithLeakCheck {
165175
}
166176
}
167177
(0..256).map {
168-
async(Dispatchers.Default) {
169-
val list = client.requestChannel(payload(0), request).onEach { it.close() }.toList()
170-
assertEquals(512, list.size)
178+
async {
179+
val count = client.requestChannel(payload(0), request).onEach { it.close() }.count()
180+
assertEquals(512, count)
171181
}
172182
}.awaitAll()
173183
}
174184

175185
@Test
176-
@Ignore //flaky, ignore for now
186+
@IgnoreNative //flaky, ignore for now
177187
fun requestChannel500NoLeak() = test {
178188
val request = flow {
179189
repeat(10_000) { emitOrClose(payload(3)) }
180190
}
181-
val list =
191+
val count =
182192
client
183193
.requestChannel(payload(3), request)
184194
.flowOn(PrefetchStrategy(Int.MAX_VALUE, 0))
185195
.take(500)
186196
.onEach {
187197
assertEquals(requesterData, it.data.readText())
188198
assertEquals(requesterMetadata, it.metadata?.readText())
189-
}.toList()
190-
assertEquals(500, list.size)
191-
delay(1000) //TODO: leak check
199+
}
200+
.count()
201+
assertEquals(500, count)
192202
}
193203

194204
@Test
@@ -202,6 +212,7 @@ abstract class TransportTest : SuspendTest, TestWithLeakCheck {
202212
}
203213

204214
@Test
215+
@IgnoreNative
205216
fun requestResponse100() = test {
206217
(1..100).map { async { client.requestResponse(payload(it)).let(Companion::checkPayload) } }.awaitAll()
207218
}
@@ -212,7 +223,7 @@ abstract class TransportTest : SuspendTest, TestWithLeakCheck {
212223
}
213224

214225
@Test
215-
@Ignore //flaky, ignore for now
226+
@IgnoreNative //flaky, ignore for now
216227
fun requestResponse10000() = test {
217228
(1..10000).map { async { client.requestResponse(payload(3)).let(Companion::checkPayload) } }.awaitAll()
218229
}
@@ -225,29 +236,29 @@ abstract class TransportTest : SuspendTest, TestWithLeakCheck {
225236

226237
@Test
227238
fun requestStream5() = test {
228-
val list =
229-
client.requestStream(payload(3)).flowOn(PrefetchStrategy(5, 0)).take(5).onEach { checkPayload(it) }.toList()
230-
assertEquals(5, list.size)
239+
val count =
240+
client.requestStream(payload(3)).flowOn(PrefetchStrategy(5, 0)).take(5).onEach { checkPayload(it) }.count()
241+
assertEquals(5, count)
231242
}
232243

233244
@Test
245+
@IgnoreNative
234246
fun requestStream10000() = test {
235-
val list = client.requestStream(payload(3)).onEach { checkPayload(it) }.toList()
236-
assertEquals(10000, list.size)
247+
val count = client.requestStream(payload(3)).onEach { checkPayload(it) }.count()
248+
assertEquals(10000, count)
237249
}
238250

239251
@Test
240-
@Ignore //flaky, ignore for now
252+
@IgnoreNative
241253
fun requestStream500NoLeak() = test {
242-
val list =
254+
val count =
243255
client
244256
.requestStream(payload(3))
245257
.flowOn(PrefetchStrategy(Int.MAX_VALUE, 0))
246258
.take(500)
247259
.onEach { checkPayload(it) }
248-
.toList()
249-
assertEquals(500, list.size)
250-
delay(1000) //TODO: leak check
260+
.count()
261+
assertEquals(500, count)
251262
}
252263

253264
companion object {

0 commit comments

Comments
 (0)