SharedFlow/StateFlow是一个热流，如文档中所述，A shared flow is called hot because its active instance exists independently of the presence of collectors.这意味着启动流集合的作用域不会自行完成。
要解决这个问题，你需要取消调用collect的作用域，因为测试的作用域是测试本身，所以取消测试是不对的，所以你需要在不同的作业中启动它。

@Test
fun `Testing a integer state flow`() = runTest {
    val _intSharedFlow = MutableStateFlow(0)
    val intSharedFlow = _intSharedFlow.asStateFlow()
    val testResults = mutableListOf<Int>()

    val job = launch {
        intSharedFlow.toList(testResults)
    }
    _intSharedFlow.value = 5

    assertEquals(2, testResults.size)
    assertEquals(0, testResults.first())
    assertEquals(5, testResults.last())
    job.cancel()
}

您的具体使用案例：

@Test
fun `observe user wallets ok`() = runTest {
    whenever(api.getAssetWallets()).thenReturn(TestUtils.getAssetsWalletResponseOk())
    whenever(api.getFiatWallets()).thenReturn(TestUtils.getFiatWalletResponseOk())

    viewModel.getUserWallets()

    val result = arrayListOf<State<UserWallets>>()
    val job = launch {
        viewModel.userWallet.toList(result) //now it should work
    }

    Assertions.assertThat(viewModel.userWallet.value is State.Success).isTrue() //works, last value enmited
    Assertions.assertThat(result.first() is State.Success) //also works
    job.cancel()
}

两件重要的事：
1.始终取消已创建的作业以避免java.lang.IllegalStateException: This job has not completed yet
1.由于这是一个StateFlow，当开始收集（在toList内部）时，您将收到最后一个状态。但是如果你第一次开始收集数据，然后调用函数viewModel.getUserWallets()，那么在result列表中，你将拥有所有的状态，以防你也想测试它。

我从Kotlin协同程序GitHub存储库中的解决方案中得出的另一种方法：

@Test fun `The StateFlow should emit all expected values`() = runTest {
    val dispatcher = UnconfinedTestDispatcher(testScheduler)
    val viewModel = MyViewModel(dispatcher)
    val results = mutableListOf<Int>()
    val job = launch(dispatcher) { viewModel.numbers.toList(results) }

    viewModel.addNumber(5)
    viewModel.addNumber(8)
    runCurrent() // Important

    assertThat(results).isEqualTo(listOf(0, 5, 8))
    job.cancel() // Important
}

这是我的 ViewModel 类：

class MyViewModel(private val dispatcher: CoroutineDispatcher) : ViewModel() {

    private val _numbers = MutableStateFlow(0)
    val numbers: StateFlow<Int> = _numbers

    fun addNumber(number: Int) {
        viewModelScope.launch(dispatcher) {
            _numbers.value = number
        }
    }
}

请注意，我使用的是Kotlin 1.6.10 和kotlinx.coroutines-test 1.6.1：

testImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:1.6.1")

另外，请参阅官方的Kotlin协程migration guide to the new test API。

runBlockingTest只是跳过用例中的延迟，但不会用测试调度器覆盖ViewModel中使用的调度器。你需要将TestCoroutineDispatcher注入到你的ViewModel中，或者因为你使用的是viewModelScope.launch {}，而默认情况下它已经使用了Dispatchers.Main，你需要通过Dispatchers.setMain(testCoroutineDispatcher)覆盖主调度器。您可以创建以下规则并将其添加到测试文件中。

class MainCoroutineRule(
        val testDispatcher: TestCoroutineDispatcher = TestCoroutineDispatcher()
) : TestWatcher() {

    override fun starting(description: Description?) {
        super.starting(description)
        Dispatchers.setMain(testDispatcher)
    }

    override fun finished(description: Description?) {
        super.finished(description)
        Dispatchers.resetMain()
        testDispatcher.cleanupTestCoroutines()
    }
}

在你的测试文件里

@get:Rule
var mainCoroutineRule = MainCoroutineRule()

@Test
fun `observe user wallets ok`() = mainCoroutineRule.testDispatcher.runBlockingTest {
}

