ContentResolver from ProviderTestRule doesn't notify ContentObserver

Question

I'm trying to test my Flow based ContentResolver extension function:

abstract class Query {
    abstract suspend fun runQuery(): Cursor?
}

private val mainThread = Handler(Looper.getMainLooper())

fun ContentResolver.observeQuery(
    uri: Uri,
    projection: Array<String>? = null,
    selection: String? = null,
    selectionArgs: Array<String>? = null,
    sortOrder: String? = null,
    notifyForDescendants: Boolean = true
): Flow<Query> {
    val query: Query = object : Query() {
        override suspend fun runQuery(): Cursor? {
            return query(uri, projection, selection, selectionArgs, sortOrder)
        }
    }

    return callbackFlow {
        val observer: ContentObserver = object : ContentObserver(mainThread) {
            override fun onChange(selfChange: Boolean) {
                trySend(query)
            }
        }

        registerContentObserver(uri, notifyForDescendants, observer)

        awaitClose {
            unregisterContentObserver(observer)
        }
    }.onStart {
        emit(query)
    }
}

Here's my androidTest code written using Cash App's Turbine Flow testing library.

class KiteContentResolverTest {

    private val dispatcher = TestCoroutineDispatcher()

    @get:Rule
    val providerRule: ProviderTestRule = ProviderTestRule
        .Builder(TestContentProvider::class.java, AUTHORITY.authority!!)
        .build()

    private val contentResolver: ContentResolver
        get() = providerRule.resolver

    @Test
    fun testCreateQueryObservesInsert() = runBlockingTest {
        contentResolver.observeQuery(TABLE).flowOn(dispatcher).test {
            awaitItem().runQuery()!!.isExhausted()
            contentResolver.insert(TABLE, values("key1", "val1"))
            //Code suspends here and never receives another item.
            awaitItem().runQuery()!!
                .hasRow("key1", "val1")
                .isExhausted()
        }
    }

    @Test
    fun testCreateQueryObservesUpdate() = runBlockingTest {
        contentResolver.insert(TABLE, values("key1", "val1"))
        contentResolver.observeQuery(TABLE).flowOn(dispatcher).test {
            awaitItem().runQuery()!!.hasRow("key1", "val1").isExhausted()
            contentResolver.update(TABLE, values("key1", "val2"), null, null)
            //Code suspends here and never receives another item.
            awaitItem().runQuery()!!.hasRow("key1", "val2").isExhausted()
        }
    }

    companion object {

        private val AUTHORITY: Uri = Uri.parse("content://test_authority")
        private val TABLE: Uri = AUTHORITY.buildUpon().appendPath("test_table").build()
        private const val KEY = "test_key"
        private const val VALUE = "test_value"
    }
}

And my mock ContentProvider

class TestContentProvider: ContentProvider() {
    private val storage: MutableMap<String, String> = LinkedHashMap()
    
    private val contentResolver: ContentResolver
        get() = context!!.contentResolver
    
    ...
    override fun insert(uri: Uri, values: ContentValues?): Uri? {
        values ?: return null
        storage[values.getAsString(KEY)] = values.getAsString(VALUE)
        contentResolver.notifyChange(uri, null)
        return Uri.parse(AUTHORITY.toString() + "/" + values.getAsString(KEY))
    }

    override fun update(
        uri: Uri,
        values: ContentValues?,
        selection: String?,
        selectionArgs: Array<String>?
    ): Int {
        values ?: return storage.size
        for (key in storage.keys) {
            storage[key] = values.getAsString(VALUE)
        }
        contentResolver.notifyChange(uri, null)
        return storage.size
    }

    override fun query(
        uri: Uri,
        projection: Array<String>?,
        selection: String?,
        selectionArgs: Array<String>?,
        sortOrder: String?
    ): Cursor {
        val result = MatrixCursor(arrayOf(KEY, VALUE))
        for ((key, value) in storage) {
            result.addRow(arrayOf<Any>(key, value))
        }
        return result
    }
}

However, It seems like the ContentObserver is never actually called when my data is inserted or updated. And my tests always fail with this stack trace:

kotlinx.coroutines.TimeoutCancellationException: Timed out waiting for 1000 ms

Oddly, enough when I run this code on a real device using the actual MediaStore ContentResolver the code executes and the flow emits new values as expected.

It would seem to me that the ContentResolver provided by the ProviderTestRule simply doesn't work with content observers. Has anyone else figured out a way to observer mock ContentResolvers in this way?