Unit Testing WithTimeout in Coroutines With Kotest

Sometimes you may be required to implement code that executes periodic actions. In Java, you can implement such a feature using a Timer. Kotlin also provided a wrapper for this, using the fixedRateTimer function. The above mentioned method of creating a timer is old-style, thread-based and JVM specific. Coroutines offer a better solution for solving this problem, in a more lightweight, efficient and cross-platform manner. One such more-kotlin styled approach, is provided by the ticker, which is part of the Channels API. However, this approach is currently marked as ObsoleteCoroutinesApi, and may be subject to change.

A suspendable, coroutine-based Timer implementation

It is not very difficult to create a custom implementation of the Timer functionality, while also leveraging the power of Kotlin coroutines. This can be based on the withTimeout and the delay functions. The withTimeout function runs a given block until the specified timeout. If the timeout is not exceeded, then it exists normally. However, if the timeout is exceeded, it throws a TimeoutCancellationException. The delay function will suspend the current coroutine until the specified amount of time has passed.

Using the abovementioned suspending functions, a trivial timer implementation can be implemented:

class CoroutineTimer(
    private val period: Long = 1000L,
    private val repeat: Boolean = false,
    private val dispatcher: CoroutineDispatcher = Dispatchers.Default,
    private val timerAction: () -> Unit
) {
    private var job: Job? = null

    /**
     * Starts the execution of the timer
     */
    @Synchronized
    fun start() {
        if (isRunning()) {
            return
        }

        job = CoroutineScope(dispatcher).launch {
            withTimeout(period) {
                delay(period + 1)
            }
        }.also {
            it.invokeOnCompletion { exc ->
                when (exc) {
                    is TimeoutCancellationException -> {
                        // Timeout reached, execute the action and reschedule if required
                        timerAction()
                        if (repeat) {
                            start()
                        }
                    }
                    is CancellationException -> {
                        // The stop method was called and I've been cancelled
                        // Ignore
                    }
                    else -> {
                        // Unknown exception received
                        // Ignore
                    }
                }
            }
        }
    }

    /**
     * Stops the execution of the timer, if it is running
     */
    fun stop() {
        job?.cancel(CancellationException())
    }

    /**
     * Returns whether the timer is currently running or not
     */
    fun isRunning(): Boolean = job?.isActive == true
}

Let’s dive into the code. The CoroutineTimer class, takes 4 arguments in its constructor:

• the period of execution in milliseconds
• whether the timer is repeatable or once-off
• the CoroutineDispatcher to use while waiting (is to better support tests)
• the action to execute, as a non-suspending lambda that takes no arguments and returns no value (obviously this can be modified accordingly)

The timer has two states, started and stopped. It is initiallised in the stopped state and can be controlled with the following methods:

• the start method starts the timer, when it is not running. If the timer is already running, it will not be started again.
• the stop method stops the timer
• the isRunning method returns whether the timer is currently running or not. When the timer is started, it is considered to be running, whereas when it is stopped, it is not considered as running. In case the timer is used in an once-off fashion, after the action has executed, the timer is considered to be stopped (i.e. not running).

The core functionality has been implemented in the start method. Internally, it launches a new coroutine that uses the withTimeout function for the timer. Inside the withTimeout, the code uses the suspending delay function, to delay until the timeout is reached. When the timeout is reached, a TimeoutCancellationException is thrown by the withTimeout function. In case, the timer is cancelled, then a CancellationException will be thrown (by the stop method). The code will inspect the result of the coroutine, with the help of the invokeOnCompletion, and if the timeout was reached, it will execute the action and then proceed to reschedule the next execution of the code (if the timer is repeatable). To determine whether the timer is running and to support the timer cancellation, a reference to the Job object returned by launch function is maintained.

Unit testing the implementation with JUnit - the naive way

Testing multithreading code is always a challenging task. In our case, we have two main test cases:

• the timer action is executed after the timer has elapsed (exactly once when the repeat flag is not set, or multiple times when it is set)
• the timer action is not executed when the timer is cancelled before it is elapsed

The naive approach would be to:

• use a mock timer action for testing if it is executed or not
• use a small timer interval and carefully timed sleeps for waiting until the timer elapses

