test: delaySafe as a safe replacement for delay when used inside …

…a `runTest` or `TestScope`. Details: https://github.com/Kotlin/kotlinx.coroutines/blob/81e17dd/kotlinx-coroutines-test/README.md#delay-skipping Kotlin/kotlinx.coroutines#3588 Signed-off-by: Artyom Shendrik <[email protected]>
fluxo-kt · Jan 20, 2023 · 0fd87fd · 0fd87fd
1 parent d286499
commit 0fd87fd
Show file tree

Hide file tree

Showing 10 changed files with 53 additions and 33 deletions.
diff --git a/fluxo-core/src/commonTest/kotlin/kt/fluxo/test/DelaySafe.kt b/fluxo-core/src/commonTest/kotlin/kt/fluxo/test/DelaySafe.kt
@@ -0,0 +1,26 @@
+package kt.fluxo.test
+
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.runTest
+import kotlinx.coroutines.withContext
+
+/**
+ * Use as a safe replacement for [delay] when used inside a [runTest]/[TestScope].
+ *
+ * Delays coroutine for a given time without blocking a thread and resumes it after a specified time.
+ *
+ * [Documentation on delay-skipping in tests](https://github.com/Kotlin/kotlinx.coroutines/blob/81e17dd/kotlinx-coroutines-test/README.md#delay-skipping)
+ *
+ * @param timeMillis timeout time in milliseconds.
+ *
+ * @see delay
+ * @see withTimeoutSafe
+ */
+@Suppress("MaxLineLength")
+suspend fun delaySafe(timeMillis: Long) {
+    withContext(Dispatchers.Default) {
+        delay(timeMillis)
+    }
+}
diff --git a/fluxo-core/src/commonTest/kotlin/kt/fluxo/test/TestFlowObserver.kt b/fluxo-core/src/commonTest/kotlin/kt/fluxo/test/TestFlowObserver.kt
@@ -7,11 +7,8 @@ import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.TimeoutCancellationException
 import kotlinx.coroutines.cancel
