Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/common/test/flow/operators/TimeoutTest.kt (Kotlin)

Code before:
        assertEquals(listOf(1, 2, 3), flow.toList())
        finish(1)
    }

    @Test
    fun testUpstreamError() = testUpstreamError(TestException())

    @Test
    fun testUpstreamErrorTimeoutException() = testUpstreamError(TimeoutCancellationException(0, Job()))

    @Test
    fun testUpstreamErrorCancellationException() = testUpstreamError(CancellationException(""))

    private inline fun <reified T: Throwable> testUpstreamError(cause: T) = runTest {
        try {
            // Workaround for JS legacy bug
            flow {
                emit(1)
                throw cause
            }.timeout(1000.milliseconds).collect()
            expectUnreached()
        } catch (e: Throwable) {
            assertIs<T>(e)
            finish(1)
        }
    }

    @Test
    fun testDownstreamError() = runTest {
        val flow = flow {
            expect(1)
            emit(1)
            hang { expect(3) }
            expectUnreached()
        }.timeout(100.milliseconds).map {
            expect(2)
            yield()
            throw TestException()
        }

        assertFailsWith<TestException>(flow)
        finish(4)
    }

    @Test
    fun testUpstreamTimeoutIsolatedContext() = runTest {
        val flow = flow {
            assertEquals("upstream", NamedDispatchers.name())
            expect(1)
            emit(1)
            expect(2)

Review comment (bug):
You raise a good point about a race not being needed. I think though that we should also check what happens in case of a race between the `flow` block throwing an exception and the `timeout` directive. WDYT?
```suggestion
    }

    @Test
    fun testUpstreamExceptionsTakingPriority() = withVirtualTime {
        val flow = flow<Unit> {
            expect(2)
            withContext(NonCancellable) {
                delay(2.milliseconds)
            }
            assertFalse(currentCoroutineContext().isActive) // cancelled already
            expect(3)
            throw TestException()
        }.timeout(1.milliseconds)
        expect(1)
        assertFailsWith<TestException> {
            flow.collect {
                expectUnreached()
            }
        }
        finish(4)
    }
```

Code after:
        assertEquals(listOf(1, 2, 3), flow.toList())
        finish(1)
    }

    @Test
    fun testUpstreamError() = testUpstreamError(TestException())

    @Test
    fun testUpstreamErrorTimeoutException() = testUpstreamError(TimeoutCancellationException(0, Job()))

    @Test
    fun testUpstreamErrorCancellationException() = testUpstreamError(CancellationException(""))

    private inline fun <reified T: Throwable> testUpstreamError(cause: T) = runTest {
        try {
            // Workaround for JS legacy bug
            flow {
                emit(1)
                throw cause
            }.timeout(1000.milliseconds).collect()
            expectUnreached()
        } catch (e: Throwable) {
            assertTrue { e is T }
            finish(1)
        }
    }

    @Test
    fun testUpstreamExceptionsTakingPriority() = withVirtualTime {
        val flow = flow<Unit> {
            expect(2)
            withContext(NonCancellable) {
                delay(2.milliseconds)
            }
            assertFalse(currentCoroutineContext().isActive) // cancelled already
            expect(3)
            throw TestException()
        }.timeout(1.milliseconds)
        expect(1)
        assertFailsWith<TestException> {
            flow.collect {
                expectUnreached()
            }
        }
        finish(4)
    }

    @Test
    fun testDownstreamError() = runTest {
        val flow = flow {
            expect(1)