From bd36c5e5d60fb3ff160d3dbc8ccaa935cab695ad Mon Sep 17 00:00:00 2001 From: Matthias Urhahn Date: Sat, 11 Jul 2026 12:15:17 +0200 Subject: [PATCH] fix(upgrade): Recover billing from stale purchase data and mid-flow errors --- .../common/upgrade/core/UpgradeRepoGplay.kt | 97 ++++++++--- .../core/client/BillingClientConnection.kt | 16 ++ .../client/BillingClientConnectionProvider.kt | 46 ++++- .../upgrade/core/data/BillingDataRepo.kt | 15 ++ .../upgrade/core/UpgradeRepoGplayTest.kt | 158 ++++++++++++++++++ 5 files changed, 308 insertions(+), 24 deletions(-) diff --git a/app/src/gplay/java/eu/darken/capod/common/upgrade/core/UpgradeRepoGplay.kt b/app/src/gplay/java/eu/darken/capod/common/upgrade/core/UpgradeRepoGplay.kt index bb05b135..17114da1 100644 --- a/app/src/gplay/java/eu/darken/capod/common/upgrade/core/UpgradeRepoGplay.kt +++ b/app/src/gplay/java/eu/darken/capod/common/upgrade/core/UpgradeRepoGplay.kt @@ -1,6 +1,7 @@ package eu.darken.capod.common.upgrade.core import android.app.Activity +import com.android.billingclient.api.BillingClient import eu.darken.capod.common.coroutine.AppScope import eu.darken.capod.common.debug.logging.Logging.Priority.INFO import eu.darken.capod.common.debug.logging.Logging.Priority.VERBOSE @@ -8,6 +9,7 @@ import eu.darken.capod.common.debug.logging.Logging.Priority.WARN import eu.darken.capod.common.debug.logging.asLog import eu.darken.capod.common.debug.logging.log import eu.darken.capod.common.debug.logging.logTag +import eu.darken.capod.common.flow.setupCommonEventHandlers import eu.darken.capod.common.upgrade.UpgradeRepo import eu.darken.capod.common.upgrade.core.client.ItemAlreadyOwnedBillingException import eu.darken.capod.common.upgrade.core.data.BillingData @@ -20,7 +22,11 @@ import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.SharingStarted import kotlinx.coroutines.flow.catch +import kotlinx.coroutines.flow.distinctUntilChanged +import kotlinx.coroutines.flow.filter +import kotlinx.coroutines.flow.launchIn import kotlinx.coroutines.flow.map +import kotlinx.coroutines.flow.onEach import kotlinx.coroutines.flow.onStart import kotlinx.coroutines.flow.shareIn import kotlinx.coroutines.withTimeoutOrNull @@ -47,6 +53,44 @@ class UpgradeRepoGplay @Inject constructor( private val anchorLock = Any() + init { + // Fresh-provenance grace stamping: freshBillingData carries every successful query result + // and push payload as an event — unlike the equality-deduped billingData state, an + // unchanged steady-owner query still stamps, and stale listener data can't sneak in. + // The reactive upgradeInfo mapping deliberately writes nothing anymore. + billingDataRepo.freshBillingData + .onEach { data -> + try { + recordProState(data) + } catch (e: CancellationException) { + throw e + } catch (e: Exception) { + // A failed DataStore write must not kill this process-lifetime collector. + log(TAG, WARN) { "Failed to record pro state: ${e.asLog()}" } + } + } + .setupCommonEventHandlers(TAG) { "proStateRecorder" } + .launchIn(scope) + + // Async variant of the launch-result ITEM_ALREADY_OWNED case: Play told us mid-flow that + // the user already owns it. Reconcile silently — Play shows its own UI for purchase-sheet + // failures, so no app-side dialog here. + billingDataRepo.purchaseFailures + .filter { it.responseCode == BillingClient.BillingResponseCode.ITEM_ALREADY_OWNED } + .onEach { + log(TAG, INFO) { "Async already-owned event -> restoring purchase" } + try { + withTimeoutOrNull(RESTORE_ON_OWNED_TIMEOUT_MS) { restorePurchaseNow() } + } catch (e: CancellationException) { + throw e + } catch (e: Exception) { + log(TAG, WARN) { "Async already-owned restore failed: ${e.asLog()}" } + } + } + .setupCommonEventHandlers(TAG) { "asyncAlreadyOwned" } + .launchIn(scope) + } + // Grace window depends on what was last owned: a permanent one-time purchase should almost // never be dropped on a Play hiccup, so it gets a long window; a subscription legitimately // lapses, so it keeps the short one (also used for unknown/legacy last SKUs). @@ -72,7 +116,9 @@ class UpgradeRepoGplay @Inject constructor( // True once we've ever confirmed a known Pro purchase on this install; drives the proactive // restore banner. Local signal only — a fresh install or switched Google account starts false. - val wasEverPro: Flow = billingCache.lastProStateAt.flow.map { it > 0 } + val wasEverPro: Flow = billingCache.lastProStateAt.flow + .map { it > 0 } + .distinctUntilChanged() // Explicit "Restore purchase": query Play now and evaluate Pro from the returned data in the // same coroutine (real happens-before), so we never read a stale upgradeInfo replay. Billing @@ -80,7 +126,11 @@ class UpgradeRepoGplay @Inject constructor( suspend fun restorePurchaseNow(): Info { log(TAG) { "restorePurchaseNow()" } return try { - billingDataRepo.refresh().toUpgradeInfo() + val data = billingDataRepo.refresh() + // Returned data is fresh by definition — stamp it even if the flows dedupe the + // unchanged result and the init collector never sees a new emission. + recordProState(data) + data.toUpgradeInfo() } catch (e: CancellationException) { throw e } catch (e: Exception) { @@ -97,30 +147,15 @@ class UpgradeRepoGplay @Inject constructor( } // Shared Pro/grace mapping used by both the reactive upgradeInfo flow and restorePurchaseNow(). - // Only relinquishes Pro if we haven't had it for a while (grace period). + // Only relinquishes Pro if we haven't had it for a while (grace period). READ-ONLY: this also + // runs on replayed shared-flow data, so it must never stamp the grace cache — a refunded + // purchase could otherwise keep re-stamping its own grace window. See recordProState(). private fun BillingData?.toUpgradeInfo(): Info { val now = System.currentTimeMillis() val proSku = this?.getProSku() log(TAG) { "toUpgradeInfo(): now=$now, lastProStateAt=$lastProStateAt, data=$this" } return when { - proSku != null -> { - val upgrades = this!!.getProSkus() - // Record which Pro SKU anchors the grace window; the permanent IAP wins when both - // are owned. An IAP anchor is sticky: purchase data may lack the IAP because that - // query failed on a fresh connection, and a subscription seen in the meantime must - // not shrink the 30d window of an owner whose IAP was never disproven. Trade-off: - // a refunded IAP keeps the long window — consistent with the fail-open cache. - // SKU first, timestamp last — the timestamp is the gate, so a crash between the - // two writes stays conservative. Locked: mappings run concurrently from the hot - // reactive flow and direct restores, and the sticky check-then-write must not race. - synchronized(anchorLock) { - preferredProSku(upgrades) - ?.takeIf { it.type == Sku.Type.IAP || !anchorIsIap() } - ?.let { lastProStateSku = it.id } - lastProStateAt = now - } - Info(billingData = this, upgrades = upgrades) - } + proSku != null -> Info(billingData = this, upgrades = this!!.getProSkus()) (now - lastProStateAt) < graceWindowMs() -> { log(TAG, VERBOSE) { "We are not pro, but were recently, did GPlay try annoy us again?" } @@ -131,6 +166,26 @@ class UpgradeRepoGplay @Inject constructor( } } + // Persists "we saw a known Pro purchase" for the grace machinery. Callers must only pass FRESH + // data (returned query results, or new emissions seen by the init collector) — never replayed + // flow data. The permanent IAP wins as anchor when both are owned, and an IAP anchor is sticky: + // purchase data may lack the IAP because that query failed on a fresh connection, and a + // subscription seen in the meantime must not shrink the 30d window of an owner whose IAP was + // never disproven (trade-off: a refunded IAP keeps the long window — consistent with the + // fail-open cache). SKU before timestamp — the timestamp is the gate, so a crash between the + // two writes stays conservative. Locked: runs concurrently from the init collector and direct + // restores, and the sticky check-then-write must not race. + private fun recordProState(data: BillingData) { + val upgrades = data.getProSkus() + val preferred = preferredProSku(upgrades) ?: return + synchronized(anchorLock) { + preferred + .takeIf { it.type == Sku.Type.IAP || !anchorIsIap() } + ?.let { lastProStateSku = it.id } + lastProStateAt = System.currentTimeMillis() + } + } + data class Info( private val gracePeriod: Boolean = false, private val billingData: BillingData?, diff --git a/app/src/gplay/java/eu/darken/capod/common/upgrade/core/client/BillingClientConnection.kt b/app/src/gplay/java/eu/darken/capod/common/upgrade/core/client/BillingClientConnection.kt index fe23f49b..78e97815 100644 --- a/app/src/gplay/java/eu/darken/capod/common/upgrade/core/client/BillingClientConnection.kt +++ b/app/src/gplay/java/eu/darken/capod/common/upgrade/core/client/BillingClientConnection.kt @@ -26,6 +26,7 @@ import kotlinx.coroutines.async import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.withContext import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.combine import kotlinx.coroutines.flow.update @@ -35,7 +36,18 @@ import kotlinx.coroutines.sync.withLock data class BillingClientConnection( private val client: BillingClient, private val purchasesGlobal: Flow>, + private val freshObservations: MutableSharedFlow>, + private val purchaseFailuresGlobal: Flow, ) { + + // Non-OK results from onPurchasesUpdated (e.g. async ITEM_ALREADY_OWNED after the Play sheet + // opened). Consumed by a single persistent collector in UpgradeRepoGplay — not an event bus. + val purchaseFailures: Flow = purchaseFailuresGlobal + + // Every conclusive fresh look at PURCHASED purchases (successful queries and push payloads), + // regardless of whether it differs from the previous one — the combined `purchases` state is + // equality-deduped and can mix in stale listener data, so grace stamping must not use it. + val freshPurchases: Flow> = freshObservations private data class QueryCaches( val iaps: Collection? = null, val subs: Collection? = null, @@ -96,6 +108,10 @@ data class BillingClientConnection( ) } + // A conclusive refresh is a fresh observation for the grace stamping, even when the result + // equals the previous one and the state flows dedupe it away. + freshObservations.tryEmit(combined) + combined } diff --git a/app/src/gplay/java/eu/darken/capod/common/upgrade/core/client/BillingClientConnectionProvider.kt b/app/src/gplay/java/eu/darken/capod/common/upgrade/core/client/BillingClientConnectionProvider.kt index 61f1e19c..615a2d37 100644 --- a/app/src/gplay/java/eu/darken/capod/common/upgrade/core/client/BillingClientConnectionProvider.kt +++ b/app/src/gplay/java/eu/darken/capod/common/upgrade/core/client/BillingClientConnectionProvider.kt @@ -14,14 +14,17 @@ import eu.darken.capod.common.debug.logging.log import eu.darken.capod.common.debug.logging.logTag import eu.darken.capod.common.flow.setupCommonEventHandlers import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.channels.BufferOverflow import kotlinx.coroutines.channels.awaitClose import kotlinx.coroutines.channels.trySendBlocking import kotlinx.coroutines.delay import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.callbackFlow import kotlinx.coroutines.flow.retryWhen import kotlinx.coroutines.launch +import kotlinx.coroutines.withTimeoutOrNull import javax.inject.Inject import javax.inject.Singleton @@ -32,6 +35,23 @@ class BillingClientConnectionProvider @Inject constructor( private val connectionProvider: Flow = callbackFlow { val purchasePublisher = MutableStateFlow>(emptySet()) + // Events, not state: fresh observations feed the grace stamping (every successful query or + // push payload counts, even if equal to the previous one — Purchase.equals would dedupe a + // StateFlow), and failures must not be conflated away (Play reuses BillingResult instances, + // so a repeated ITEM_ALREADY_OWNED could be a same-instance emission). + // replay=1 on observations: the connect-time query can complete before the grace recorder + // subscribes (construction order race) — the latest fresh observation must not be lost. + // Failures stay replay=0: they can only originate from a purchase flow, which requires the + // consumer to already exist, and a consumed event must not be re-delivered. + val freshPurchaseObservations = MutableSharedFlow>( + replay = 1, + extraBufferCapacity = 16, + onBufferOverflow = BufferOverflow.DROP_OLDEST, + ) + val purchaseFailureEvents = MutableSharedFlow( + extraBufferCapacity = 8, + onBufferOverflow = BufferOverflow.DROP_OLDEST, + ) val client = newBuilder(context).apply { enablePendingPurchases( @@ -46,10 +66,16 @@ class BillingClientConnectionProvider @Inject constructor( "onPurchasesUpdated(code=${result.responseCode}, message=${result.debugMessage}, purchases=$purchases)" } purchasePublisher.value = purchases.orEmpty() + freshPurchaseObservations.tryEmit( + purchases.orEmpty().filter { it.purchaseState == Purchase.PurchaseState.PURCHASED } + ) } else { log(TAG, WARN) { "error: onPurchasesUpdated(code=${result.responseCode}, message=${result.debugMessage}, purchases=$purchases)" } + // Failures are published too: async ITEM_ALREADY_OWNED (Play telling us mid-flow + // that the user already owns it) drives the auto-restore in UpgradeRepoGplay. + purchaseFailureEvents.tryEmit(result) } } }.build() @@ -64,14 +90,27 @@ class BillingClientConnectionProvider @Inject constructor( when (result.responseCode) { BillingResponseCode.OK -> { - val connection = BillingClientConnection(client, purchasePublisher) + val connection = BillingClientConnection( + client = client, + purchasesGlobal = purchasePublisher, + freshObservations = freshPurchaseObservations, + purchaseFailuresGlobal = purchaseFailureEvents, + ) trySendBlocking(connection) launch { try { - connection.refreshPurchases() - log(TAG) { "Initial purchase query successful." } + // Bounded: a hung Play callback would otherwise hold the refresh + // lock indefinitely and starve every later refresh on this + // connection (foreground, manual restore, already-owned recovery). + val initial = withTimeoutOrNull(INITIAL_QUERY_TIMEOUT_MS) { + connection.refreshPurchases() + } + if (initial != null) log(TAG) { "Initial purchase query successful." } + else log(TAG, WARN) { "Initial purchase query timed out." } + } catch (e: CancellationException) { + throw e } catch (e: Exception) { log(TAG, ERROR) { "Initial purchase query failed:\n${e.asLog()}" } } @@ -127,6 +166,7 @@ class BillingClientConnectionProvider @Inject constructor( } companion object { + private const val INITIAL_QUERY_TIMEOUT_MS = 30_000L val TAG: String = logTag("Upgrade", "Gplay", "Billing", "Client", "ConnectionProvider") } } \ No newline at end of file diff --git a/app/src/gplay/java/eu/darken/capod/common/upgrade/core/data/BillingDataRepo.kt b/app/src/gplay/java/eu/darken/capod/common/upgrade/core/data/BillingDataRepo.kt index 38f37704..27e5d663 100644 --- a/app/src/gplay/java/eu/darken/capod/common/upgrade/core/data/BillingDataRepo.kt +++ b/app/src/gplay/java/eu/darken/capod/common/upgrade/core/data/BillingDataRepo.kt @@ -2,6 +2,7 @@ package eu.darken.capod.common.upgrade.core.data import android.app.Activity import com.android.billingclient.api.BillingClient +import com.android.billingclient.api.BillingResult import com.android.billingclient.api.Purchase import eu.darken.capod.common.AppForegroundState import eu.darken.capod.common.TimeSource @@ -49,6 +50,20 @@ class BillingDataRepo @Inject constructor( .setupCommonEventHandlers(TAG) { "billingData" } .replayingShare(scope) + // Async purchase failures from onPurchasesUpdated; UpgradeRepoGplay reconciles + // ITEM_ALREADY_OWNED silently. + val purchaseFailures: Flow = connectionProvider + .flatMapLatest { it.purchaseFailures } + .setupCommonEventHandlers(TAG) { "purchaseFailures" } + + // Every fresh observation of PURCHASED purchases (successful queries and push payloads) — + // unlike billingData this is not equality-deduped state and never mixes in stale listener + // data, so it is the only valid source for grace stamping. + val freshBillingData: Flow = connectionProvider + .flatMapLatest { it.freshPurchases } + .map { BillingData(purchases = it) } + .setupCommonEventHandlers(TAG) { "freshBillingData" } + init { connectionProvider .flatMapLatest { client -> diff --git a/app/src/testGplay/java/eu/darken/capod/common/upgrade/core/UpgradeRepoGplayTest.kt b/app/src/testGplay/java/eu/darken/capod/common/upgrade/core/UpgradeRepoGplayTest.kt index e98d0b3f..1381374f 100644 --- a/app/src/testGplay/java/eu/darken/capod/common/upgrade/core/UpgradeRepoGplayTest.kt +++ b/app/src/testGplay/java/eu/darken/capod/common/upgrade/core/UpgradeRepoGplayTest.kt @@ -14,12 +14,18 @@ import io.kotest.matchers.longs.shouldBeGreaterThan import io.kotest.matchers.nulls.shouldBeNull import io.kotest.matchers.nulls.shouldNotBeNull import io.kotest.matchers.shouldBe +import com.android.billingclient.api.BillingClient.BillingResponseCode +import com.android.billingclient.api.BillingResult +import eu.darken.capod.common.datastore.DataStoreValue import io.mockk.coEvery +import io.mockk.coVerify import io.mockk.every import io.mockk.mockk import kotlinx.coroutines.cancel import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.emptyFlow import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.flowOf import kotlinx.coroutines.flow.toList import kotlinx.coroutines.launch import kotlinx.coroutines.test.TestScope @@ -38,6 +44,7 @@ class UpgradeRepoGplayTest : BaseTest() { lateinit var tempDir: File private lateinit var billingDataFlow: MutableSharedFlow + private lateinit var freshDataFlow: MutableSharedFlow private lateinit var billingDataRepo: BillingDataRepo private lateinit var billingCache: BillingCache @@ -46,8 +53,11 @@ class UpgradeRepoGplayTest : BaseTest() { @BeforeEach fun setup() { billingDataFlow = MutableSharedFlow() + freshDataFlow = MutableSharedFlow() billingDataRepo = mockk { every { billingData } returns billingDataFlow + every { freshBillingData } returns freshDataFlow + every { purchaseFailures } returns emptyFlow() } val dataStore = PreferenceDataStoreFactory.create( produceFile = { File(tempDir, "test_billing_cache_${dsCounter++}.preferences_pb") } @@ -430,4 +440,152 @@ class UpgradeRepoGplayTest : BaseTest() { testScope.cancel() } + + private fun mockBillingResult(code: Int): BillingResult = mockk { + every { responseCode } returns code + every { debugMessage } returns "mock" + } + + @Test + fun `fresh observations stamp the grace cache via the persistent collector`() = runTest2 { + val testScope = TestScope(UnconfinedTestDispatcher(testScheduler)) + createRepo(testScope) + + // No upgradeInfo collection at all — the init collector alone must stamp. + freshDataFlow.emit(BillingData(purchases = listOf(mockPurchase(CapodSku.Iap.PRO_UPGRADE.id)))) + + billingCache.lastProStateAt.valueBlocking shouldBeGreaterThan 0L + billingCache.lastProStateSku.valueBlocking shouldBe CapodSku.Iap.PRO_UPGRADE.id + + testScope.cancel() + } + + @Test + fun `the reactive mapping is read-only, only the collector stamps`() = runTest2 { + val testScope = TestScope(UnconfinedTestDispatcher(testScheduler)) + // Count writes via mocked DataStore values: the cold freshBillingData flow delivers one + // pro observation to the init collector (1 stamp), and collecting upgradeInfo runs the + // mapping on top (onStart-null + pro data) — if the mapping still stamped, the count + // would exceed 1. + val proData = BillingData(purchases = listOf(mockPurchase(CapodSku.Iap.PRO_UPGRADE.id))) + val lastProAtMock = mockk>(relaxed = true) { + every { flow } returns flowOf(0L) + } + val lastProSkuMock = mockk>(relaxed = true) { + every { flow } returns flowOf("") + } + val cache = mockk { + every { lastProStateAt } returns lastProAtMock + every { lastProStateSku } returns lastProSkuMock + } + val repo = UpgradeRepoGplay( + scope = testScope, + billingDataRepo = mockk { + every { billingData } returns flowOf(proData) + every { freshBillingData } returns flowOf(proData) + every { purchaseFailures } returns emptyFlow() + }, + billingCache = cache, + ) + + repo.upgradeInfo.first { it.isPro }.isPro shouldBe true + + coVerify(exactly = 1) { lastProAtMock.update(any()) } + + testScope.cancel() + } + + @Test + fun `an observation emitted before the recorder subscribes still stamps via replay`() = runTest2 { + val testScope = TestScope(UnconfinedTestDispatcher(testScheduler)) + // The connect-time query can complete before UpgradeRepoGplay is constructed — the + // observation stream carries replay=1 so that first Pro observation isn't lost. + val replayingFresh = MutableSharedFlow(replay = 1) + replayingFresh.tryEmit(BillingData(purchases = listOf(mockPurchase(CapodSku.Iap.PRO_UPGRADE.id)))) + every { billingDataRepo.freshBillingData } returns replayingFresh + + createRepo(testScope) + + billingCache.lastProStateAt.valueBlocking shouldBeGreaterThan 0L + + testScope.cancel() + } + + @Test + fun `equal consecutive fresh observations both stamp`() = runTest2 { + val testScope = TestScope(UnconfinedTestDispatcher(testScheduler)) + // Purchase equality dedupes the state flows for a steady owner — the observation stream + // must not dedupe, or a long-lived process would stop refreshing the grace timestamp. + val proData = BillingData(purchases = listOf(mockPurchase(CapodSku.Iap.PRO_UPGRADE.id))) + val lastProAtMock = mockk>(relaxed = true) { + every { flow } returns flowOf(0L) + } + val lastProSkuMock = mockk>(relaxed = true) { + every { flow } returns flowOf("") + } + val cache = mockk { + every { lastProStateAt } returns lastProAtMock + every { lastProStateSku } returns lastProSkuMock + } + UpgradeRepoGplay( + scope = testScope, + billingDataRepo = mockk { + every { billingData } returns emptyFlow() + every { freshBillingData } returns flowOf(proData, proData) + every { purchaseFailures } returns emptyFlow() + }, + billingCache = cache, + ) + + coVerify(exactly = 2) { lastProAtMock.update(any()) } + + testScope.cancel() + } + + @Test + fun `the same failure instance delivered twice triggers two restores`() = runTest2 { + val testScope = TestScope(UnconfinedTestDispatcher(testScheduler)) + coEvery { billingDataRepo.refresh() } returns BillingData( + purchases = listOf(mockPurchase(CapodSku.Iap.PRO_UPGRADE.id)) + ) + // Play reuses static BillingResult instances — a repeat of the same object must still + // trigger (a conflating/deduping state flow would drop it). + val sameInstance = mockBillingResult(BillingResponseCode.ITEM_ALREADY_OWNED) + every { billingDataRepo.purchaseFailures } returns flowOf(sameInstance, sameInstance) + + createRepo(testScope) + + coVerify(exactly = 2) { billingDataRepo.refresh() } + + testScope.cancel() + } + + @Test + fun `async already-owned purchase event triggers a silent restore`() = runTest2 { + val testScope = TestScope(UnconfinedTestDispatcher(testScheduler)) + coEvery { billingDataRepo.refresh() } returns BillingData( + purchases = listOf(mockPurchase(CapodSku.Iap.PRO_UPGRADE.id)) + ) + every { billingDataRepo.purchaseFailures } returns + flowOf(mockBillingResult(BillingResponseCode.ITEM_ALREADY_OWNED)) + + createRepo(testScope) + + coVerify(exactly = 1) { billingDataRepo.refresh() } + + testScope.cancel() + } + + @Test + fun `other async purchase failures do not trigger a restore`() = runTest2 { + val testScope = TestScope(UnconfinedTestDispatcher(testScheduler)) + every { billingDataRepo.purchaseFailures } returns + flowOf(mockBillingResult(BillingResponseCode.DEVELOPER_ERROR)) + + createRepo(testScope) + + coVerify(exactly = 0) { billingDataRepo.refresh() } + + testScope.cancel() + } }