/
StateInTest.kt
89 lines (80 loc) · 3.32 KB
/
StateInTest.kt
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
/*
* Copyright 2016-2020 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 kotlinx.coroutines.channels.*
import kotlin.test.*
/**
* It is mostly covered by [ShareInTest], this just add state-specific checks.
*/
class StateInTest : TestBase() {
@Test
fun testOperatorFusion() = runTest {
val state = flowOf("OK").stateIn(this)
assertTrue(state !is MutableStateFlow<*>) // cannot be cast to mutable state flow
assertSame(state, (state as Flow<*>).cancellable())
assertSame(state, (state as Flow<*>).distinctUntilChanged())
assertSame(state, (state as Flow<*>).flowOn(Dispatchers.Default))
assertSame(state, (state as Flow<*>).conflate())
assertSame(state, state.buffer(Channel.CONFLATED))
assertSame(state, state.buffer(Channel.RENDEZVOUS))
assertSame(state, state.buffer(onBufferOverflow = BufferOverflow.DROP_OLDEST))
assertSame(state, state.buffer(0, onBufferOverflow = BufferOverflow.DROP_OLDEST))
assertSame(state, state.buffer(1, onBufferOverflow = BufferOverflow.DROP_OLDEST))
coroutineContext.cancelChildren()
}
@Test
fun testUpstreamCompletedNoInitialValue() =
testUpstreamCompletedOrFailedReset(failed = false, withInitialValue = false)
@Test
fun testUpstreamFailedNoInitialValue() =
testUpstreamCompletedOrFailedReset(failed = true, withInitialValue = false)
@Test
fun testUpstreamCompletedWithInitialValue() =
testUpstreamCompletedOrFailedReset(failed = false, withInitialValue = true)
@Test
fun testUpstreamFailedWithInitialValue() =
testUpstreamCompletedOrFailedReset(failed = true, withInitialValue = true)
private fun testUpstreamCompletedOrFailedReset(failed: Boolean, withInitialValue: Boolean) = runTest {
val emitted = Job()
val terminate = Job()
val sharingJob = CompletableDeferred<Unit>()
val upstream = flow {
emit("OK")
emitted.complete()
terminate.join()
if (failed) throw TestException()
}
val scope = this + sharingJob
val shared: StateFlow<String?>
if (withInitialValue) {
shared = upstream.stateIn(scope, SharingStarted.Eagerly, null)
assertEquals(null, shared.value)
} else {
shared = upstream.stateIn(scope)
assertEquals("OK", shared.value) // waited until upstream emitted
}
emitted.join() // should start sharing, emit & cache
assertEquals("OK", shared.value)
terminate.complete()
sharingJob.complete(Unit)
sharingJob.join() // should complete sharing
assertEquals("OK", shared.value) // value is still there
if (failed) {
assertTrue(sharingJob.getCompletionExceptionOrNull() is TestException)
} else {
assertNull(sharingJob.getCompletionExceptionOrNull())
}
}
@Test
fun testUpstreamFailedIMmediatelyWithInitialValue() = runTest {
val ceh = CoroutineExceptionHandler { _, _ -> expect(2) }
val flow = flow<Int> {
expect(1)
throw TestException()
}
assertFailsWith<TestException> { flow.stateIn(CoroutineScope(ceh)) }
finish(3)
}
}