There are significant shortcomings with blocking the test with the sleeps method until the timer elapses. First of all, the sleeps must be carefully timed, or you may end up having non-deterministic test outcomes, due to the non-deterministic nature of thread scheduling. Blocking the thread that executes your test, increases your overall test execution time (and unit tests must not take too long). In more complex situations, using sleeps will become very difficult to manage.

If we had to write our tests using this inefficient manner, let’s see how this could be done (using JUnit 5 and Mockk). First of all, we want to test that when the repeat flag is false and the timer is started, the action will be executed after the timer has elapsed.

    @Timeout(3500, unit = TimeUnit.MILLISECONDS)
    @Test
    fun without_repeat_calling_start_will_execute_the_timer_action_once() {
        val testCoroutineTimer = CoroutineTimer(2000, false, timerAction = mockTimerAction)

        testCoroutineTimer.start()

        assertEquals(true, testCoroutineTimer.isRunning())
        while (testCoroutineTimer.isRunning()) {
            sleep(1000)
        }
        assertEquals(false, testCoroutineTimer.isRunning())
        verify { mockTimerAction.invoke() }
    }

The above test creates a new non-repeatable CoroutineTimer, with a time period of 2000 msec, which will execute the mockTimerAction. It proceeds to start the timer, verify that it is running, and then sleep until the timer stops running. When this happens, the test is concluded by verifying that the mock timer action was executed. Currently, the test runs successfully. But what would happen if while updating the code, a bug is introduced in the isRunning method and it always reports that the timer is running. In this case, the test would fall into an infinite loop. To cater for this eventuality, the JUnit5 Timeout annotation was added, which instructs the test runner to terminate the test execution after the specified time interval elapses.

To test that if the timer action is not executed when the timer is cancelled, we can use the following test:

    @Timeout(3500, unit = TimeUnit.MILLISECONDS)
    @Test
    fun without_repeat_calling_stop_will_stop_the_timer_without_executing_the_timer_action() {
        val testCoroutineTimer = CoroutineTimer(2000, false, timerAction = mockTimerAction)

        testCoroutineTimer.start()
        sleep(1000)
        assertEquals(true, testCoroutineTimer.isRunning())
        testCoroutineTimer.stop()

        while (testCoroutineTimer.isRunning()) {
            sleep(1000)
        }
        assertEquals(false, testCoroutineTimer.isRunning())
        verify { mockTimerAction wasNot called }
    }

In the above test, the timer is initialised with a large enough period, such that when the stop method is called, the timer will not have expired. The isRunning checks after the test must still be performed, because stop may not be instantaneous. Again, to avoid waiting forever for the timer to stop, the Timeout JUnit 5 annotation is used.

To test the multiple execution of the timer action (i.e. the repeat flag is set to true), the following test can be used:

    @Test
    fun with_repeat_wait_to_execute_twice_and_then_stop() {
        val testCoroutineTimer = CoroutineTimer(2000, true, timerAction = mockTimerAction)

        testCoroutineTimer.start()
        sleep(2100)
        assertEquals(true, testCoroutineTimer.isRunning())
        verify(atLeast = 1, atMost = 1) { mockTimerAction.invoke() }
        sleep(2100)
        testCoroutineTimer.stop()
        verify(atLeast = 2, atMost = 2) { mockTimerAction.invoke() }

        assertEquals(false, testCoroutineTimer.isRunning())
    }

The above test starts the timer, sleeps until the timer expires once, and then sleeps a bit more to verify that the timer expires again. Finally, it stops the timer execution and verifies that the timer has stopped.

Unit testing the implementation with JUnit - the coroutine way

Coroutines are a more advanced, modern and fine grained tool for achieving concurrency. As such, they include much better testing support, which can be found in the kotlinx-coroutines-test artifact. The main testing functionalities provided by the package are the following:

• The TestCoroutineDispatcher which executes tasks synchronously, but also allows the developer to control the dispatcher’s internal clock.
• Controlling the Dispatchers.Main dispatcher, using the Dispatchers.setMain and Dispatchers.resetMain functions.
• The runBlockingTest function, which executes a test by bypassing any delays introduced by async and launch and skipping the calls to delay. In essence, this is similar to the runBlocking coroutine builder function.

