Kotlinx.coroutines: EspressoIdlingResource integration

[ZakTaccardi]

Would be nice to have a wrapper to easily provide support for IdlingResource.

Some examples

• RxJava: RxIdler
• OkHttpIdlingResource

EDIT: I'm currently injecting my CoroutineDispatchers like so:

open class AppDispatchers(
    val ui: CoroutineDispatcher = UI,
    val background: CoroutineDispatcher = backgroundDefault,
    val network: CoroutineDispatcher = networkDefault
)

For espresso testing, which monitors the async task pool and UI thread for idle conditions, I'm injecting the following:

class EspressoDispatchers : AppDispatchers(
    ui = UI,
    background = AsyncTask.THREAD_POOL_EXECUTOR.asCoroutineDispatcher(),
    network = AsyncTask.THREAD_POOL_EXECUTOR.asCoroutineDispatcher()
)

Here's the problem I'm experiencing, which happens about 1% of the time on our automated tests.

1. Network call completes.
   
   
   
   • monitored by okhttp-idling-resource
   
   
2. a repository level ConflatedBroadcastChannel is updated with the information from the network call.
   
   
   
   • done by the background dispatcher
   
   
3. Espresso thinks app is now idle, and throws exception because the app isn't idle yet (see number 4)
4. a ConflatedBroadcastChannel in the ViewModel (which has been observing the repository level ConflatedBroadcastChannel the whole time) is updated
   
   
   
   • done by the background dispatcher
   
   
   • this happens after Espresso thinks the app is idle (espresso is wrong)
   
   

[jcornaz]

Something like the folowing ?

suspend fun IdlingResource.awaitIdle() {
  if (isIdleNow) return

  suspendCoroutine<Unit> { cont ->
    registerIdleTransitionCallback { cont.resume(Unit) }
  }
}

[jcornaz]

Or maybe you meant a way to create an IdlingResource from a Job ?

fun Job.asIdlingResource() = object : IdlingResource {
  override fun getName() = "Coroutine job ${this@asIdlingResource}"

  override fun isIdleNow() = isCompleted

  override fun registerIdleTransitionCallback(callback: IdlingResource.ResourceCallback) {
    invokeOnCompletion { callback.onTransitionToIdle() }
  }
}

[ZakTaccardi]