-import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.launch
-import kotlinx.coroutines.withContext
-import kotlinx.coroutines.withTimeout
 
 
 /**
@@ -45,18 +42,15 @@ class TestFlowObserver<T>(flow: Flow<T>) {
      * @param throwTimeout Throw if timed out (by default)
      */
     suspend fun awaitFor(timeout: Long = 5000L, throwTimeout: Boolean = true, condition: TestFlowObserver<T>.() -> Boolean) {
-        // https://github.com/Kotlin/kotlinx.coroutines/blob/master/kotlinx-coroutines-test/README.md#using-withtimeout-inside-runtest
-        withContext(Dispatchers.Default) {
-            try {
-                withTimeout(timeout) {
-                    @Suppress("UNUSED_EXPRESSION")
-                    while (!condition()) {
-                        delay(AWAIT_TIMEOUT_MS)
-                    }
+        try {
+            withTimeoutSafe(timeout) {
+                @Suppress("UNUSED_EXPRESSION")
+                while (!condition()) {
+                    delaySafe(AWAIT_TIMEOUT_MS)
                 }
-            } catch (e: TimeoutCancellationException) {
-                if (throwTimeout) throw e
             }
+        } catch (e: TimeoutCancellationException) {
+            if (throwTimeout) throw e
         }
     }
 

diff --git a/fluxo-core/src/commonTest/kotlin/kt/fluxo/test/WithTimeoutSafe.kt b/fluxo-core/src/commonTest/kotlin/kt/fluxo/test/WithTimeoutSafe.kt
@@ -24,6 +24,7 @@ import kotlin.contracts.contract
  * @param timeMillis timeout time in milliseconds.
  *
  * @see withTimeout
+ * @see delaySafe
  */
 @Suppress("MaxLineLength")
 suspend inline fun <T> withTimeoutSafe(timeMillis: Long, crossinline block: suspend CoroutineScope.() -> T): T {

diff --git a/fluxo-core/src/commonTest/kotlin/kt/fluxo/tests/ContainerExceptionHandlerTest.kt b/fluxo-core/src/commonTest/kotlin/kt/fluxo/tests/ContainerExceptionHandlerTest.kt
@@ -2,22 +2,20 @@ package kt.fluxo.tests
 
 import kotlinx.coroutines.CoroutineExceptionHandler
 import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.cancel
-import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.isActive
-import kotlinx.coroutines.withContext
 import kotlinx.coroutines.yield
 import kt.fluxo.core.FluxoClosedException
 import kt.fluxo.core.closeAndWait
 import kt.fluxo.core.container
 import kt.fluxo.core.updateState
+import kt.fluxo.test.delaySafe
 import kt.fluxo.test.getValue
 import kt.fluxo.test.runUnitTest
 import kt.fluxo.test.setValue
@@ -116,9 +114,7 @@ class ContainerExceptionHandlerTest {
                 flow {
                     while (true) {
                         emit(Unit)
-                        withContext(Dispatchers.Default) {
-                            delay(1000)
-                        }
+                        delaySafe(1000)
                     }
                 }.collect()
             } catch (ce: CancellationException) {

diff --git a/fluxo-core/src/commonTest/kotlin/kt/fluxo/tests/ContainerThreadingTest.kt b/fluxo-core/src/commonTest/kotlin/kt/fluxo/tests/ContainerThreadingTest.kt
@@ -1,13 +1,13 @@
 package kt.fluxo.tests
 
 import kotlinx.coroutines.coroutineScope
-import kotlinx.coroutines.delay
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.test.runTest
 import kt.fluxo.core.Container
 import kt.fluxo.core.container
 import kt.fluxo.core.updateState
 import kt.fluxo.test.CoroutineScopeAwareTest
+import kt.fluxo.test.delaySafe
 import kt.fluxo.test.test
 import kotlin.random.Random
 import kotlin.test.Test
@@ -24,7 +24,7 @@ internal class ContainerThreadingTest : CoroutineScopeAwareTest() {
         val newState = Random.nextInt()
 
         container.send {
-            delay(Long.MAX_VALUE)
+            delaySafe(Long.MAX_VALUE)
         }
         container.send {
             value = newState
@@ -104,7 +104,7 @@ internal class ContainerThreadingTest : CoroutineScopeAwareTest() {
 
     private fun Container<TestState, Nothing>.one(delay: Boolean = false) = send {
         if (delay) {
-            delay(Random.nextLong(20))
+            delaySafe(Random.nextLong(20))
         }
         updateState {
             it.copy(ids = value.ids + 1)
@@ -113,7 +113,7 @@ internal class ContainerThreadingTest : CoroutineScopeAwareTest() {
 
     private fun Container<TestState, Nothing>.two(delay: Boolean = false) = send {
         if (delay) {
-            delay(Random.nextLong(20))
+            delaySafe(Random.nextLong(20))
         }
         updateState {
             it.copy(ids = value.ids + 2)
@@ -122,7 +122,7 @@ internal class ContainerThreadingTest : CoroutineScopeAwareTest() {
 
     private fun Container<TestState, Nothing>.three(delay: Boolean = false) = send {
         if (delay) {
-            delay(Random.nextLong(20))
+            delaySafe(Random.nextLong(20))
         }
         updateState {
             it.copy(ids = value.ids + 3)

diff --git a/fluxo-core/src/commonTest/kotlin/kt/fluxo/tests/InputStrategyTest.kt b/fluxo-core/src/commonTest/kotlin/kt/fluxo/tests/InputStrategyTest.kt
@@ -3,7 +3,6 @@ package kt.fluxo.tests
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.channels.Channel
-import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.collectLatest
 import kotlinx.coroutines.flow.consumeAsFlow
 import kotlinx.coroutines.flow.first
@@ -24,6 +23,7 @@ import kt.fluxo.core.store
 import kt.fluxo.core.updateState
 import kt.fluxo.test.CoroutineScopeAwareTest
 import kt.fluxo.test.IgnoreNativeAndJs
+import kt.fluxo.test.delaySafe
 import kt.fluxo.test.runUnitTest
 import kotlin.random.Random
 import kotlin.test.Test
@@ -50,11 +50,11 @@ internal class InputStrategyTest : CoroutineScopeAwareTest() {
 
         val results = ArrayList<Int>()
         val store = this.store(ints.first(), handler = { intent: Int ->
-            if (!generic) delay(timeMillis = 1)
+            if (!generic) delaySafe(timeMillis = 1)
             resLock.withLock { results.add(intent) }
             value = intent
             if (intent == ints.last()) {
-                if (generic) delay(timeMillis = 10)
+                if (generic) delaySafe(timeMillis = 10)
                 finishLock.unlock()
             }
         }) { inputStrategy = strategy }
@@ -65,7 +65,7 @@ internal class InputStrategyTest : CoroutineScopeAwareTest() {
         if (generic) {
             // TODO: Store.awaitIdle()
             while (results.size < NUMBER_OF_ITEMS) {
-                delay(timeMillis = 10)
+                delaySafe(timeMillis = 10)
             }
         }
         store.closeAndWait()

diff --git a/fluxo-core/src/commonTest/kotlin/kt/fluxo/tests/RepeatOnSubscriptionTest.kt b/fluxo-core/src/commonTest/kotlin/kt/fluxo/tests/RepeatOnSubscriptionTest.kt
@@ -1,6 +1,5 @@
 package kt.fluxo.tests
 
-import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.take
 import kotlinx.coroutines.flow.toList
 import kt.fluxo.core.Bootstrapper
@@ -14,6 +13,7 @@ import kt.fluxo.core.repeatOnSubscription
 import kt.fluxo.core.updateState
 import kt.fluxo.test.CoroutineScopeAwareTest
 import kt.fluxo.test.IgnoreNative
+import kt.fluxo.test.delaySafe
 import kt.fluxo.test.runUnitTest
 import kotlin.test.Test
 import kotlin.test.assertContentEquals
@@ -61,7 +61,7 @@ internal class RepeatOnSubscriptionTest : CoroutineScopeAwareTest() {
         val states = container.take(2).toList()
         assertContentEquals(listOf(initialState, State(1)), states)
 
-        delay(stopTimeout)
+        delaySafe(stopTimeout)
         val states2 = container.take(2).toList()
         assertContentEquals(listOf(State(1), State(2)), states2)
 

diff --git a/fluxo-core/src/commonTest/kotlin/kt/fluxo/tests/SideJobTest.kt b/fluxo-core/src/commonTest/kotlin/kt/fluxo/tests/SideJobTest.kt
@@ -9,6 +9,7 @@ import kt.fluxo.core.dsl.accept
 import kt.fluxo.core.intent
 import kt.fluxo.core.store
 import kt.fluxo.core.updateState
+import kt.fluxo.test.delaySafe
 import kt.fluxo.test.runUnitTest
 import kotlin.test.Test
 import kotlin.test.assertEquals
@@ -131,7 +132,7 @@ internal class SideJobTest {
         store.intent {
             sideJob { wasRestarted ->
                 assertFalse(wasRestarted)
-                delay(timeMillis = 1_000)
+                delaySafe(timeMillis = 1_000)
                 updateState { "a" }
             }
             sideJob { wasRestarted ->

diff --git a/fluxo-core/src/commonTest/kotlin/kt/fluxo/tests/dsl/CoroutinesTest.kt b/fluxo-core/src/commonTest/kotlin/kt/fluxo/tests/dsl/CoroutinesTest.kt
@@ -51,13 +51,15 @@ class CoroutinesTest {
             repeat(times = 10_000) {
                 val mutex = Mutex(locked = true)
                 val job = backgroundScope.launch(Dispatchers.Default) {
+                    // Deliberately time-skipping delay here! (Dispatchers.Default used)
                     delay(10)
                     mutex.unlock()
                     // busy wait
                     @Suppress("ControlFlowWithEmptyBody", "EmptyWhileBlock")
                     while (currentCoroutineContext().isActive) {
                     }
                 }
+                // Deliberately unsafe withTimeout here!
                 withTimeout(TIMEOUT_MS) {
                     mutex.withLock {}
                 }

diff --git a/fluxo-core/src/commonTest/kotlin/kt/fluxo/tests/dsl/DslThreadingTest.kt b/fluxo-core/src/commonTest/kotlin/kt/fluxo/tests/dsl/DslThreadingTest.kt
@@ -8,7 +8,6 @@ import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers.Default
 import kotlinx.coroutines.cancelAndJoin
 import kotlinx.coroutines.currentCoroutineContext
-import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.first
@@ -29,6 +28,7 @@ import kt.fluxo.test.IgnoreJs
 import kt.fluxo.test.KMM_PLATFORM
 import kt.fluxo.test.Platform
 import kt.fluxo.test.TestLoggingStoreFactory
+import kt.fluxo.test.delaySafe
 import kt.fluxo.test.runUnitTest
 import kt.fluxo.test.testLog
 import kt.fluxo.test.withTimeoutSafe
@@ -307,7 +307,7 @@ internal class DslThreadingTest {
 
         fun suspendingIntent() = intent {
             intentMutex.doUnlock()
-            delay(Int.MAX_VALUE.toLong())
+            delaySafe(Int.MAX_VALUE.toLong())
         }
 
         fun simpleIntent() = intent {