Armed with the abovementioned tools, we can rewrite our tests so that:

• they are much more manageable
• they use the provided coroutine tools
• they run faster There are several general guidelines that will assist you in effectively testing coroutine code. First of all, you should always inject the dispatcher in use. You should also provide the desired dispatcher as a default value. Using this simple tip, allows you to override the dispatcher in your tests with the TestCoroutineDispatcher. Also, your test functions should always contain a single runBlockingTest call, with the test code inside. The runBlockingTest can use a predefined coroutine dispatcher, allowing you to easily control its internal clock. Finally, it is essentially a coroutine builder function allowing you to call other suspend functions immediately.

First of all, we declare an instance of the TestCoroutineDispatcher in our test class, which we recreate before each test. This Dispatcher will give us access to the coroutine’s internal clock (e.g. move time forward!). Next, we wrap the body of the three aforementioned tests, inside a runBlockingTest call, such that we eliminate delays from async and launch. We also pass the reference of our TestCoroutineDispatcher to the runBlockingTest method, and also inject this dispatcher to our Timer class, such that they both use the same CoroutineContext (we are thus able to call the advanceTimeBy inside the body of our test, rather than explicitly on the test dispatcher). Finally, we replace the sleep related calls, with calls to advanceTimeBy (and remove any loops we may have been using).

Consequently, the first test becomes:

    @Timeout(3500, unit = TimeUnit.MILLISECONDS)
    @Test
    fun without_repeat_calling_start_will_execute_the_timer_action_once() = runBlockingTest(testCoroutineDispatcher) {
        val testCoroutineTimer = CoroutineTimer(2000, false, testCoroutineDispatcher, mockTimerAction)

        testCoroutineTimer.start()

        assertEquals(true, testCoroutineTimer.isRunning())
        advanceTimeBy(2100)

        assertEquals(false, testCoroutineTimer.isRunning())
        verify { mockTimerAction.invoke() }
    }

The second test becomes:

    @Timeout(3500, unit = TimeUnit.MILLISECONDS)
    @Test
    fun without_repeat_calling_stop_will_stop_the_timer_without_executing_the_timer_action() = runBlockingTest(testCoroutineDispatcher) {
        val testCoroutineTimer = CoroutineTimer(2000, false, testCoroutineDispatcher, mockTimerAction)

        testCoroutineTimer.start()
        advanceTimeBy(1000)
        assertEquals(true, testCoroutineTimer.isRunning())
        testCoroutineTimer.stop()
        advanceTimeBy(1500)

        assertEquals(false, testCoroutineTimer.isRunning())
        verify { mockTimerAction wasNot called }
    }

And finally, the third test becomes:

    @Test
    fun with_repeat_wait_to_execute_twice_and_then_stop() = runBlockingTest(testCoroutineDispatcher) {
        val testCoroutineTimer = CoroutineTimer(2000, true, testCoroutineDispatcher, mockTimerAction)

        testCoroutineTimer.start()
        advanceTimeBy(2100)
        assertEquals(true, testCoroutineTimer.isRunning())
        verify(atLeast = 1, atMost = 1) { mockTimerAction.invoke() }
        advanceTimeBy(2100)
        testCoroutineTimer.stop()
        verify(atLeast = 2, atMost = 2) { mockTimerAction.invoke() }

        assertEquals(false, testCoroutineTimer.isRunning())
    }

Unit testing the implementation with Kotest

Kotest is a more modern testing framework for Kotlin. It makes heavy use of coroutines for its implementation, and therefore includes first-class support for them. For example, you can unit test suspend functions without enclosing them in runBlocking or runBlockingTest calls. In the upcoming 5.0.0 release there will be support for the TestCoroutinDispatcher, which is currently available in the 5.0.0.M3 snapshot release.

The changes required for the conversion of the tests to the Kotest library, can be found here. One of the great advantages of Kotest is that it supports using multiple testing styles. The unit tests of the CoroutineTimer were converted to the BehaviorSpec testing style. The TestCoroutineDispatcher in Kotest is exported through the [DelayController interface]{https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-test/kotlinx.coroutines.test/-delay-controller/index.html}. We can enable it (along with a default timeout of all test cases) in our Kotest BehaviorSpec tests, with the following code:

@ExperimentalStdlibApi
@ExperimentalKotest
@ExperimentalCoroutinesApi
class CoroutineTimerTest : BehaviorSpec({
    // Sets per-test case timeout for all test cases in the spec
    timeout = 3500
    // Enable the TestCoroutineDispatcher
    testCoroutineDispatcher = true

To get access to the coroutine dispatcher in use you can use the following code:

coroutineContext[CoroutineDispatcher.Key]

Kotest uses the TestCoroutineDispatcher in the current coroutineContext, which can be acquired through the CoroutineDispatcher.Key. Thus we can create an initialisation function like the following:

    fun setupTest(repeat: Boolean, coroutineContext: CoroutineContext) {
        mockTimerAction = mockk()
        every { mockTimerAction.invoke() } just Runs

        testCoroutineTimer = CoroutineTimer(
            2000,
            repeat,
            timerAction = mockTimerAction,
            dispatcher = coroutineContext[CoroutineDispatcher.Key]!!
        )
    }

In our tests we need not use the runBlockingTest function call like before. In order to adjust the internal coroutine dispatcher timer, Kotest exposes the delayController property of the TestContext. Apart from the different testing style introduced, very little needs to be changed, in comparison with the JUnit approach and the kotlinx-coroutines-test artifact. Thus the tests can be rewritten:

Given("a CoroutineTimer") {
    And("the timer executes once") {
        And("the timer is running") {
            When("the timer is executed") {
                setupTest(repeat = false, testContext.coroutineContext)
                 testCoroutineTimer.start()
                val timerWasRunning = testCoroutineTimer.isRunning()

                delayController.advanceTimeBy(2100)

                Then("the timer was running") {
                    timerWasRunning shouldBe true
                }
                Then("the timer stops running") {
                    testCoroutineTimer.isRunning() shouldBe false
                }
                Then("the timerAction is executed once") {
                    verify(atMost = 1, atLeast = 1) { mockTimerAction.invoke() }
                }
            }
            When("the timer is stopped") {
                setupTest(repeat = false, testContext.coroutineContext)
                testCoroutineTimer.start()

                delayController.advanceTimeBy(1000)

                val timerWasRunning = testCoroutineTimer.isRunning()

                 testCoroutineTimer.stop()
                delayController.advanceTimeBy(1500)

                Then("verify the time was running") {
                    timerWasRunning shouldBe true
                }
                Then("the timer stops running") {
                    assertEquals(false, testCoroutineTimer.isRunning())
                }
                Then("the timer actions is not executed") {
                    verify { mockTimerAction wasNot called }
                }
            }
        }
    }
    And("the timer executes indefinitely") {
        When("a single time period has elapsed") {
            setupTest(repeat = true, testContext.coroutineContext)
            testCoroutineTimer.start()
            delayController.advanceTimeBy(2100)
            val timerWasRunning = testCoroutineTimer.isRunning()
            delayController.advanceTimeBy(2100)

            testCoroutineTimer.stop()

            Then("the timer was running") {
                timerWasRunning shouldBe true
            }
            Then("the timer action has executed twice") {
                verify(atLeast = 2, atMost = 2) { mockTimerAction.invoke() }
            }
            Then("the timer is not running") {
                testCoroutineTimer.isRunning() shouldBe false
            }
        }
    }}
})

Benchmarking the different test implementations

To benchmark the performance gains of using the kotlinx-coroutine-test means of testing versus the traditional sleep approach versus the Kotest implementation, the tests were executed 10 times with each different approach. The tables below summarises the results, which show that the tests using the coroutine tools were significantly faster.

The tests executed with Kotest were blazingly fast. It is worth pointing out, that on the machine were the tests were run, it took a bit more than 2 seconds to initialise the JUnit coroutine test suite. This time was added in the first test executed. This means, that excluding the test initialisation, the tests using coroutine tools took milliseconds to complete, making them comparable to the Kotest tests. Clearly, using coroutines results in code which is much cleaner, more testable and much faster.