diff --git a/kotlinx-coroutines-core/api/kotlinx-coroutines-core.api b/kotlinx-coroutines-core/api/kotlinx-coroutines-core.api index d227eb879b..57b33056b3 100644 --- a/kotlinx-coroutines-core/api/kotlinx-coroutines-core.api +++ b/kotlinx-coroutines-core/api/kotlinx-coroutines-core.api @@ -381,6 +381,12 @@ public final class kotlinx/coroutines/Job$DefaultImpls { public final class kotlinx/coroutines/Job$Key : kotlin/coroutines/CoroutineContext$Key { } +public class kotlinx/coroutines/JobImpl : kotlinx/coroutines/JobSupport, kotlinx/coroutines/CompletableJob { + public fun (Lkotlinx/coroutines/Job;)V + public fun complete ()Z + public fun completeExceptionally (Ljava/lang/Throwable;)Z +} + public final class kotlinx/coroutines/JobKt { public static final fun Job (Lkotlinx/coroutines/Job;)Lkotlinx/coroutines/CompletableJob; public static final synthetic fun Job (Lkotlinx/coroutines/Job;)Lkotlinx/coroutines/Job; diff --git a/kotlinx-coroutines-core/common/src/JobSupport.kt b/kotlinx-coroutines-core/common/src/JobSupport.kt index 0a3dd23472..1b5975c8bc 100644 --- a/kotlinx-coroutines-core/common/src/JobSupport.kt +++ b/kotlinx-coroutines-core/common/src/JobSupport.kt @@ -1312,6 +1312,7 @@ private class Empty(override val isActive: Boolean) : Incomplete { override fun toString(): String = "Empty{${if (isActive) "Active" else "New" }}" } +@PublishedApi // for a custom job in the test module internal open class JobImpl(parent: Job?) : JobSupport(true), CompletableJob { init { initParentJob(parent) } override val onCancelComplete get() = true diff --git a/kotlinx-coroutines-core/common/src/internal/StackTraceRecovery.common.kt b/kotlinx-coroutines-core/common/src/internal/StackTraceRecovery.common.kt index 8e0b558cc4..2812b931de 100644 --- a/kotlinx-coroutines-core/common/src/internal/StackTraceRecovery.common.kt +++ b/kotlinx-coroutines-core/common/src/internal/StackTraceRecovery.common.kt @@ -40,7 +40,7 @@ internal expect suspend inline fun recoverAndThrow(exception: Throwable): Nothin * The opposite of [recoverStackTrace]. * It is guaranteed that `unwrap(recoverStackTrace(e)) === e` */ -@PublishedApi // only published for the multiplatform tests in our own code +@PublishedApi // published for the multiplatform implementation of kotlinx-coroutines-test internal expect fun unwrap(exception: E): E internal expect class StackTraceElement diff --git a/kotlinx-coroutines-core/common/src/internal/ThreadSafeHeap.kt b/kotlinx-coroutines-core/common/src/internal/ThreadSafeHeap.kt index 2100454be3..ad88103142 100644 --- a/kotlinx-coroutines-core/common/src/internal/ThreadSafeHeap.kt +++ b/kotlinx-coroutines-core/common/src/internal/ThreadSafeHeap.kt @@ -37,6 +37,16 @@ public open class ThreadSafeHeap : SynchronizedObject() where T: ThreadSafeHe _size.value = 0 } + public fun find( + predicate: (value: T) -> Boolean + ): T? = synchronized(this) block@{ + for (i in 0 until size) { + val value = a?.get(i)!! + if (predicate(value)) return@block value + } + null + } + public fun peek(): T? = synchronized(this) { firstImpl() } public fun removeFirstOrNull(): T? = synchronized(this) { diff --git a/kotlinx-coroutines-test/api/kotlinx-coroutines-test.api b/kotlinx-coroutines-test/api/kotlinx-coroutines-test.api index 4786b81bf9..bf639235d0 100644 --- a/kotlinx-coroutines-test/api/kotlinx-coroutines-test.api +++ b/kotlinx-coroutines-test/api/kotlinx-coroutines-test.api @@ -105,6 +105,7 @@ public final class kotlinx/coroutines/test/TestDispatchers { } public abstract interface class kotlinx/coroutines/test/TestScope : kotlinx/coroutines/CoroutineScope { + public abstract fun getBackgroundScope ()Lkotlinx/coroutines/CoroutineScope; public abstract fun getTestScheduler ()Lkotlinx/coroutines/test/TestCoroutineScheduler; } diff --git a/kotlinx-coroutines-test/common/src/TestBuilders.kt b/kotlinx-coroutines-test/common/src/TestBuilders.kt index d8e9357611..80dc8d6083 100644 --- a/kotlinx-coroutines-test/common/src/TestBuilders.kt +++ b/kotlinx-coroutines-test/common/src/TestBuilders.kt @@ -164,7 +164,11 @@ public fun TestScope.runTest( ): TestResult = asSpecificImplementation().let { it.enter() createTestResult { - runTestCoroutine(it, dispatchTimeoutMs, TestScopeImpl::tryGetCompletionCause, testBody) { it.leave() } + runTestCoroutine(it, dispatchTimeoutMs, TestScopeImpl::tryGetCompletionCause, testBody) { + backgroundScope.cancel() + testScheduler.advanceUntilIdleOr { false } + it.leave() + } } } @@ -172,7 +176,7 @@ public fun TestScope.runTest( * Runs [testProcedure], creating a [TestResult]. */ @Suppress("NO_ACTUAL_FOR_EXPECT") // actually suppresses `TestResult` -internal expect fun createTestResult(testProcedure: suspend () -> Unit): TestResult +internal expect fun createTestResult(testProcedure: suspend CoroutineScope.() -> Unit): TestResult /** A coroutine context element indicating that the coroutine is running inside `runTest`. */ internal object RunningInRunTest : CoroutineContext.Key, CoroutineContext.Element { @@ -195,7 +199,7 @@ internal const val DEFAULT_DISPATCH_TIMEOUT_MS = 60_000L * The [cleanup] procedure may either throw [UncompletedCoroutinesError] to denote that child coroutines were leaked, or * return a list of uncaught exceptions that should be reported at the end of the test. */ -internal suspend fun > runTestCoroutine( +internal suspend fun > CoroutineScope.runTestCoroutine( coroutine: T, dispatchTimeoutMs: Long, tryGetCompletionCause: T.() -> Throwable?, @@ -207,6 +211,27 @@ internal suspend fun > runTestCoroutine( coroutine.start(CoroutineStart.UNDISPATCHED, coroutine) { testBody() } + /** + * The general procedure here is as follows: + * 1. Try running the work that the scheduler knows about, both background and foreground. + * + * 2. Wait until we run out of foreground work to do. This could mean one of the following: + * * The main coroutine is already completed. This is checked separately; then we leave the procedure. + * * It's switched to another dispatcher that doesn't know about the [TestCoroutineScheduler]. + * * Generally, it's waiting for something external (like a network request, or just an arbitrary callback). + * * The test simply hanged. + * * The main coroutine is waiting for some background work. + * + * 3. We await progress from things that are not the code under test: + * the background work that the scheduler knows about, the external callbacks, + * the work on dispatchers not linked to the scheduler, etc. + * + * When we observe that the code under test can proceed, we go to step 1 again. + * If there is no activity for [dispatchTimeoutMs] milliseconds, we consider the test to have hanged. + * + * The background work is not running on a dedicated thread. + * Instead, the test thread itself is used, by spawning a separate coroutine. + */ var completed = false while (!completed) { scheduler.advanceUntilIdle() @@ -216,16 +241,29 @@ internal suspend fun > runTestCoroutine( completed = true continue } - select { - coroutine.onJoin { - completed = true - } - scheduler.onDispatchEvent { - // we received knowledge that `scheduler` observed a dispatch event, so we reset the timeout + // in case progress depends on some background work, we need to keep spinning it. + val backgroundWorkRunner = launch(CoroutineName("background work runner")) { + while (true) { + scheduler.tryRunNextTaskUnless { !isActive } + // yield so that the `select` below has a chance to check if its conditions are fulfilled + yield() } - onTimeout(dispatchTimeoutMs) { - handleTimeout(coroutine, dispatchTimeoutMs, tryGetCompletionCause, cleanup) + } + try { + select { + coroutine.onJoin { + // observe that someone completed the test coroutine and leave without waiting for the timeout + completed = true + } + scheduler.onDispatchEvent { + // we received knowledge that `scheduler` observed a dispatch event, so we reset the timeout + } + onTimeout(dispatchTimeoutMs) { + handleTimeout(coroutine, dispatchTimeoutMs, tryGetCompletionCause, cleanup) + } } + } finally { + backgroundWorkRunner.cancelAndJoin() } } coroutine.getCompletionExceptionOrNull()?.let { exception -> diff --git a/kotlinx-coroutines-test/common/src/TestCoroutineDispatchers.kt b/kotlinx-coroutines-test/common/src/TestCoroutineDispatchers.kt index 4cc48f47d0..e99fe8b124 100644 --- a/kotlinx-coroutines-test/common/src/TestCoroutineDispatchers.kt +++ b/kotlinx-coroutines-test/common/src/TestCoroutineDispatchers.kt @@ -96,7 +96,7 @@ private class UnconfinedTestDispatcherImpl( @Suppress("INVISIBLE_MEMBER") override fun dispatch(context: CoroutineContext, block: Runnable) { checkSchedulerInContext(scheduler, context) - scheduler.sendDispatchEvent() + scheduler.sendDispatchEvent(context) /** copy-pasted from [kotlinx.coroutines.Unconfined.dispatch] */ /** It can only be called by the [yield] function. See also code of [yield] function. */ @@ -151,8 +151,7 @@ private class StandardTestDispatcherImpl( ) : TestDispatcher() { override fun dispatch(context: CoroutineContext, block: Runnable) { - checkSchedulerInContext(scheduler, context) - scheduler.registerEvent(this, 0, block) { false } + scheduler.registerEvent(this, 0, block, context) { false } } override fun toString(): String = "${name ?: "StandardTestDispatcher"}[scheduler=$scheduler]" diff --git a/kotlinx-coroutines-test/common/src/TestCoroutineScheduler.kt b/kotlinx-coroutines-test/common/src/TestCoroutineScheduler.kt index 9aa90fac1d..e735c6d4de 100644 --- a/kotlinx-coroutines-test/common/src/TestCoroutineScheduler.kt +++ b/kotlinx-coroutines-test/common/src/TestCoroutineScheduler.kt @@ -62,17 +62,20 @@ public class TestCoroutineScheduler : AbstractCoroutineContextElement(TestCorout dispatcher: TestDispatcher, timeDeltaMillis: Long, marker: T, + context: CoroutineContext, isCancelled: (T) -> Boolean ): DisposableHandle { require(timeDeltaMillis >= 0) { "Attempted scheduling an event earlier in time (with the time delta $timeDeltaMillis)" } + checkSchedulerInContext(this, context) val count = count.getAndIncrement() + val isForeground = context[BackgroundWork] === null return synchronized(lock) { val time = addClamping(currentTime, timeDeltaMillis) - val event = TestDispatchEvent(dispatcher, count, time, marker as Any) { isCancelled(marker) } + val event = TestDispatchEvent(dispatcher, count, time, marker as Any, isForeground) { isCancelled(marker) } events.addLast(event) /** can't be moved above: otherwise, [onDispatchEvent] could consume the token sent here before there's * actually anything in the event queue. */ - sendDispatchEvent() + sendDispatchEvent(context) DisposableHandle { synchronized(lock) { events.remove(event) @@ -82,10 +85,12 @@ public class TestCoroutineScheduler : AbstractCoroutineContextElement(TestCorout } /** - * Runs the next enqueued task, advancing the virtual time to the time of its scheduled awakening. + * Runs the next enqueued task, advancing the virtual time to the time of its scheduled awakening, + * unless [condition] holds. */ - private fun tryRunNextTask(): Boolean { + internal fun tryRunNextTaskUnless(condition: () -> Boolean): Boolean { val event = synchronized(lock) { + if (condition()) return false val event = events.removeFirstOrNull() ?: return false if (currentTime > event.time) currentTimeAheadOfEvents() @@ -105,9 +110,15 @@ public class TestCoroutineScheduler : AbstractCoroutineContextElement(TestCorout * functionality, query [currentTime] before and after the execution to achieve the same result. */ @ExperimentalCoroutinesApi - public fun advanceUntilIdle() { - while (!synchronized(lock) { events.isEmpty }) { - tryRunNextTask() + public fun advanceUntilIdle(): Unit = advanceUntilIdleOr { events.none(TestDispatchEvent<*>::isForeground) } + + /** + * [condition]: guaranteed to be invoked under the lock. + */ + internal fun advanceUntilIdleOr(condition: () -> Boolean) { + while (true) { + if (!tryRunNextTaskUnless(condition)) + return } } @@ -169,24 +180,19 @@ public class TestCoroutineScheduler : AbstractCoroutineContextElement(TestCorout /** * Checks that the only tasks remaining in the scheduler are cancelled. */ - internal fun isIdle(strict: Boolean = true): Boolean { + internal fun isIdle(strict: Boolean = true): Boolean = synchronized(lock) { - if (strict) - return events.isEmpty - // TODO: also completely empties the queue, as there's no nondestructive way to iterate over [ThreadSafeHeap] - val presentEvents = mutableListOf>() - while (true) { - presentEvents += events.removeFirstOrNull() ?: break - } - return presentEvents.all { it.isCancelled() } + if (strict) events.isEmpty else events.none { !it.isCancelled() } } - } /** * Notifies this scheduler about a dispatch event. + * + * [context] is the context in which the task will be dispatched. */ - internal fun sendDispatchEvent() { - dispatchEvents.trySend(Unit) + internal fun sendDispatchEvent(context: CoroutineContext) { + if (context[BackgroundWork] !== BackgroundWork) + dispatchEvents.trySend(Unit) } /** @@ -216,6 +222,8 @@ private class TestDispatchEvent( private val count: Long, @JvmField val time: Long, @JvmField val marker: T, + @JvmField val isForeground: Boolean, + // TODO: remove once the deprecated API is gone @JvmField val isCancelled: () -> Boolean ) : Comparable>, ThreadSafeHeapNode { override var heap: ThreadSafeHeap<*>? = null @@ -224,7 +232,7 @@ private class TestDispatchEvent( override fun compareTo(other: TestDispatchEvent<*>) = compareValuesBy(this, other, TestDispatchEvent<*>::time, TestDispatchEvent<*>::count) - override fun toString() = "TestDispatchEvent(time=$time, dispatcher=$dispatcher)" + override fun toString() = "TestDispatchEvent(time=$time, dispatcher=$dispatcher${if (isForeground) "" else ", background"})" } // works with positive `a`, `b` @@ -238,3 +246,17 @@ internal fun checkSchedulerInContext(scheduler: TestCoroutineScheduler, context: } } } + +/** + * A coroutine context key denoting that the work is to be executed in the background. + * @see [TestScope.backgroundScope] + */ +internal object BackgroundWork : CoroutineContext.Key, CoroutineContext.Element { + override val key: CoroutineContext.Key<*> + get() = this + + override fun toString(): String = "BackgroundWork" +} + +private fun ThreadSafeHeap.none(predicate: (T) -> Boolean) where T: ThreadSafeHeapNode, T: Comparable = + find(predicate) == null diff --git a/kotlinx-coroutines-test/common/src/TestDispatcher.kt b/kotlinx-coroutines-test/common/src/TestDispatcher.kt index f434572663..348cc2f185 100644 --- a/kotlinx-coroutines-test/common/src/TestDispatcher.kt +++ b/kotlinx-coroutines-test/common/src/TestDispatcher.kt @@ -10,14 +10,14 @@ import kotlin.jvm.* /** * A test dispatcher that can interface with a [TestCoroutineScheduler]. - * + * * The available implementations are: * * [StandardTestDispatcher] is a dispatcher that places new tasks into a queue. * * [UnconfinedTestDispatcher] is a dispatcher that behaves like [Dispatchers.Unconfined] while allowing to control * the virtual time. */ @ExperimentalCoroutinesApi -public abstract class TestDispatcher internal constructor(): CoroutineDispatcher(), Delay { +public abstract class TestDispatcher internal constructor() : CoroutineDispatcher(), Delay { /** The scheduler that this dispatcher is linked to. */ @ExperimentalCoroutinesApi public abstract val scheduler: TestCoroutineScheduler @@ -30,16 +30,13 @@ public abstract class TestDispatcher internal constructor(): CoroutineDispatcher /** @suppress */ override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation) { - checkSchedulerInContext(scheduler, continuation.context) val timedRunnable = CancellableContinuationRunnable(continuation, this) - scheduler.registerEvent(this, timeMillis, timedRunnable, ::cancellableRunnableIsCancelled) + scheduler.registerEvent(this, timeMillis, timedRunnable, continuation.context, ::cancellableRunnableIsCancelled) } /** @suppress */ - override fun invokeOnTimeout(timeMillis: Long, block: Runnable, context: CoroutineContext): DisposableHandle { - checkSchedulerInContext(scheduler, context) - return scheduler.registerEvent(this, timeMillis, block) { false } - } + override fun invokeOnTimeout(timeMillis: Long, block: Runnable, context: CoroutineContext): DisposableHandle = + scheduler.registerEvent(this, timeMillis, block, context) { false } } /** diff --git a/kotlinx-coroutines-test/common/src/TestScope.kt b/kotlinx-coroutines-test/common/src/TestScope.kt index 60585a1d50..15d48a2ae2 100644 --- a/kotlinx-coroutines-test/common/src/TestScope.kt +++ b/kotlinx-coroutines-test/common/src/TestScope.kt @@ -6,6 +6,7 @@ package kotlinx.coroutines.test import kotlinx.coroutines.* import kotlinx.coroutines.internal.* +import kotlinx.coroutines.test.internal.* import kotlin.coroutines.* import kotlin.time.* @@ -46,6 +47,43 @@ public sealed interface TestScope : CoroutineScope { */ @ExperimentalCoroutinesApi public val testScheduler: TestCoroutineScheduler + + /** + * A scope for background work. + * + * This scope is automatically cancelled when the test finishes. + * Additionally, while the coroutines in this scope are run as usual when + * using [advanceTimeBy] and [runCurrent], [advanceUntilIdle] will stop advancing the virtual time + * once only the coroutines in this scope are left unprocessed. + * + * Failures in coroutines in this scope do not terminate the test. + * Instead, they are reported at the end of the test. + * Likewise, failure in the [TestScope] itself will not affect its [backgroundScope], + * because there's no parent-child relationship between them. + * + * A typical use case for this scope is to launch tasks that would outlive the tested code in + * the production environment. + * + * In this example, the coroutine that continuously sends new elements to the channel will get + * cancelled: + * ``` + * @Test + * fun testExampleBackgroundJob() = runTest { + * val channel = Channel() + * backgroundScope.launch { + * var i = 0 + * while (true) { + * channel.send(i++) + * } + * } + * repeat(100) { + * assertEquals(it, channel.receive()) + * } + * } + * ``` + */ + @ExperimentalCoroutinesApi + public val backgroundScope: CoroutineScope } /** @@ -170,6 +208,11 @@ internal class TestScopeImpl(context: CoroutineContext) : private val uncaughtExceptions = mutableListOf() private val lock = SynchronizedObject() + override val backgroundScope: CoroutineScope = + CoroutineScope(coroutineContext + BackgroundWork + ReportingSupervisorJob { + if (it !is CancellationException) reportException(it) + }) + /** Called upon entry to [runTest]. Will throw if called more than once. */ fun enter() { val exceptions = synchronized(lock) { @@ -190,8 +233,7 @@ internal class TestScopeImpl(context: CoroutineContext) : /** Called at the end of the test. May only be called once. */ fun leave(): List { val exceptions = synchronized(lock) { - if(!entered || finished) - throw IllegalStateException("An internal error. Please report to the Kotlinx Coroutines issue tracker") + check(entered && !finished) finished = true uncaughtExceptions } @@ -218,6 +260,12 @@ internal class TestScopeImpl(context: CoroutineContext) : if (finished) { throw throwable } else { + @Suppress("INVISIBLE_MEMBER") + for (existingThrowable in uncaughtExceptions) { + // avoid reporting exceptions that already were reported. + if (unwrap(throwable) == unwrap(existingThrowable)) + return + } uncaughtExceptions.add(throwable) if (!entered) throw UncaughtExceptionsBeforeTest().apply { addSuppressed(throwable) } @@ -233,6 +281,7 @@ internal class TestScopeImpl(context: CoroutineContext) : } /** Use the knowledge that any [TestScope] that we receive is necessarily a [TestScopeImpl]. */ +@Suppress("NO_ELSE_IN_WHEN") // TODO: a problem with `sealed` in MPP not allowing total pattern-matching internal fun TestScope.asSpecificImplementation(): TestScopeImpl = when (this) { is TestScopeImpl -> this } diff --git a/kotlinx-coroutines-test/common/src/internal/ReportingSupervisorJob.kt b/kotlinx-coroutines-test/common/src/internal/ReportingSupervisorJob.kt new file mode 100644 index 0000000000..e3091bc263 --- /dev/null +++ b/kotlinx-coroutines-test/common/src/internal/ReportingSupervisorJob.kt @@ -0,0 +1,24 @@ +/* + * Copyright 2016-2022 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license. + */ + +package kotlinx.coroutines.test.internal + +import kotlinx.coroutines.* + +/** + * A variant of [SupervisorJob] that additionally notifies about child failures via a callback. + */ +@Suppress("INVISIBLE_MEMBER", "INVISIBLE_REFERENCE", "CANNOT_OVERRIDE_INVISIBLE_MEMBER") +internal class ReportingSupervisorJob(parent: Job? = null, val onChildCancellation: (Throwable) -> Unit) : + JobImpl(parent) { + override fun childCancelled(cause: Throwable): Boolean = + try { + onChildCancellation(cause) + } catch (e: Throwable) { + cause.addSuppressed(e) + /* the coroutine context does not matter here, because we're only interested in reporting this exception + to the platform-specific global handler, not to a [CoroutineExceptionHandler] of any sort. */ + handleCoroutineException(this, cause) + }.let { false } +} diff --git a/kotlinx-coroutines-test/common/test/RunTestTest.kt b/kotlinx-coroutines-test/common/test/RunTestTest.kt index 3b6272c062..1430d830cc 100644 --- a/kotlinx-coroutines-test/common/test/RunTestTest.kt +++ b/kotlinx-coroutines-test/common/test/RunTestTest.kt @@ -71,6 +71,7 @@ class RunTestTest { /** Tests that a dispatch timeout of `0` will fail the test if there are some dispatches outside the scheduler. */ @Test + @NoNative // TODO: timeout leads to `Cannot execute task because event loop was shut down` on Native fun testRunTestWithZeroTimeoutWithUncontrolledDispatches() = testResultMap({ fn -> assertFailsWith { fn() } }) { @@ -357,4 +358,15 @@ class RunTestTest { } } } + + /** + * Tests that if the main coroutine is completed without a dispatch, [runTest] will not consider this to be + * inactivity. + * + * The test will hang if this is not the case. + */ + @Test + fun testCoroutineCompletingWithoutDispatch() = runTest(dispatchTimeoutMs = Long.MAX_VALUE) { + launch(Dispatchers.Default) { delay(100) } + } } diff --git a/kotlinx-coroutines-test/common/test/TestScopeTest.kt b/kotlinx-coroutines-test/common/test/TestScopeTest.kt index 7031056f11..4138ca058f 100644 --- a/kotlinx-coroutines-test/common/test/TestScopeTest.kt +++ b/kotlinx-coroutines-test/common/test/TestScopeTest.kt @@ -5,11 +5,13 @@ package kotlinx.coroutines.test import kotlinx.coroutines.* +import kotlinx.coroutines.channels.* +import kotlinx.coroutines.flow.* import kotlin.coroutines.* import kotlin.test.* class TestScopeTest { - /** Tests failing to create a [TestCoroutineScope] with incorrect contexts. */ + /** Tests failing to create a [TestScope] with incorrect contexts. */ @Test fun testCreateThrowsOnInvalidArguments() { for (ctx in invalidContexts) { @@ -19,7 +21,7 @@ class TestScopeTest { } } - /** Tests that a newly-created [TestCoroutineScope] provides the correct scheduler. */ + /** Tests that a newly-created [TestScope] provides the correct scheduler. */ @Test fun testCreateProvidesScheduler() { // Creates a new scheduler. @@ -169,6 +171,312 @@ class TestScopeTest { } } + /** Tests that the background work is being run at all. */ + @Test + fun testBackgroundWorkBeingRun(): TestResult = runTest { + var i = 0 + var j = 0 + backgroundScope.launch { + ++i + } + backgroundScope.launch { + delay(10) + ++j + } + assertEquals(0, i) + assertEquals(0, j) + delay(1) + assertEquals(1, i) + assertEquals(0, j) + delay(10) + assertEquals(1, i) + assertEquals(1, j) + } + + /** + * Tests that the background work gets cancelled after the test body finishes. + */ + @Test + fun testBackgroundWorkCancelled(): TestResult { + var cancelled = false + return testResultMap({ + it() + assertTrue(cancelled) + }) { + runTest { + var i = 0 + backgroundScope.launch { + try { + while (isActive) { + ++i + yield() + } + } catch (e: CancellationException) { + cancelled = true + } + } + repeat(5) { + assertEquals(i, it) + yield() + } + } + } + } + + /** Tests the interactions between the time-control commands and the background work. */ + @Test + fun testBackgroundWorkTimeControl(): TestResult = runTest { + var i = 0 + var j = 0 + backgroundScope.launch { + while (true) { + ++i + delay(100) + } + } + backgroundScope.launch { + while (true) { + ++j + delay(50) + } + } + advanceUntilIdle() // should do nothing, as only background work is left. + assertEquals(0, i) + assertEquals(0, j) + val job = launch { + delay(1) + // the background work scheduled for earlier gets executed before the normal work scheduled for later does + assertEquals(1, i) + assertEquals(1, j) + } + job.join() + advanceTimeBy(199) // should work the same for the background tasks + assertEquals(2, i) + assertEquals(4, j) + advanceUntilIdle() // once again, should do nothing + assertEquals(2, i) + assertEquals(4, j) + runCurrent() // should behave the same way as for the normal work + assertEquals(3, i) + assertEquals(5, j) + launch { + delay(1001) + assertEquals(13, i) + assertEquals(25, j) + } + advanceUntilIdle() // should execute the normal work, and with that, the background one, too + } + + /** + * Tests that an error in a background coroutine does not cancel the test, but is reported at the end. + */ + @Test + fun testBackgroundWorkErrorReporting(): TestResult { + var testFinished = false + val exception = RuntimeException("x") + return testResultMap({ + try { + it() + fail("unreached") + } catch (e: Throwable) { + assertSame(e, exception) + assertTrue(testFinished) + } + }) { + runTest { + backgroundScope.launch { + throw exception + } + delay(1000) + testFinished = true + } + } + } + + /** + * Tests that the background work gets to finish what it's doing after the test is completed. + */ + @Test + fun testBackgroundWorkFinalizing(): TestResult { + var taskEnded = 0 + val nTasks = 10 + return testResultMap({ + try { + it() + fail("unreached") + } catch (e: TestException) { + assertEquals(2, e.suppressedExceptions.size) + assertEquals(nTasks, taskEnded) + } + }) { + runTest { + repeat(nTasks) { + backgroundScope.launch { + try { + while (true) { + delay(1) + } + } finally { + ++taskEnded + if (taskEnded <= 2) + throw TestException() + } + } + } + delay(100) + throw TestException() + } + } + } + + /** + * Tests using [Flow.stateIn] as a background job. + */ + @Test + fun testExampleBackgroundJob1() = runTest { + val myFlow = flow { + var i = 0 + while (true) { + emit(++i) + delay(1) + } + } + val stateFlow = myFlow.stateIn(backgroundScope, SharingStarted.Eagerly, 0) + var j = 0 + repeat(100) { + assertEquals(j++, stateFlow.value) + delay(1) + } + } + + /** + * A test from the documentation of [TestScope.backgroundScope]. + */ + @Test + fun testExampleBackgroundJob2() = runTest { + val channel = Channel() + backgroundScope.launch { + var i = 0 + while (true) { + channel.send(i++) + } + } + repeat(100) { + assertEquals(it, channel.receive()) + } + } + + /** + * Tests that the test will timeout due to idleness even if some background tasks are running. + */ + @Test + fun testBackgroundWorkNotPreventingTimeout(): TestResult = testResultMap({ + try { + it() + fail("unreached") + } catch (_: UncompletedCoroutinesError) { + + } + }) { + runTest(dispatchTimeoutMs = 100) { + backgroundScope.launch { + while (true) { + yield() + } + } + backgroundScope.launch { + while (true) { + delay(1) + } + } + val deferred = CompletableDeferred() + deferred.await() + } + + } + + /** + * Tests that the background work will not prevent the test from timing out even in some cases + * when the unconfined dispatcher is used. + */ + @Test + fun testUnconfinedBackgroundWorkNotPreventingTimeout(): TestResult = testResultMap({ + try { + it() + fail("unreached") + } catch (_: UncompletedCoroutinesError) { + + } + }) { + runTest(UnconfinedTestDispatcher(), dispatchTimeoutMs = 100) { + /** + * Having a coroutine like this will still cause the test to hang: + backgroundScope.launch { + while (true) { + yield() + } + } + * The reason is that even the initial [advanceUntilIdle] will never return in this case. + */ + backgroundScope.launch { + while (true) { + delay(1) + } + } + val deferred = CompletableDeferred() + deferred.await() + } + } + + /** + * Tests that even the exceptions in the background scope that don't typically get reported and need to be queried + * (like failures in [async]) will still surface in some simple scenarios. + */ + @Test + fun testAsyncFailureInBackgroundReported() = testResultMap({ + try { + it() + fail("unreached") + } catch (e: TestException) { + assertEquals("z", e.message) + assertEquals(setOf("x", "y"), e.suppressedExceptions.map { it.message }.toSet()) + } + }) { + runTest { + backgroundScope.async { + throw TestException("x") + } + backgroundScope.produce { + throw TestException("y") + } + delay(1) + throw TestException("z") + } + } + + /** + * Tests that, if an exception reaches the [TestScope] exception reporting mechanism via several + * channels, it will only be reported once. + */ + @Test + fun testNoDuplicateExceptions() = testResultMap({ + try { + it() + fail("unreached") + } catch (e: TestException) { + assertEquals("y", e.message) + assertEquals(listOf("x"), e.suppressedExceptions.map { it.message }) + } + }) { + runTest { + backgroundScope.launch { + throw TestException("x") + } + delay(1) + throw TestException("y") + } + } + companion object { internal val invalidContexts = listOf( Dispatchers.Default, // not a [TestDispatcher] diff --git a/kotlinx-coroutines-test/js/src/TestBuilders.kt b/kotlinx-coroutines-test/js/src/TestBuilders.kt index 3976885991..9da91ffc39 100644 --- a/kotlinx-coroutines-test/js/src/TestBuilders.kt +++ b/kotlinx-coroutines-test/js/src/TestBuilders.kt @@ -9,7 +9,7 @@ import kotlin.js.* @Suppress("ACTUAL_WITHOUT_EXPECT", "ACTUAL_TYPE_ALIAS_TO_CLASS_WITH_DECLARATION_SITE_VARIANCE") public actual typealias TestResult = Promise -internal actual fun createTestResult(testProcedure: suspend () -> Unit): TestResult = +internal actual fun createTestResult(testProcedure: suspend CoroutineScope.() -> Unit): TestResult = GlobalScope.promise { testProcedure() - } \ No newline at end of file + } diff --git a/kotlinx-coroutines-test/jvm/src/TestBuildersJvm.kt b/kotlinx-coroutines-test/jvm/src/TestBuildersJvm.kt index 7cafb54753..06fbe81064 100644 --- a/kotlinx-coroutines-test/jvm/src/TestBuildersJvm.kt +++ b/kotlinx-coroutines-test/jvm/src/TestBuildersJvm.kt @@ -8,8 +8,8 @@ import kotlinx.coroutines.* @Suppress("ACTUAL_WITHOUT_EXPECT") public actual typealias TestResult = Unit -internal actual fun createTestResult(testProcedure: suspend () -> Unit) { +internal actual fun createTestResult(testProcedure: suspend CoroutineScope.() -> Unit) { runBlocking { testProcedure() } -} \ No newline at end of file +} diff --git a/kotlinx-coroutines-test/jvm/src/migration/TestBuildersDeprecated.kt b/kotlinx-coroutines-test/jvm/src/migration/TestBuildersDeprecated.kt index a60d65c1b8..eabdffb2c8 100644 --- a/kotlinx-coroutines-test/jvm/src/migration/TestBuildersDeprecated.kt +++ b/kotlinx-coroutines-test/jvm/src/migration/TestBuildersDeprecated.kt @@ -87,11 +87,14 @@ public fun runBlockingTestOnTestScope( scope.testBody() } scope.testScheduler.advanceUntilIdle() - try { + val throwable = try { scope.getCompletionExceptionOrNull() } catch (e: IllegalStateException) { null // the deferred was not completed yet; `scope.leave()` should complain then about unfinished jobs - }?.let { + } + scope.backgroundScope.cancel() + scope.testScheduler.advanceUntilIdleOr { false } + throwable?.let { val exceptions = try { scope.leave() } catch (e: UncompletedCoroutinesError) { diff --git a/kotlinx-coroutines-test/jvm/src/migration/TestCoroutineDispatcher.kt b/kotlinx-coroutines-test/jvm/src/migration/TestCoroutineDispatcher.kt index ec2a3046ee..08f428f249 100644 --- a/kotlinx-coroutines-test/jvm/src/migration/TestCoroutineDispatcher.kt +++ b/kotlinx-coroutines-test/jvm/src/migration/TestCoroutineDispatcher.kt @@ -41,24 +41,24 @@ public class TestCoroutineDispatcher(public override val scheduler: TestCoroutin override fun dispatch(context: CoroutineContext, block: Runnable) { checkSchedulerInContext(scheduler, context) if (dispatchImmediately) { - scheduler.sendDispatchEvent() + scheduler.sendDispatchEvent(context) block.run() } else { - post(block) + post(block, context) } } /** @suppress */ override fun dispatchYield(context: CoroutineContext, block: Runnable) { checkSchedulerInContext(scheduler, context) - post(block) + post(block, context) } /** @suppress */ override fun toString(): String = "TestCoroutineDispatcher[scheduler=$scheduler]" - private fun post(block: Runnable) = - scheduler.registerEvent(this, 0, block) { false } + private fun post(block: Runnable, context: CoroutineContext) = + scheduler.registerEvent(this, 0, block, context) { false } /** @suppress */ override suspend fun pauseDispatcher(block: suspend () -> Unit) { diff --git a/kotlinx-coroutines-test/jvm/test/migration/RunBlockingTestOnTestScopeTest.kt b/kotlinx-coroutines-test/jvm/test/migration/RunBlockingTestOnTestScopeTest.kt index 174baa0819..806592079c 100644 --- a/kotlinx-coroutines-test/jvm/test/migration/RunBlockingTestOnTestScopeTest.kt +++ b/kotlinx-coroutines-test/jvm/test/migration/RunBlockingTestOnTestScopeTest.kt @@ -5,6 +5,7 @@ package kotlinx.coroutines.test import kotlinx.coroutines.* +import kotlinx.coroutines.channels.* import kotlinx.coroutines.flow.* import kotlin.test.* @@ -162,4 +163,211 @@ class RunBlockingTestOnTestScopeTest { } } } -} \ No newline at end of file + + @Test + fun testBackgroundWorkBeingRun() = runBlockingTestOnTestScope { + var i = 0 + var j = 0 + backgroundScope.launch { + yield() + ++i + } + backgroundScope.launch { + yield() + delay(10) + ++j + } + assertEquals(0, i) + assertEquals(0, j) + delay(1) + assertEquals(1, i) + assertEquals(0, j) + delay(10) + assertEquals(1, i) + assertEquals(1, j) + } + + @Test + fun testBackgroundWorkCancelled() { + var cancelled = false + runBlockingTestOnTestScope { + var i = 0 + backgroundScope.launch { + yield() + try { + while (isActive) { + ++i + yield() + } + } catch (e: CancellationException) { + cancelled = true + } + } + repeat(5) { + assertEquals(i, it) + yield() + } + } + assertTrue(cancelled) + } + + @Test + fun testBackgroundWorkTimeControl(): TestResult = runBlockingTestOnTestScope { + var i = 0 + var j = 0 + backgroundScope.launch { + yield() + while (true) { + ++i + delay(100) + } + } + backgroundScope.launch { + yield() + while (true) { + ++j + delay(50) + } + } + advanceUntilIdle() // should do nothing, as only background work is left. + assertEquals(0, i) + assertEquals(0, j) + val job = launch { + delay(1) + // the background work scheduled for earlier gets executed before the normal work scheduled for later does + assertEquals(1, i) + assertEquals(1, j) + } + job.join() + advanceTimeBy(199) // should work the same for the background tasks + assertEquals(2, i) + assertEquals(4, j) + advanceUntilIdle() // once again, should do nothing + assertEquals(2, i) + assertEquals(4, j) + runCurrent() // should behave the same way as for the normal work + assertEquals(3, i) + assertEquals(5, j) + launch { + delay(1001) + assertEquals(13, i) + assertEquals(25, j) + } + advanceUntilIdle() // should execute the normal work, and with that, the background one, too + } + + @Test + fun testBackgroundWorkErrorReporting() { + var testFinished = false + val exception = RuntimeException("x") + try { + runBlockingTestOnTestScope { + backgroundScope.launch { + throw exception + } + delay(1000) + testFinished = true + } + fail("unreached") + } catch (e: Throwable) { + assertSame(e, exception) + assertTrue(testFinished) + } + } + + @Test + fun testBackgroundWorkFinalizing() { + var taskEnded = 0 + val nTasks = 10 + try { + runBlockingTestOnTestScope { + repeat(nTasks) { + backgroundScope.launch { + try { + while (true) { + delay(1) + } + } finally { + ++taskEnded + if (taskEnded <= 2) + throw TestException() + } + } + } + delay(100) + throw TestException() + } + fail("unreached") + } catch (e: TestException) { + assertEquals(2, e.suppressedExceptions.size) + assertEquals(nTasks, taskEnded) + } + } + + @Test + fun testExampleBackgroundJob1() = runBlockingTestOnTestScope { + val myFlow = flow { + yield() + var i = 0 + while (true) { + emit(++i) + delay(1) + } + } + val stateFlow = myFlow.stateIn(backgroundScope, SharingStarted.Eagerly, 0) + var j = 0 + repeat(100) { + assertEquals(j++, stateFlow.value) + delay(1) + } + } + + @Test + fun testExampleBackgroundJob2() = runBlockingTestOnTestScope { + val channel = Channel() + backgroundScope.launch { + var i = 0 + while (true) { + channel.send(i++) + } + } + repeat(100) { + assertEquals(it, channel.receive()) + } + } + + @Test + fun testAsyncFailureInBackgroundReported() = + try { + runBlockingTestOnTestScope { + backgroundScope.async { + throw TestException("x") + } + backgroundScope.produce { + throw TestException("y") + } + delay(1) + throw TestException("z") + } + fail("unreached") + } catch (e: TestException) { + assertEquals("z", e.message) + assertEquals(setOf("x", "y"), e.suppressedExceptions.map { it.message }.toSet()) + } + + @Test + fun testNoDuplicateExceptions() = + try { + runBlockingTestOnTestScope { + backgroundScope.launch { + throw TestException("x") + } + delay(1) + throw TestException("y") + } + fail("unreached") + } catch (e: TestException) { + assertEquals("y", e.message) + assertEquals(listOf("x"), e.suppressedExceptions.map { it.message }) + } +} diff --git a/kotlinx-coroutines-test/native/src/TestBuilders.kt b/kotlinx-coroutines-test/native/src/TestBuilders.kt index c3176a03de..a959901919 100644 --- a/kotlinx-coroutines-test/native/src/TestBuilders.kt +++ b/kotlinx-coroutines-test/native/src/TestBuilders.kt @@ -8,7 +8,7 @@ import kotlinx.coroutines.* @Suppress("ACTUAL_WITHOUT_EXPECT") public actual typealias TestResult = Unit -internal actual fun createTestResult(testProcedure: suspend () -> Unit) { +internal actual fun createTestResult(testProcedure: suspend CoroutineScope.() -> Unit) { runBlocking { testProcedure() }