one problem is that the application can be idle when a coroutine is still running (you're observing a receive channel)

[ZakTaccardi]

updated original description with details.

I'd like the IdlingResource to be done at the CoroutineDispatcher level, similar to how RxIdler works, if possible. Also note: I'll have .consumeEach { } calls on ReceiveChannels, so a coroutine can still be suspended when the app is actually idle.

[matejdro]

This is what I came up with:

class JobCheckingDispatcherWrapper(private val parent: CoroutineDispatcher) :
    CoroutineDispatcher() {
    private val jobs = Collections.newSetFromMap(WeakHashMap<Job, Boolean>())

    var completionEvent: (() -> Unit)? = null

    override fun dispatch(context: CoroutineContext, block: Runnable) {
        context[Job]?.let { addNewJob(it) }
        parent.dispatch(context, block)
    }

    @InternalCoroutinesApi
    override fun dispatchYield(context: CoroutineContext, block: Runnable) {
        context[Job]?.let { addNewJob(it) }
        parent.dispatchYield(context, block)
    }

    private fun addNewJob(job: Job): Boolean {
        job.invokeOnCompletion {
            completionEvent?.invoke()
        }
        return jobs.add(job)
    }

    @ExperimentalCoroutinesApi
    override fun isDispatchNeeded(context: CoroutineContext): Boolean {
        context[Job]?.let { addNewJob(it) }
        return parent.isDispatchNeeded(context)
    }

    val isAnyJobRunning: Boolean
        get() {
            jobs.removeAll { !it.isActive }
            return jobs.isNotEmpty()
        }
}

It seem to work okay for me. It wraps around existing dispatcher and steals its job objects to check whether those are running.

I cannot share IdlingResource implementation, since it is a bit specific to my setup, but you basically check if any injected dispatcher has isAnyJobRunning == true. You can also register completionEvent callback and forward it to IdlingResource.ResourceCallback if there is no other job running.

What this class does not handle though is your ReceiveChannel example. I'm not sure if this is even possible to do generally though (you may want to wait on some receive channels, but not on the others). Maybe inject separate dispatcher for the receive channel and separate one for jobs that need to finish?

[yigit]

I would actually like to have a solution that can work similar to how we can swap Dispatchers.Main (so ability to replace IO and Default as well).

In my toy app, I wrote a dispatcher like this and replace default and IO dispatchers with it for each test (via reflection :-1: ).

I would very much prefer a solution that does work with Dispatchers.IO, Default and Main out of the box so that developers do not need to pass down dispatchers around. It might be still preferred, just shouldn't be mandatory just to make things testable.

below is the tracked dispatcher:

@InternalCoroutinesApi
class TrackedDispatcher(
    private val name : String,
    private val onSubmit: () -> Unit,
    private val onFinish: () -> Unit,
    private val scheduled : ScheduledExecutorService = ScheduledThreadPoolExecutor(10,
        object : ThreadFactory {
            private val idCounter = AtomicInteger(0)
            override fun newThread(runnable: java.lang.Runnable?): Thread {
                return Thread(runnable, "[Tracked]$name-${idCounter.incrementAndGet()}")
            }

        })
) : CoroutineDispatcher(), Delay {
    override fun scheduleResumeAfterDelay(
        timeMillis: Long,
        continuation: CancellableContinuation<Unit>
    ) {
        onSubmit()
        scheduled.schedule(Runnable {
            try {
                continuation.resumeWith(Result.success(Unit))
            } finally {
                onFinish()
            }

        }, timeMillis, TimeUnit.MILLISECONDS)
    }
    @InternalCoroutinesApi
    override fun dispatch(context: CoroutineContext, block: Runnable) {
        onSubmit()
        scheduled.execute {
            try {
                block.run()
            } finally {
                onFinish()
            }
        }
    }

    fun shutdown() {
        scheduled.shutdown()
        scheduled.awaitTermination(10, TimeUnit.SECONDS)
    }
}

[elizarov]

This is related to the discussion in #890 about testing coroutine framework. It seem that an ability to replace built-in dispatchers Dispatchers.Default and Dispatchers.IO for test purposes just like replacing of Dispatchers.Main would greatly help with Espresso integration as you'd be able to to set a dispatcher that registers itself as IdlingResource. Here is a separate issue: #982

Consider the scenario by @ZakTaccardi:

When a repository level ConflatedBroadcastChannel is updated with the information from the network call it would resume a coroutine in the ViewModel (which has been observing the repository level ConflatedBroadcastChannel the whole time). If that coroutine's dispatcher is integrated with Espresso, it receives the the new task (to resume observing coroutine) and is not idle anymore, so Espresso knows that application is not idle yet.

[yigit]

I was checking how this can be implemented but not sure which direction to take.
An option might be to do the same MainDispatchers loader mechanism similar to Dispatchers.setMain.
Another option might be to open up CommonPool and DefaultScheduler to the test module and use it like TestBase.

Also, seems like there are 2 use cases here:
1) A basic Executor tracking where a Dispatcher is idle only if # of enqueued runnables is 0.
2) A Dispatcher that awaits for all jobs that are ever scheduled, similar to this one.

I think both are necessary. 2 is useful when interacting with external systems where nothing might be running in the app but it might still be awaiting for a system callback to continue.
But it has the disadvantage in working with delayed tasks (or never ending jobs). Hence, option 1 might be necessary in some cases.

[elizarov]

It seems to me that if we add ability to setDefault, then one can simply replace it with #890 virtual-time enabled TestDispatcher. Would it help or not?

[matejdro]

We would also need setIO in this case, right?

[yigit]

https://github.com/Kotlin/kotlinx.coroutines/issues/242#issuecomment-485339796 works fine. Should we implement the same ServiceLocator discovery mechanism for the Default dispatcher? (similar to the one in Main?) Or do you have a different implementation in mind?

[elizarov]

The roadmap I have in mind:

1. #261 -- Add extended APIs to DefaultDispatcher to create its subviews like "IO"
2. #982 -- Add Dispatchers.setDefault (no setIO needed because of the above)
3. Do everything else.

[matejdro]

Wouldn't that be hardcoding implementation details? IO dispatchers is now subview of default dispatchers, but it might not be in the future.

[elizarov]