顺便说一句，注入调度程序总是一个很好的做法。例如，如果您在协同程序作用域（如viewModelScope.launch(Dispatchers.Default)）中使用了Dispatchers.Main以外的调度程序，那么即使您使用了测试调度程序，测试也会再次失败。原因是你只能用Dispatchers.setMain()覆盖主调度器，因为它可以从它的名字中理解，而不是Dispatchers.IO或Dispatchers.Default。在这种情况下，您需要将mainCoroutineRule.testDispatcher注入到视图模型中，并使用注入的分派器，而不是硬编码它。

您面临的问题是因为toList（）需要流来完成，而根据文档，“状态流永远不会完成”。
https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/-state-flow/
https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/to-list.html

我们可以为given创建一个协程，为whenever创建一个协程
在whenever代码之后，我们可以使用yield，这样我们给定的代码就完成了，可以Assert了！

要做到这一点，你需要扩展CouroutinScope，如你所见：

搞定！

• 可以使用emit而不是tryEmit

这就是我正在使用的（不需要自定义VM调度器）：

...

@get:Rule
val coroutineRule = MainCoroutineRule()
...

@Test
fun `blablabla`() = runTest {
    val event = mutableListOf<SealedCustomEvent>()
    viewModel.screenEvent
        .onEach { event.add(it) }
        .launchIn(CoroutineScope(UnconfinedTestDispatcher(testScheduler)))
    
    viewModel.onCtaClick()
    advanceUntilIdle()

    Assertions.assertThat(event.last()).isInstanceOf(SealedCustomEvent.OnCtaClick::class.java)

    ...more checks
}

使用launchIn和advanceUntilIdle可以解决您的测试问题。

通过一些小的改进来使用它https://github.com/Kotlin/kotlinx.coroutines/issues/3143#issuecomment-1097428912

import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.Job
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.launch
import kotlinx.coroutines.test.TestCoroutineScheduler
import kotlinx.coroutines.test.TestScope
import kotlinx.coroutines.test.UnconfinedTestDispatcher
import org.junit.Assert.assertEquals

@OptIn(ExperimentalCoroutinesApi::class)
/**
 * Test observer for Flow to be able to capture and verify all states.
 */
class TestObserver<T>(
  scope: CoroutineScope,
  testScheduler: TestCoroutineScheduler,
  flow: Flow<T>
) {
  private val values = mutableListOf<T>()

  private val job: Job = scope.launch(UnconfinedTestDispatcher(testScheduler)) {
    flow.collect { values.add(it) }
  }

  /**
   * Assert no values
   */
  fun assertNoValues(): TestObserver<T> {
    assertEquals(emptyList<T>(), this.values)
    return this
  }

  /**
   * Assert the values. Important [TestObserver.finish] needs to be called at the end of the test.
   */
  fun assertValues(vararg values: T): TestObserver<T> {
    assertEquals(values.toList(), this.values)
    return this
  }

  /**
   * Assert the values and finish. Convenient to avoid having to call finish if done last in the test.
   */
  fun assertValuesAndFinish(vararg values: T): TestObserver<T> {
    assertEquals(values.toList(), this.values)
    finish()
    return this
  }

  /**
   * Finish the job
   */
  fun finish() {
    job.cancel()
  }
}

@OptIn(ExperimentalCoroutinesApi::class)
/**
 * Test function for the [TestObserver]
 */
fun <T> Flow<T>.test(
  scope: TestScope
): TestObserver<T> {
  return TestObserver(scope, scope.testScheduler, this)
}

我现在可以在测试中执行以下操作

@Test
fun `test some states`() = runTest {
  val viewModel = ViewModel(
    repository = repository
  )
  val observer = viewModel.state.test(this)
  advanceUntilIdle()
  verify(repository).getData()
  observer.assertValuesAndFinish(
    defaultState,
    defaultState.copy(isLoading = true),
    defaultState.copy(title = "Some title")
  )
}

我的ViewModel

@HiltViewModel
internal class ViewModel @Inject constructor(
  private val repository: Repository
) : ViewModel() {

  private val _state = MutableStateFlow(State())
  val state: StateFlow<State> = _state

  init {
    fetch()
  }

  private fun fetch() {
    _state.value = state.value.copy(
      isLoading = true
    )
    val someData = repository.getData()
    _state.value = state.value.copy(
      isLoading = false,
      title = someData.title
    )
  }
}