Unit testing coroutines runBlockingTest: This job has not completed yet

Question

Please find below a function using a coroutine to replace callback :

override suspend fun signUp(authentication: Authentication): AuthenticationError {
    return suspendCancellableCoroutine {
        auth.createUserWithEmailAndPassword(authentication.email, authentication.password)
            .addOnCompleteListener(activityLifeCycleService.getActivity()) { task ->
                if (task.isSuccessful) {
                    it.resume(AuthenticationError.SignUpSuccess)
                } else {
                    Log.w(this.javaClass.name, "createUserWithEmail:failure", task.exception)
                    it.resume(AuthenticationError.SignUpFail)
                }
            }
    }
}

Now I would like to unit testing this function. I am using Mockk :

  @Test
  fun `signup() must be delegated to createUserWithEmailAndPassword()`() = runBlockingTest {

      val listener = slot<OnCompleteListener<AuthResult>>()
      val authentication = mockk<Authentication> {
        every { email } returns "email"
        every { password } returns "pswd"
      }
      val task = mockk<Task<AuthResult>> {
        every { isSuccessful } returns true
      }

      every { auth.createUserWithEmailAndPassword("email", "pswd") } returns
          mockk {
            every { addOnCompleteListener(activity, capture(listener)) } returns mockk()
          }

    service.signUp(authentication)

      listener.captured.onComplete(task)
    }

Unfortunately this test failed due to the following exception : java.lang.IllegalStateException: This job has not completed yet

I tried to replace runBlockingTest with runBlocking but the test seems to wait in an infinite loop.

Can someone help me with this UT please?

Thanks in advance

Answer 1

As can be seen in this post:

This exception usually means that some coroutines from your tests were scheduled outside the test scope (more specifically the test dispatcher).

Instead of performing this:

private val networkContext: CoroutineContext = TestCoroutineDispatcher()

private val sut = Foo(
  networkContext,
  someInteractor
)

fun `some test`() = runBlockingTest() {
  // given
  ...

  // when
  sut.foo()

  // then
  ...
}

Create a test scope passing test dispatcher:

private val testDispatcher = TestCoroutineDispatcher()
private val testScope = TestCoroutineScope(testDispatcher)
private val networkContext: CoroutineContext = testDispatcher

private val sut = Foo(
  networkContext,
  someInteractor
)

Then in test perform testScope.runBlockingTest

fun `some test`() = testScope.runBlockingTest {
  ...
}

See also Craig Russell's "Unit Testing Coroutine Suspend Functions using TestCoroutineDispatcher"

Answer 2

In case of Flow testing:

• Don't use flow.collect directly inside runBlockingTest. It should be wrapped in launch
• Don't forget to cancel TestCoroutineScope in the end of a test. It will stop a Flow collecting.

Example:

class CoroutinesPlayground {

    private val job = Job()
    private val testDispatcher = StandardTestDispatcher()
    private val testScope = TestScope(job + testDispatcher)

    @Test
    fun `play with coroutines here`() = testScope.runBlockingTest {

        val flow = MutableSharedFlow<Int>()

        launch {
            flow.collect { value ->
                println("Value: $value")
            }
        }

        launch {
            repeat(10) { value ->
                flow.emit(value)
                delay(1000)
            }
            job.cancel()
        }
    }
}

Answer 3

This is not an official solution, so use it at your own risk.

This is similar to what @azizbekian posted, but instead of calling runBlocking, you call launch. As this is using TestCoroutineDispatcher, any tasks scheduled to be run without delay are immediately executed. This might not be suitable if you have several tasks running asynchronously.

It might not be suitable for every case but I hope that it helps for simple cases.

You can also follow up on this issue here:

• https://github.com/Kotlin/kotlinx.coroutines/issues/1204

If you know how to solve this using the already existing runBlockingTest and runBlocking, please be so kind and share with the community.

class MyTest {
    private val dispatcher = TestCoroutineDispatcher()
    private val testScope = TestCoroutineScope(dispatcher)

    @Test
    fun myTest {
       val apiService = mockk<ApiService>()
       val repository = MyRepository(apiService)
       
       testScope.launch {
            repository.someSuspendedFunction()
       }
       
       verify { apiService.expectedFunctionToBeCalled() }
    }
}

Answer 4

None of these answers quite worked for my setup due to frequent changes in the coroutines API.

This specifically works using version 1.6.0 of kotlin-coroutines-test, added as a testImplementation dependency.

    @Test
    fun `test my function causes flow emission`() = runTest {
        // calling this function will result in my flow emitting a value
        viewModel.myPublicFunction("1234")
        
        val job = launch {
            // Force my flow to update via collect invocation
            viewModel.myMemberFlow.collect()
        }
        // immediately cancel job
        job.cancel()

        assertEquals("1234", viewModel.myMemberFlow.value)
    }

Answer 5

According to my understanding, this exception occurs when you are using a different dispatcher in your code inside the runBlockingTest { } block with the one that started runBlockingTest { }.

So in order to avoid this, you first have to make sure you inject Dispatchers in your code, instead of hardcoding it throughout your app. If you haven't done it, there's nowhere to begin because you cannot assign a test dispatcher to your test codes.

Then, in your BaseUnitTest, you should have something like this:

@get:Rule
val coroutineRule = CoroutineTestRule()

@ExperimentalCoroutinesApi
class CoroutineTestRule(
    val testDispatcher: TestCoroutineDispatcher = TestCoroutineDispatcher()
) : TestWatcher() {

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

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

Next step really depends on how you do Depedency Injection. The main point is to make sure your test codes are using coroutineRule.testDispatcher after the injection.

Finally, call runBlockingTest { } from this testDispatcher:

@Test
fun `This should pass`() = coroutineRule.testDispatcher.runBlockingTest {
    //Your test code where dispatcher is injected
}

Answer 6

There is an open issue for this problem: https://github.com/Kotlin/kotlinx.coroutines/issues/1204

The solution is to use the CoroutineScope intead of the TestCoroutinScope until the issue is resolved, you can do by replacing

@Test
fun `signup() must be delegated to createUserWithEmailAndPassword()`() = 
runBlockingTest {

with

@Test
fun `signup() must be delegated to createUserWithEmailAndPassword()`() = 
runBlocking {

Answer 7

If you have any

Channel

inside the launch, you must call to

Channel.close()

Example code:

val channel = Channel<Success<Any>>()
val flow = channel.consumeAsFlow()
    
launch {
      channel.send(Success(Any()))
      channel.close()
}

Answer 8

runBlockingTest deprecated since 1.6.0 and replaced with runTest.

Answer 9

You need to swap arch background executor with one that execute tasks synchronously. eg. For room suspend functions, live data etc.

You need the following dependency for core testing

androidTestImplementation 'androidx.arch.core:core-testing:2.1.0'

Then add the following at the top of test class

@get:Rule
val instantExecutor = InstantTaskExecutorRule()

Explanations

InstantTaskExecutorRule A JUnit Test Rule that swaps the background executor used by the Architecture Components with a different one which executes each task synchronously. You can use this rule for your host side tests that use Architecture Components

Answer 10

As I mentioned here about fixing runBlockingTest, maybe it could help you too.

Add this dependency if you don't have it

testImplementation "androidx.arch.core:core-testing:$versions.testCoreTesting" (2.1.0)

Then in your test class declare InstantTaskExecutorRule rule:

@get:Rule
val instantTaskExecutorRule = InstantTaskExecutorRule()