@matejdro That's not just an implementation that. That is the whole idea to have Dispatchers.IO in the core library, because it is a subview of Dispatchars.Default and you don't need to switch threads to go between the two (see #261 for details)

[objcode]

@elizarov I'm starting to take a look at this a bit more and there's a common test case to consider as a use case:

suspend fun foo() {
    withContext(Dispatchers.IO) { }
}

suspend fun bar() {
    withContext(Dispatchers.Default) { }
}

// tests
@Test
fun foo_runsOnIO() = runBlockingTest {
    foo()
    // here I want to assert that a task was dispatched to IO
}

@Test
fun bar_runsOnDefault() = runBlockingTest {
    bar()
    // here I want to assert bar runs on Default
}

I'm currently thinking the isIdle method @yigit mentioned in the code review for #890 is the right solution for those assertions - but it'd be good to avoid introducing a separate API than what's needed for Espresso if possible.

[elizarov]

Can you please clarify the code in the above comment. When you say "here I want to assert that a task was dispatched to IO" do you really mean that you want to assert that the task was already complete in IO dispatcher (and it is now idle) or what? What is the expected behavior of test coroutine dispatcher when the task goes to "outside" dispatcher? Shall it wait for all other dispatchers to become idle or... ?

[objcode]

Ah yes I see the confusion, let me clarify a bit with the rest of the test.

suspend fun foo() {
    withContext(Dispatchers.IO) { }
}

@Test
fun foo_runsOnIO() = runBlockingTest {
    // setup a testing IO dispatcher
    val testDispatcher = TestCoroutineDispatcher()
    testDispatcher.pauseDispatcher()
    Dispatchers.setIO(TestCoroutineDispatcher)

    foo()

    // assertion that the IO dispatcher was dispatched to
    assertThat(testDispatcher).hasPendingCoroutine()

    // cleanup stuff
    testDispatcher.resumeDispatcher()
    testDispatcher.cleanupTestCoroutines()
}

The actual API surface for how to write that assertion could take a few forms. Imo idle status is probably the easiest one to understand.

 // calls testDispatcher.isIdle to get the status of dispatcher
assertThat(testDispatcher).hasPendingCoroutine() 

As an alternative, the arguments passed to withContext could be intercepted directly by the test (in an argument captor style). That would be a great solution that doesn't involve diving into dispatcher implementation to test this - however it is not obvious how one would intercept that since withContext is a top level function.

As an alternative, all of this can be solved by wrapping either withContext or Dispatchers.Default behind an appropriate abstraction. However, it would be better if the APIs exposed a testable surface.

The important part, to me, is that I need to have some way of determining that IO was dispatched to instead of Default in a test to ensure the correctness of the code.

[objcode]

Re: Espresso dispatchers (the original issue). After thinking it over I think @yigit has the right direction in https://github.com/Kotlin/kotlinx.coroutines/issues/242#issuecomment-485316699

In most cases, an Espresso test does not need (or want) time control and should simply wrap the existing Dispatchers with instrumentation. Both of the wrappers Yigit spelled out make sense in different cases, with the default option of "idle when nothing is currently running or queued." A delegate pattern would be a great to add this tracking:

val trackedDispatcher = IdlingResourceDispatcher(Dispatchers.Default)
Dispatchers.setDefault(trackedDispatcher)
idlingRegistry.register(trackedDispatcher)

The second option, "idle when all jobs that have ever passed through the dispatcher are complete" is a more complicated (and surprising) IdlingResource to work with but interesting only for one-shot requests. It could use the same delegate pattern:

val trackedDispatcher = OneShotIdlingResourceDispatcher(Dispatchers.Default)
Dispatchers.setDefault(trackedDispatcher)
idlingRegistry.register(trackedDispatcher)

In both cases, I don't think the correct choice would be to use TestCoroutineDispatcher here since a test of e.g. a button click should not be testing the implementation details of other layers. If a UI test did need to control the return order of multiple coroutines, it could do so without TestCoroutineDispatcher by supplying fakes and mocks.

Q: Are there any use cases that would require a TestCoroutineDispatcher to supply an espresso IdlingResource?

[JoseAlcerreca]

Q: Are there any use cases that would require a TestCoroutineDispatcher to supply an espresso IdlingResource?

I agree with you: _If a UI test did need to control the return order of multiple coroutines, it could do so without TestCoroutineDispatcher by supplying fakes and mocks._

But, would it be hard to allow TestCoroutineDispatcher to be used in a IdlingResourceDispatcher or OneShotIdlingResourceDispatcher in case someone needs to control dispatcher timing inside a UI test? Actually, can you even prevent it?

What if an app uses Dispatchers.Default for both one-shot operations and channels? You'd have to use both types of IdlingResourceDispatchers and I'm not sure they can be used at the same time. In that case you might want to inject different test dispatchers (even if you use the same one, Default, in production).

[objcode]

Yea, once you have anything other than a one shot request you'd have to use a IdlingResourceDispatcher. I don't see a way to consider a suspended coroutine busy in the presence of a potentially infinite loop.

From a larger perspective - the underlying resource that's causing the suspend (e.g. Retrofit, Room etc) should also expose an idling resource in this case to tell espresso work is happening, or the code should be updated to use a counting idling resource.

If the underlying resource exposed an idling resource, the flow would be complicated but create the desired effect. Consider a streaming database read.

(all idle) -> (coroutine idle, database busy) -> (database busy, coroutine busy) -> (database idle, coroutine busy) -> result sent to UI which blocks main -> (all idle)

Q: Integrating with TestCoroutineDispatcher

As for integrating this with TestCoroutineDispatcher, right now there's strict type checking in TestCoroutineScope that would make both of these delegates not work with runBlockingTest. It would work with setMain.

The two options I see there:

Relax the type checking to allow any DelayController that's also a dispatcher to be passed.

This has a disadvantage of requiring separate idling resource implementations for TestCoroutineDispatcher and regular dispatchers, but it does allow the same pattern to be used for both.

Provide an idle mechanism (https://github.com/Kotlin/kotlinx.coroutines/issues/1202) that allows for the creation of IdlingResources from a TestCoroutineDispatcher.

This creates a separate API, but maybe that's OK since they're quite different - however it may be surprising that runBlockingTest fails when a TestCoroutineDispatcher is wrapped in IdlingResourceDispatcher

cc @JoseAlcerreca ^

[LUwaisA]

We have been using a delegate pattern as mentioned in objcode's comment. It is a similar idea to the code written by @yigit but allows tests to behave more similarly to production code by wrapping production dispatchers.

class EspressoTrackedDispatcher(private val wrappedCoroutineDispatcher: CoroutineDispatcher) : CoroutineDispatcher() {
    private val counter: CountingIdlingResource = CountingIdlingResource("EspressoTrackedDispatcher for $wrappedCoroutineDispatcher")
    init {
        IdlingRegistry.getInstance().register(counter)
    }

    override fun dispatch(context: CoroutineContext, block: Runnable) {
        counter.increment()
        val blockWithDecrement = Runnable {
            try {
                block.run()
            } finally {
                counter.decrement()
            }
        }
        wrappedCoroutineDispatcher.dispatch(context, blockWithDecrement)
    }

    fun cleanUp() {
        IdlingRegistry.getInstance().unregister(counter)
    }
}

I'm then using the above Dispatcher in a TestRule:

class DispatcherIdlerRule: TestRule {
    override fun apply(base: Statement?, description: Description?): Statement =
        object : Statement() {
            override fun evaluate() {
                val espressoTrackedDispatcherIO = EspressoTrackedDispatcher(Dispatchers.IO)
                val espressoTrackedDispatcherDefault = EspressoTrackedDispatcher(Dispatchers.Default)
                MyDispatchers.IO = espressoTrackedDispatcherIO
                MyDispatchers.Default = espressoTrackedDispatcherDefault
                try {
                    base?.evaluate()
                } finally {
                    espressoTrackedDispatcherIO.cleanUp()
                    espressoTrackedDispatcherDefault.cleanUp()
                    MyDispatchers.resetAll()
                }
            }
        }
}

In the absence of setIO()/setDefault() methods the MyDispatchers class is something we've added to prod code to allow the extra flexibility of setting the dispatchers. It's fairly simple and mimics (and adds to) the public API of Dispatchers. I would rather not have this class but thought it is a small addition and is easily replaced when better options become available:

object MyDispatchers {
    var Main: CoroutineDispatcher = Dispatchers.Main
    var IO: CoroutineDispatcher = Dispatchers.IO
    var Default: CoroutineDispatcher = Dispatchers.Default

    fun resetMain() {
        Main = Dispatchers.Main
    }

    fun resetIO() {
        IO = Dispatchers.IO
    }

    fun resetDefault() {
        Default = Dispatchers.Default
    }

    fun resetAll() {
        resetMain()
        resetIO()
        resetDefault()
    }
}

With this approach, we've unblocked our Espresso testing. Caveat: For our use case we don't currently need it to behave more effectively but the above code tells Espresso that the app is idle during a delay() in production code.

Hope this helps for anyone else struggling with this issue

[ZakTaccardi]

I have found that simply monitoring a CoroutineDispatcher is not sufficient.

Imagine the following scenario :

2 coroutines pass data between each other using .offer(..) (not sure if this matters) via a Channel. Is it not possible for both coroutines to be suspended at the same time and therefore a monitored CoroutineDispatcher would still report that the app is idle, even if there is an item in the Channel waiting to be sent to the other coroutine?

EDIT: this seems to be the scenario that @objcode raised an issue for https://github.com/Kotlin/kotlinx.coroutines/issues/1202#issue-445209002

[matejdro]

You can use CoroutineDispatcher as a hook to get all the jobs though, and then listen to all jobs manually: https://github.com/Kotlin/kotlinx.coroutines/issues/242#issuecomment-458046292

[VysotskiVadim]

Thank you guys, I'm glad that so many people shared their implementation of dispatcher wrappers. You're amaizing :+1:

I used to work with Rx using Rx Idler which is basically scheduler wrapper, example:

override fun onCreate(savedInstanceState: Bundle?) {
    super.onCreate(savedInstanceState)
    setContentView(R.layout.activity_main)
    disposable = Single.create<Int> {
        Thread.sleep(3000)
        it.onSuccess(1)
    }
    .observeOn(AndroidSchedulers.mainThread())
    .subscribeOn(Schedulers.computation())
    .subscribe { next ->
        textView.text = next.toString()
    }
}

@Test
fun testExample() {
    onView(withId(R.id.textView)).check(matches(withText("1")))
}

Example works fine because result of computation schedules to UI thread and then computation scheduler marked as Idle. Almost always (I run this test for a half an hour in cycle with 100% success execution, but I think it should fails sometime) UI thread have enough time to update view with result of computation before test checks text view.

I tried to do something similar for coroutines using dispatcher wrapper provided by @matejdro

private val _state = MutableLiveData<ScreenState>()
val state by lazy<LiveData<ScreenState>> {
        _state.value = ScreenState.Loading
        viewModelScope.launch {
            _state.value = ScreenState.Loaded(useCase.getFeatureData())
        }
        _state
    }

class FeatureUseCaseImpl @Inject constructor(
    @IODispatcher private val ioDispatcher: CoroutineDispatcher
): FeatureUseCase {
    override suspend fun getFeatureData(): FeatureData = withContext(ioDispatcher) {
        delay(1500)
        FeatureData("boo")
    }
}

@Test
fun testDataUpdated() {
    onView(withId(R.id.textView)).check(matches(withText("boo")))
}

Full example

When you run the test you can see how UI is updated with the text 'boo', but test fails. The trick is coroutines have different order of execution rather then in Rx. With coroutines dispatcher wrapper is notified that task is completed and only then result is dispatched to Main dispatcher to update UI. As a result, test checks UI few milliseconds before it's updated.

I tried to fix it by modifying @matejdro `s dispatcher wrapper:

private const val ONE_FRAME = 17L
private fun addNewJob(job: Job): Boolean {
        job.invokeOnCompletion {
            if (isAnyJobRunning.not()) {
                GlobalScope.launch(Dispatchers.Main) {
                    delay(ONE_FRAME * 2)
                    completionEvent?.invoke()
                }
            }
        }
        return jobs.add(job)
    }

This workaround works, but not very stable, if you run test in cycle for about half an hour it would fails at least once.

@matejdro , does your dispatcher work stable at your project? Can you suggest something to improve stability?
@elizarov , is the any way to notify dispatcher wrapper after computation result was dispatched to parent coroutine?

[fkirc]

What is the current status about this?

I believe that something like @LUwaisA 's solution is the right way to go.
In particular, I would like to have the following properties:

• Ability to instrument Dispatchers.IO/Default via a delegate pattern.
• Setup this instrumentation from the tests, such that the production code is not polluted.
• Out of scope for the core library: By using such an instrumentation, make Coroutines and Espresso work out of the box (similar to AsyncTask, which has been working with Espresso since years).

An example of how this could be implemented is Dispatchers.setMain() in org.jetbrains.kotlinx:kotlinx-coroutines-test.
As outlined by @elizarov, a Dispatchers.setDefault() method might do the job just fine.

In the meantime, a workaround is to replace Dispatchers.IO/Default with custom wrapper classes.