Increase code coverage in core module (#3962)

* Increase code coverage in the core module
* Add missing tests for reasonable things (i.e. invariants are tested, but things like utility 'toString' are not)
* Remove some unused code
* Add a few more excludes
* Move Broadcast-related deprecations to 'Deprecated.kt' and promote them to ERROR
* Update Kover and verification rules
* Remove unused dead code

gradle.properties

@@ -23,7 +23,7 @@ rxjava3_version=3.0.2
 javafx_version=17.0.2
 javafx_plugin_version=0.0.8
 binary_compatibility_validator_version=0.13.2
-kover_version=0.7.0-Beta
+kover_version=0.7.4
 blockhound_version=1.0.8.RELEASE
 jna_version=5.9.0
 

kotlinx-coroutines-core/api/kotlinx-coroutines-core.api

@@ -265,7 +265,6 @@ public final class kotlinx/coroutines/CoroutineStart : java/lang/Enum {
 	public static final field LAZY Lkotlinx/coroutines/CoroutineStart;
 	public static final field UNDISPATCHED Lkotlinx/coroutines/CoroutineStart;
 	public static fun getEntries ()Lkotlin/enums/EnumEntries;
-	public final fun invoke (Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/Continuation;)V
 	public final fun invoke (Lkotlin/jvm/functions/Function2;Ljava/lang/Object;Lkotlin/coroutines/Continuation;)V
 	public final fun isLazy ()Z
 	public static fun valueOf (Ljava/lang/String;)Lkotlinx/coroutines/CoroutineStart;

kotlinx-coroutines-core/build.gradle

@@ -351,12 +351,6 @@ static void configureJvmForLincheck(task, additional = false) {
 task moreTest(dependsOn: [jvmStressTest, jvmLincheckTest, jvmLincheckTestAdditional])
 check.dependsOn moreTest
 
-def commonKoverExcludes =
-        ["kotlinx.coroutines.debug.*", // Tested by debug module
-         "kotlinx.coroutines.channels.ChannelsKt__DeprecatedKt.*", // Deprecated
-         "kotlinx.coroutines.scheduling.LimitingDispatcher", // Deprecated
-         "kotlinx.coroutines.scheduling.ExperimentalCoroutineDispatcher" // Deprecated
-        ]
 
 kover {
     excludeTests {
@@ -375,9 +369,12 @@ koverReport {
         excludes {
             classes(
                     "kotlinx.coroutines.debug.*", // Tested by debug module
-                    "kotlinx.coroutines.channels.ChannelsKt__DeprecatedKt.*", // Deprecated
+                    "kotlinx.coroutines.channels.ChannelsKt__DeprecatedKt*", // Deprecated
                     "kotlinx.coroutines.scheduling.LimitingDispatcher", // Deprecated
                     "kotlinx.coroutines.scheduling.ExperimentalCoroutineDispatcher", // Deprecated
+                    "kotlinx.coroutines.flow.FlowKt__MigrationKt*", // Migrations
+                    "kotlinx.coroutines.flow.LintKt*", // Migrations
+                    "kotlinx.coroutines.internal.WeakMapCtorCache", // Fallback implementation that we never test
                     "_COROUTINE._CREATION", // For IDE navigation
                     "_COROUTINE._BOUNDARY", // For IDE navigation
             )

kotlinx-coroutines-core/common/src/CoroutineStart.kt

@@ -75,25 +75,6 @@ public enum class CoroutineStart {
      */
     UNDISPATCHED;
 
-    /**
-     * Starts the corresponding block as a coroutine with this coroutine's start strategy.
-     *
-     * * [DEFAULT] uses [startCoroutineCancellable].
-     * * [ATOMIC] uses [startCoroutine].
-     * * [UNDISPATCHED] uses [startCoroutineUndispatched].
-     * * [LAZY] does nothing.
-     *
-     * @suppress **This an internal API and should not be used from general code.**
-     */
-    @InternalCoroutinesApi
-    public operator fun <T> invoke(block: suspend () -> T, completion: Continuation<T>): Unit =
-        when (this) {
-            DEFAULT -> block.startCoroutineCancellable(completion)
-            ATOMIC -> block.startCoroutine(completion)
-            UNDISPATCHED -> block.startCoroutineUndispatched(completion)
-            LAZY -> Unit // will start lazily
-        }
-
     /**
      * Starts the corresponding block with receiver as a coroutine with this coroutine start strategy.
      *

kotlinx-coroutines-core/common/src/channels/Channels.common.kt

@@ -17,27 +17,6 @@ internal const val DEFAULT_CLOSE_MESSAGE = "Channel was closed"
 
 // -------- Operations on BroadcastChannel --------
 
-/**
- * Opens subscription to this [BroadcastChannel] and makes sure that the given [block] consumes all elements
- * from it by always invoking [cancel][ReceiveChannel.cancel] after the execution of the block.
- *
- * **Note: This API is obsolete since 1.5.0 and deprecated for removal since 1.7.0**
- * It is replaced with [SharedFlow][kotlinx.coroutines.flow.SharedFlow].
- *
- * Safe to remove in 1.9.0 as was inline before.
- */
-@ObsoleteCoroutinesApi
-@Suppress("DEPRECATION")
-@Deprecated(level = DeprecationLevel.WARNING, message = "BroadcastChannel is deprecated in the favour of SharedFlow and is no longer supported")
-public inline fun <E, R> BroadcastChannel<E>.consume(block: ReceiveChannel<E>.() -> R): R {
-    val channel = openSubscription()
-    try {
-        return channel.block()
-    } finally {
-        channel.cancel()
-    }
-}
-
 /**
  * This function is deprecated in the favour of [ReceiveChannel.receiveCatching].
  *
@@ -118,19 +97,6 @@ public suspend fun <E> ReceiveChannel<E>.toList(): List<E> = buildList {
     }
 }
 
-/**
- * Subscribes to this [BroadcastChannel] and performs the specified action for each received element.
- *
- * **Note: This API is obsolete since 1.5.0 and deprecated for removal since 1.7.0**
- */
-@Deprecated(level = DeprecationLevel.WARNING, message = "BroadcastChannel is deprecated in the favour of SharedFlow and is no longer supported")
-@Suppress("DEPRECATION")
-public suspend inline fun <E> BroadcastChannel<E>.consumeEach(action: (E) -> Unit): Unit =
-    consume {
-        for (element in this) action(element)
-    }
-
-
 @PublishedApi
 internal fun ReceiveChannel<*>.cancelConsumed(cause: Throwable?) {
     cancel(cause?.let {

kotlinx-coroutines-core/common/src/channels/Deprecated.kt

@@ -11,6 +11,39 @@ import kotlinx.coroutines.*
 import kotlin.coroutines.*
 import kotlin.jvm.*
 
+/**
+ * Opens subscription to this [BroadcastChannel] and makes sure that the given [block] consumes all elements
+ * from it by always invoking [cancel][ReceiveChannel.cancel] after the execution of the block.
+ *
+ * **Note: This API is obsolete since 1.5.0 and deprecated for removal since 1.7.0**
+ * It is replaced with [SharedFlow][kotlinx.coroutines.flow.SharedFlow].
+ *
+ * Safe to remove in 1.9.0 as was inline before.
+ */
+@ObsoleteCoroutinesApi
+@Suppress("DEPRECATION")
+@Deprecated(level = DeprecationLevel.ERROR, message = "BroadcastChannel is deprecated in the favour of SharedFlow and is no longer supported")
+public inline fun <E, R> BroadcastChannel<E>.consume(block: ReceiveChannel<E>.() -> R): R {
+    val channel = openSubscription()
+    try {
+        return channel.block()
+    } finally {
+        channel.cancel()
+    }
+}
+
+/**
+ * Subscribes to this [BroadcastChannel] and performs the specified action for each received element.
+ *
+ * **Note: This API is obsolete since 1.5.0 and deprecated for removal since 1.7.0**
+ */
+@Deprecated(level = DeprecationLevel.ERROR, message = "BroadcastChannel is deprecated in the favour of SharedFlow and is no longer supported")
+@Suppress("DEPRECATION", "DEPRECATION_ERROR")
+public suspend inline fun <E> BroadcastChannel<E>.consumeEach(action: (E) -> Unit): Unit =
+    consume {
+        for (element in this) action(element)
+    }
+
 /** @suppress **/
 @PublishedApi // Binary compatibility
 internal fun consumesAll(vararg channels: ReceiveChannel<*>): CompletionHandler =

kotlinx-coroutines-core/common/src/flow/operators/Delay.kt

@@ -308,12 +308,9 @@ public fun <T> Flow<T>.sample(periodMillis: Long): Flow<T> {
  */
 internal fun CoroutineScope.fixedPeriodTicker(
     delayMillis: Long,
-    initialDelayMillis: Long = delayMillis
 ): ReceiveChannel<Unit> {
-    require(delayMillis >= 0) { "Expected non-negative delay, but has $delayMillis ms" }
-    require(initialDelayMillis >= 0) { "Expected non-negative initial delay, but has $initialDelayMillis ms" }
     return produce(capacity = 0) {
-        delay(initialDelayMillis)
+        delay(delayMillis)
         while (true) {
             channel.send(Unit)
             delay(delayMillis)

kotlinx-coroutines-core/common/src/flow/operators/Zip.kt

@@ -4,7 +4,7 @@
 
 @file:JvmMultifileClass
 @file:JvmName("FlowKt")
-@file:Suppress("UNCHECKED_CAST", "NON_APPLICABLE_CALL_FOR_BUILDER_INFERENCE") // KT-32203
+@file:Suppress("UNCHECKED_CAST")
 
 package kotlinx.coroutines.flow
 

kotlinx-coroutines-core/common/src/internal/ThreadSafeHeap.kt

@@ -32,11 +32,6 @@ public open class ThreadSafeHeap<T> : SynchronizedObject() where T: ThreadSafeHe
 
     public val isEmpty: Boolean get() = size == 0
 
-    public fun clear(): Unit = synchronized(this) {
-        a?.fill(null)
-        _size.value = 0
-    }
-
     public fun find(
         predicate: (value: T) -> Boolean
     ): T? = synchronized(this) block@{

kotlinx-coroutines-core/common/src/intrinsics/Undispatched.kt

@@ -20,19 +20,6 @@ internal fun <T> (suspend () -> T).startCoroutineUnintercepted(completion: Conti
     }
 }
 
-/**
- * Use this function to start a new coroutine in [CoroutineStart.UNDISPATCHED] mode &mdash;
- * immediately execute the coroutine in the current thread until the next suspension.
- * It does not use [ContinuationInterceptor], but updates the context of the current thread for the new coroutine.
- */
-internal fun <T> (suspend () -> T).startCoroutineUndispatched(completion: Continuation<T>) {
-    startDirect(completion) { actualCompletion ->
-        withCoroutineContext(completion.context, null) {
-            startCoroutineUninterceptedOrReturn(actualCompletion)
-        }
-    }
-}
-
 /**
  * Use this function to start a new coroutine in [CoroutineStart.UNDISPATCHED] mode &mdash;
  * immediately execute the coroutine in the current thread until the next suspension.

kotlinx-coroutines-core/common/test/channels/BroadcastTest.kt

@@ -1,8 +1,8 @@
 /*
- * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ * Copyright 2016-2023 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
  */
 
-@file:Suppress("NAMED_ARGUMENTS_NOT_ALLOWED")
+@file:Suppress("DEPRECATION")
 
 package kotlinx.coroutines.channels
 
@@ -25,13 +25,14 @@ class BroadcastTest : TestBase() {
         }
         yield() // has no effect, because default is lazy
         expect(2)
-        b.consume {
-            expect(3)
-            assertEquals(1, receive()) // suspends
-            expect(7)
-            assertEquals(2, receive()) // suspends
-            expect(8)
-        }
+
+        val subscription = b.openSubscription()
+        expect(3)
+        assertEquals(1, subscription.receive()) // suspends
+        expect(7)
+        assertEquals(2, subscription.receive()) // suspends
+        expect(8)
+        subscription.cancel()
         expect(9)
         yield() // to broadcast
         finish(11)
@@ -138,4 +139,4 @@ class BroadcastTest : TestBase() {
         yield() // to cancel broadcast
         finish(6)
     }
-}
+}

kotlinx-coroutines-core/common/test/flow/BuildersTest.kt

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2016-2023 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.flow
+
+import kotlinx.coroutines.*
+import kotlin.test.*
+
+class BuildersTest : TestBase() {
+
+    @Test
+    fun testSuspendLambdaAsFlow() = runTest {
+        val lambda = suspend { 42 }
+        assertEquals(42, lambda.asFlow().single())
+    }
+
+    @Test
+    fun testRangeAsFlow() = runTest {
+        assertEquals((0..9).toList(), (0..9).asFlow().toList())
+        assertEquals(emptyList(), (0..-1).asFlow().toList())
+
+        assertEquals((0L..9L).toList(), (0L..9L).asFlow().toList())
+        assertEquals(emptyList(), (0L..-1L).asFlow().toList())
+    }
+
+    @Test
+    fun testArrayAsFlow() = runTest {
+        assertEquals((0..9).toList(), IntArray(10) { it }.asFlow().toList())
+        assertEquals(emptyList(), intArrayOf().asFlow().toList())
+
+        assertEquals((0L..9L).toList(), LongArray(10) { it.toLong() }.asFlow().toList())
+        assertEquals(emptyList(), longArrayOf().asFlow().toList())
+    }
+
+    @Test
+    fun testSequence() = runTest {
+        val expected = (0..9).toList()
+        assertEquals(expected, expected.iterator().asFlow().toList())
+        assertEquals(expected, expected.asIterable().asFlow().toList())
+    }
+}

kotlinx-coroutines-core/common/test/flow/operators/BufferTest.kt

@@ -184,5 +184,14 @@ class BufferTest : TestBase() {
             .toList()
         assertEquals(listOf(1, 2), result)
     }
+
+    @Test
+    fun testFailsOnIllegalArguments() {
+        val flow = emptyFlow<Int>()
+        assertFailsWith<IllegalArgumentException> { flow.buffer(capacity = -3) }
+        assertFailsWith<IllegalArgumentException> { flow.buffer(capacity = Int.MIN_VALUE) }
+        assertFailsWith<IllegalArgumentException> { flow.buffer(capacity = Channel.CONFLATED, onBufferOverflow = BufferOverflow.DROP_LATEST) }
+        assertFailsWith<IllegalArgumentException> { flow.buffer(capacity = Channel.CONFLATED, onBufferOverflow = BufferOverflow.DROP_OLDEST) }
+    }
 }
 

kotlinx-coroutines-core/common/test/flow/operators/CombineTest.kt

@@ -261,6 +261,10 @@ class CombineTest : CombineTestBase() {
     override fun <T1, T2, R> Flow<T1>.combineLatest(other: Flow<T2>, transform: suspend (T1, T2) -> R): Flow<R> = combineOriginal(other, transform)
 }
 
+class CombineOverloadTest : CombineTestBase() {
+    override fun <T1, T2, R> Flow<T1>.combineLatest(other: Flow<T2>, transform: suspend (T1, T2) -> R): Flow<R> = combineOriginal(this, other, transform)
+}
+
 class CombineTransformTest : CombineTestBase() {
     override fun <T1, T2, R> Flow<T1>.combineLatest(other: Flow<T2>, transform: suspend (T1, T2) -> R): Flow<R> = combineTransformOriginal(other) { a, b ->
         emit(transform(a, b))

kotlinx-coroutines-core/common/test/flow/operators/DebounceTest.kt

@@ -315,4 +315,10 @@ class DebounceTest : TestBase() {
         assertEquals(listOf("A", "C", "D", "E"), result)
         finish(5)
     }
+
+    @Test
+    fun testFailsWithIllegalArgument() {
+        val flow = emptyFlow<Int>()
+        assertFailsWith<IllegalArgumentException> { flow.debounce(-1) }
+    }
 }

kotlinx-coroutines-core/common/test/flow/operators/SampleTest.kt

@@ -296,4 +296,10 @@ class SampleTest : TestBase() {
         assertEquals(listOf("A", "B", "D"), result)
         finish(5)
     }
+
+    @Test
+    fun testFailsWithIllegalArgument() {
+        val flow = emptyFlow<Int>()
+        assertFailsWith<IllegalArgumentException> { flow.debounce(-1) }
+    }
 }

kotlinx-coroutines-core/common/test/flow/operators/TimeoutTest.kt

@@ -6,8 +6,12 @@ package kotlinx.coroutines.flow.operators
 
 import kotlinx.coroutines.*
 import kotlinx.coroutines.flow.*
+import kotlinx.coroutines.flow.internal.*
+import kotlin.coroutines.*
 import kotlin.test.*
+import kotlin.time.*
 import kotlin.time.Duration.Companion.milliseconds
+import kotlin.time.Duration.Companion.seconds
 
 class TimeoutTest : TestBase() {
     @Test
@@ -228,4 +232,20 @@ class TimeoutTest : TestBase() {
         assertEquals((0 until 10).toList(), list)
         finish(6)
     }
+
+    @Test
+    fun testImmediateTimeout() {
+        testImmediateTimeout(Duration.ZERO)
+        reset()
+        testImmediateTimeout(-1.seconds)
+    }
+
+    private fun testImmediateTimeout(timeout: Duration) {
+        expect(1)
+        val flow = emptyFlow<Int>().timeout(timeout)
+        flow::collect.startCoroutine(NopCollector, Continuation(EmptyCoroutineContext) {
+            assertTrue(it.exceptionOrNull() is TimeoutCancellationException)
+            finish(2)
+        })
+    }
 }

kotlinx-coroutines-core/common/test/sync/MutexTest.kt

@@ -59,6 +59,21 @@ class MutexTest : TestBase() {
         assertFalse(mutex.isLocked)
     }
 
+    @Test
+    fun testWithLockFailureUnlocksTheMutex() = runTest {
+        val mutex = Mutex()
+        assertFalse(mutex.isLocked)
+        try {
+            mutex.withLock {
+                expect(1)
+                throw TestException()
+            }
+        } catch (e: TestException) {
+            finish(2)
+        }
+        assertFalse(mutex.isLocked)
+    }
+
     @Test
     fun testUnconfinedStackOverflow() {
         val waiters = 10000