Updated on 2026-08-14

.claude/rules/unit-testing.md

@@ -0,0 +1,214 @@
+# Unit Testing Rules
+
+This document covers **unit tests** only — sources under `src/test`, running on the JVM via JUnit 5 (Jupiter). UI / instrumentation tests (`src/androidTest`, Kaspresso + Espresso on the JUnit 4 on-device runner) are a separate concern and out of scope here.
+
+## Stack
+
+| Purpose | Library | Version source |
+|---|---|---|
+| Test runner | JUnit 5 (Jupiter) | `deps.test.junit5` |
+| Mocking | MockK | `deps.test.mockk` |
+| Flow testing | Turbine | `deps.test.turbine` |
+| Assertions | Google Truth | `deps.test.truth` |
+| Coroutines | `kotlinx-coroutines-test` | `deps.test.coroutine` |
+
+All unit tests run on JUnit 5. JUnit 4 (`deps.test.junit` = `junit:junit`) is **not** used in `src/test` at all — it survives only in `src/androidTest` instrumentation. Don't add new JUnit 4 unit tests.
+
+Versions live in `gradle/dependencies.toml`. Do not hardcode library coordinates in module build scripts — always go through the catalog.
+
+## Shared test modules
+
+Depend on these via `testImplementation(projects.*)` — never copy their utilities inline.
+
+Build test fixtures with **factory functions that default every argument** (`createXxx(id = 1, name = "Cat", … )`) rather than calling bloated constructors at each call site. A test then overrides only the fields relevant to it, so the intent stays visible and adding a model field doesn't churn every test. This is the idiom behind the `Mock*Factory` classes below — extend them instead of hand-rolling fixtures.
+
+### `:test:core` (pure JVM)
+`test/core/src/main/java/com/tangem/test/core/`. Re-exports as `api`: `test.coroutine`, `test.junit5`, `test.mockk`, `test.truth`, `test.turbine`. Use it as the one-line entry point to pull the whole unit-testing stack into a JVM module. Depends on `domain:core` and `arrow.core` (so its utilities can reference domain abstractions like `FlowProducer`).
+
+Utilities:
+- `TestCoroutineExt.getEmittedValues(flow)` — collect a `Flow` into a `List` from a `TestScope`.
+- `TestFlowProducerTools(scope, dispatcher)` — test double for `FlowProducerTools` that mirrors production `DefaultFlowProducerTools` (retryWhen + fallback + `distinctUntilChanged` + `shareIn`) on a caller-provided test scope/dispatcher, without analytics/logging. Pass `TestScope.backgroundScope` + a dispatcher built from `testScheduler` so the 2s retry delay is virtual-time-controllable. Use it for `FlowProducer` tests instead of mocking `FlowProducerTools`.
+- `@ProvideTestModels` — meta-annotation over JUnit 5 `@MethodSource("provideTestModels")` for parameterized tests.
+- `TruthArrowExt` — `assertEither`, `assertEitherRight`, `assertEitherLeft`, `assertSome`, `assertNone` for Arrow types.
+
+### `:common:test` (Android library — legacy, being retired)
+`common/test/src/main/java/com/tangem/common/test/`. Factories and fakes for domain/data models. Being phased out in favour of `:test:core` (JVM mechanisms) and `:test:mock` (mock factories); don't add new utilities here.
+
+- `TestAppCoroutineScope(testScope)` — test implementation of `AppCoroutineScope`.
+- `MockStateDataStore` — in-memory `DataStore` for tests.
+- `Mock*Factory` classes for `CryptoCurrency`, `UserWallet`, `NetworkStatus`, `ScanResponse`, `YieldDTO`, `QuoteResponse`, `UpdateWalletManagerResult` etc.
+
+### `:test:mock`
+`test/mock/`. Mock data for models not yet covered elsewhere (currently `MockAccounts`). Add to this module rather than creating new ad-hoc mock files.
+
+## Dispatchers
+
+Never use `Dispatchers.Main`/`IO`/`Default` directly in production code — always inject `CoroutineDispatcherProvider` from `core/utils`.
+
+In tests, override with `TestingCoroutineDispatcherProvider` (defined in `core/utils/src/main/java/com/tangem/utils/coroutines/CoroutineDispatcherProvider.kt`). By default `main`/`mainImmediate`/`io`/`default` are `Dispatchers.Unconfined`, while `single` is a single-thread `Executors.newFixedThreadPool(1)` dispatcher.
+
+For `Model`-layer tests inside features, construct it with a single `StandardTestDispatcher(testScheduler)` for all five roles (built from the enclosing `TestScope`) so `advanceUntilIdle()` controls execution. See any `features/*/impl` model test for the `TestScope.createTestingCoroutineDispatcherProvider()` helper.
+
+## Naming & placement
+
+- **Test class**: `FooTest` (singular noun). Not `FooSpec`, not `FooBehavior`, not `FooTests`.
+- **Test method**: backtick-quoted sentence that **must** follow `GIVEN … WHEN … THEN …` (uppercase). The name states the behaviour under test — precondition, action, expected outcome — not the implementation.
+  ```kotlin
+  @Test
+  fun `GIVEN currency status emitted WHEN model created THEN analytics sent`() = runTest { … }
+  ```
+  A part may collapse when trivial (e.g. `GIVEN no wallets WHEN load THEN returns empty`), but all three keywords stay present.
+- **Test body**: if the body is more than a one-liner (i.e. has distinct setup / action / check phases), it **must** be marked with `// Arrange`, `// Act`, `// Assert` comments. GWT names the behaviour from the outside; AAA structures the code inside.
+- **Location**: mirrored packages under `src/test/kotlin/`. No `src/testFixtures/` — shared helpers go to the modules above.
+
+## Unit-test skeleton (JUnit 5)
+
+```kotlin
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+internal class FooTest {
+
+    private val barUseCase: BarUseCase = mockk()
+    private val dispatchers = TestingCoroutineDispatcherProvider()
+
+    private val foo = Foo(barUseCase, dispatchers)
+
+    @BeforeEach
+    fun resetMocks() {
+        clearMocks(barUseCase)
+    }
+
+    @Test
+    fun `GIVEN bar returns right WHEN invoke THEN emits value`() = runTest {
+        // Arrange
+        coEvery { barUseCase(any()) } returns Either.Right(expected)
+
+        // Act
+        val actual = foo.invoke(input)
+
+        // Assert
+        assertThat(actual).isEqualTo(expected)
+        coVerify(exactly = 1) { barUseCase(input) }
+    }
+}
+```
+
+- `@TestInstance(Lifecycle.PER_CLASS)` is **opt-in per class, not the project default.** Add it only where you need a non-static `@MethodSource`/`provideTestModels` provider or want to share expensive setup across methods (~half of test classes do). The JUnit default stays `PER_METHOD` (a fresh instance per test). Beware: `PER_CLASS` reuses one instance across all methods, so mutable fields leak between tests — reset them in `@BeforeEach`.
+- **Group by method under test.** When a class/file exposes several functions and each accumulates many tests, don't keep one flat list — give each function its own `@Nested @TestInstance(Lifecycle.PER_CLASS) inner class`. The nesting maps the test structure onto the production API and keeps per-function setup local to its group.
+  ```kotlin
+  internal class DesignControllerTest {
+
+      @Nested
+      @TestInstance(TestInstance.Lifecycle.PER_CLASS)
+      inner class GetDesigns {
+          @Test fun `GIVEN … WHEN getDesigns THEN all fields included`() { … }
+          @Test fun `GIVEN limit WHEN getDesigns THEN list is capped`() { … }
+      }
+
+      @Nested
+      @TestInstance(TestInstance.Lifecycle.PER_CLASS)
+      inner class DeleteDesign {
+          @Test fun `GIVEN existing id WHEN deleteDesign THEN removed from db`() { … }
+      }
+  }
+  ```
+- You do **not** declare `useJUnitPlatform()` per module — the `configuration` convention plugin applies it (and the JUnit 5 engine) to every module. See "Gradle wiring" below.
+
+## MockK conventions
+
+- Field-level init: `private val x: T = mockk()`; use `mockk(relaxed = true)` only when stubs are not the subject of the test.
+- Stub coroutines with `coEvery { … } returns …` / `returnsMany(...)`; verify with `coVerify { … }`, `coVerify(exactly = n) { … }`, `coVerifyOrder { … }`.
+- Create mocks once as `val` fields and reset them with `clearMocks(...)` in `@BeforeEach` — recreating mocks (`x = mockk()` inside `@BeforeEach`) every test is measurably expensive (MockK instantiation dominates the runtime of small tests). Only recreate a field when the subject-under-test itself holds mutable state that must be fresh per test.
+- For companion/top-level objects use `mockkObject(Obj)` and pair with `unmockkObject(Obj)` in teardown.
+
+## Flow testing
+
+- Default to `TestScope.getEmittedValues(flow)` (from `:test:core`) when you just want the list of values produced during the test scope — this is the most common approach in the codebase.
+- Use **Turbine** (`flow.test { … }`) when you specifically need to assert on the emission *sequence* (ordering, intermediate items, completion/error timing), or for hot `SharedFlow`s where you must control collection start/stop.
+- Drive hot sources via `MutableSharedFlow` / `MutableStateFlow` and `advanceUntilIdle()` between emission and assertion.
+- For `FlowProducer` tests (retry/fallback/shareIn semantics), inject `TestFlowProducerTools` from `:test:core` and use Turbine + `advanceTimeBy(2001); runCurrent()` to step over the 2s retry window deterministically.
+
+## Parameterized tests
+
+When the same behaviour is exercised over a set of inputs, write **one parameterized test** — not several near-identical methods, and not one method with a stack of `assertThat(...)` calls over different inputs. Repeated asserts hide *which* input failed and stop at the first failure; a parameterized test reports each case separately. Add a new case = add a row to the provider.
+
+Use the project's `@ProvideTestModels` annotation — it wires `@MethodSource("provideTestModels")` for you.
+
+```kotlin
+@ParameterizedTest
+@ProvideTestModels
+fun create(model: CreateModel) = runTest { … }
+
+private data class CreateModel(val input: Input, val expected: Either<Error, Value>)
+
+private fun provideTestModels() = listOf(
+    CreateModel(input = …, expected = Either.Right(…)),
+    CreateModel(input = …, expected = Either.Left(Error.Foo)),
+)
+```
+
+`provideTestModels` is a non-static instance method, so the class needs `@TestInstance(Lifecycle.PER_CLASS)` (or a `@JvmStatic` provider in a companion).
+
+## Assertions
+
+- Default to Truth: `assertThat(actual).isEqualTo(expected)`, `.isInstanceOf(T::class.java)`, `.hasMessageThat().isEqualTo(…)`, `.isNull()`.
+- **Assert whole objects, not field-by-field.** When the type is a `data class`, build the expected instance and compare with one `isEqualTo(expected)` — the structural `equals`/`toString` gives a self-explanatory diff. For collections use `.containsExactly(…)` (add `.inOrder()` when order matters). Prefer this over a series of `assertThat(actual.id)…`, `assertThat(actual.name)…` checks, which produce opaque failures and miss unexpected fields.
+- For Arrow `Either`/`Option`, prefer `assertEither`, `assertEitherLeft`, `assertEitherRight`, `assertSome`, `assertNone` from `:test:core`.
+- Exception testing: `runCatching { … }.exceptionOrNull()` + Truth, not `assertThrows`.
+
+## Feature model tests
+
+`features/*/impl` Decompose models share a heavy dependency graph — extract a `XxxModelTestBase` with pre-built mocks/fixtures and inherit per-scenario test classes from it (see `features/staking/impl/.../presentation/model/StakingModelTestBase` as reference).
+
+Lifecycle:
+```kotlin
+val model = createModel(testScope = this)
+advanceUntilIdle()
+// assertions…
+model.onDestroy()
+```
+
+## Running tests
+
+```bash
+./gradlew unitTest                         # all JVM + debug/googleDebug unit tests (root aggregator)
+./gradlew :<module>:testDebugUnitTest      # single Android library module
+./gradlew :app:testGoogleDebugUnitTest     # app module
+./gradlew :<jvm-module>:test               # pure JVM module
+./gradlew :<module>:testDebugUnitTest --tests "com.tangem.<Fqn>Test"   # single class
+```
+
+The `unitTest` aggregator lives in the root `build.gradle.kts`; it is wired automatically for every `com.android.application`, `com.android.library`, and pure `org.jetbrains.kotlin.jvm` subproject — no need to touch it when adding a new module.
+
+## Gradle wiring for a new test-bearing module
+
+The `configuration` convention plugin (`configureUnitTests` in `plugins/configuration/.../TestConfigurations.kt`) centralizes the JUnit 5 setup for **every** module:
+
+1. `useJUnitPlatform()` on all `Test` tasks — so Jupiter tests are discovered (without it the default JUnit 4 runner runs zero Jupiter tests).
+2. `testRuntimeOnly(<test-junit5-engine>)` — the Jupiter runtime engine. The platform without the engine silently runs **zero** tests, so these two are paired in one place.
+3. Test logging (full exception format, standard streams, PASSED/SKIPPED/FAILED events, per-task summary).
+
+So a test module must **not** re-declare `useJUnitPlatform()`, the engine, or `testLogging { … }`. It only needs the Jupiter **API** (provided transitively by `:test:core`, or declared explicitly):
+
+```kotlin
+// Any module (JVM or Android library) — plugin already supplies platform + engine + logging
+plugins {
+    alias(deps.plugins.kotlin.jvm)   // or the android-library convention
+    id("configuration")
+}
+
+dependencies {
+    testImplementation(projects.test.core)   // junit5 (api) + mockk + turbine + truth + coroutine-test
+    testImplementation(projects.common.test) // add only if the tests need legacy model factories / fakes
+}
+```
+
+If a module doesn't want the full `:test:core` bundle, declare the Jupiter API directly with `testImplementation(deps.test.junit5)` — the engine still comes from the plugin, so never add `testRuntimeOnly(deps.test.junit5.engine)` per module.
+
+## Module type vs. layer
+
+The domain layer is **not** uniformly pure-JVM: domain modules are split roughly evenly between `org.jetbrains.kotlin.jvm` (pure JVM) and `com.android.library` modules. Don't assume the layer dictates the module type — check the `plugins { }` block to pick the right test task:
+
+- `kotlin.jvm` (pure JVM) → `./gradlew :<module>:test`
+- `com.android.library` / `com.android.application` → `./gradlew :<module>:testDebugUnitTest` (`:app` → `testGoogleDebugUnitTest`)
+
+`./gradlew unitTest` runs the right task for every module regardless of type.

app/build.gradle.kts

@@ -419,7 +419,6 @@ dependencies {
     /** Testing libraries */
     testImplementation(projects.test.core)
     testImplementation(projects.common.test)
-    testImplementation(deps.test.junit)
     androidTestImplementation(deps.test.junit.android)
     androidTestImplementation(deps.test.espresso)
     androidTestImplementation(deps.test.espresso.intents)

data/account/src/test/java/com/tangem/data/account/producer/DefaultMultiAccountListProducerTest.kt

@@ -8,22 +8,27 @@ import com.tangem.domain.core.flow.FlowProducerTools
 import com.tangem.domain.models.TokensSortType
 import com.tangem.domain.models.wallet.UserWallet
 import com.tangem.domain.models.wallet.UserWalletId
+import app.cash.turbine.test
+import com.tangem.test.core.TestFlowProducerTools
 import com.tangem.test.core.getEmittedValues
 import com.tangem.utils.coroutines.TestingCoroutineDispatcherProvider
 import io.mockk.*
+import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.emptyFlow
 import kotlinx.coroutines.flow.filterNotNull
 import kotlinx.coroutines.flow.flowOf
+import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.UnconfinedTestDispatcher
 import kotlinx.coroutines.test.runTest
 import org.junit.jupiter.api.AfterEach
-import org.junit.jupiter.api.Disabled
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.TestInstance
 
 /**
 [REDACTED_AUTHOR]
  */
+@OptIn(ExperimentalCoroutinesApi::class)
 @Suppress("UnusedFlow")
 @TestInstance(TestInstance.Lifecycle.PER_CLASS)
 class DefaultMultiAccountListProducerTest {
@@ -40,6 +45,23 @@ class DefaultMultiAccountListProducerTest {
         dispatchers = TestingCoroutineDispatcherProvider(),
     )
 
+    private fun TestScope.createProducer(): DefaultMultiAccountListProducer {
+        val testDispatcher = UnconfinedTestDispatcher(testScheduler)
+        return DefaultMultiAccountListProducer(
+            params = Unit,
+            flowProducerTools = TestFlowProducerTools(scope = backgroundScope, dispatcher = testDispatcher),
+            userWalletsListRepository = userWalletsListRepository,
+            singleAccountListSupplier = singleAccountListSupplier,
+            dispatchers = TestingCoroutineDispatcherProvider(
+                main = testDispatcher,
+                mainImmediate = testDispatcher,
+                io = testDispatcher,
+                default = testDispatcher,
+                single = testDispatcher,
+            ),
+        )
+    }
+
     private val userWalletId = UserWalletId("011")
     private val userWallet = mockk<UserWallet> {
         every { this@mockk.walletId } returns userWalletId
@@ -144,7 +166,6 @@ class DefaultMultiAccountListProducerTest {
         }
     }
 
-    @Disabled
     @Test
     fun `flow returns empty list if factory throws exception`() = runTest {
         // Arrange
@@ -154,12 +175,12 @@ class DefaultMultiAccountListProducerTest {
         val exception = RuntimeException("Converter error")
         every { singleAccountListSupplier.invoke(userWalletId) } throws exception
 
-        // Act
+        // Act / Assert: the factory throws -> retryWhen emits the empty fallback.
-        val actual = producer.produceWithFallback().let(::getEmittedValues)
+        // Stop before the 2s retry fires so the upstream is collected exactly once.
-
+        createProducer().produceWithFallback().test {
-        // Assert
+            Truth.assertThat(awaitItem()).isEqualTo(emptyList<AccountList>())
-        val expected = emptyList<AccountList>()
+            cancelAndIgnoreRemainingEvents()
-        Truth.assertThat(actual).containsExactly(expected)
+        }
 
         coVerifySequence {
             userWalletsListRepository.load()

data/networks/src/test/java/com/tangem/data/networks/multi/DefaultMultiNetworkStatusProducerTest.kt

@@ -1,5 +1,6 @@
 package com.tangem.data.networks.multi
 
+import app.cash.turbine.test
 import com.google.common.truth.Truth
 import com.tangem.common.test.domain.card.MockScanResponseFactory
 import com.tangem.common.test.domain.network.MockNetworkStatusFactory
@@ -11,23 +12,28 @@ import com.tangem.data.networks.store.NetworksStatusesStore
 import com.tangem.data.networks.toSimple
 import com.tangem.domain.card.configs.GenericCardConfig
 import com.tangem.domain.common.wallets.UserWalletsListRepository
-import com.tangem.domain.common.wallets.getSyncOrNull
 import com.tangem.domain.core.flow.FlowProducerTools
 import com.tangem.domain.models.network.NetworkStatus
 import com.tangem.domain.networks.multi.MultiNetworkStatusProducer
+import com.tangem.test.core.TestFlowProducerTools
 import com.tangem.test.core.getEmittedValues
 import com.tangem.utils.coroutines.TestingCoroutineDispatcherProvider
 import io.mockk.*
+import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.flow.*
+import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.UnconfinedTestDispatcher
+import kotlinx.coroutines.test.advanceTimeBy
+import kotlinx.coroutines.test.runCurrent
 import kotlinx.coroutines.test.runTest
 import org.junit.jupiter.api.BeforeEach
-import org.junit.jupiter.api.Disabled
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.TestInstance
 
 /**
 [REDACTED_AUTHOR]
  */
+@OptIn(ExperimentalCoroutinesApi::class)
 @TestInstance(TestInstance.Lifecycle.PER_CLASS)
 internal class DefaultMultiNetworkStatusProducerTest {
 
@@ -48,6 +54,24 @@ internal class DefaultMultiNetworkStatusProducerTest {
         flowProducerTools = flowProducerTools,
     )
 
+    private fun TestScope.createProducer(): DefaultMultiNetworkStatusProducer {
+        val testDispatcher = UnconfinedTestDispatcher(testScheduler)
+        return DefaultMultiNetworkStatusProducer(
+            params = params,
+            networksStatusesStore = networksStatusesStore,
+            userWalletsListRepository = userWalletsListRepository,
+            networkFactory = networkFactory,
+            dispatchers = TestingCoroutineDispatcherProvider(
+                main = testDispatcher,
+                mainImmediate = testDispatcher,
+                io = testDispatcher,
+                default = testDispatcher,
+                single = testDispatcher,
+            ),
+            flowProducerTools = TestFlowProducerTools(scope = backgroundScope, dispatcher = testDispatcher),
+        )
+    }
+
     @BeforeEach
     fun resetMocks() {
         clearMocks(networksStatusesStore, userWalletsListRepository, networkFactory)
@@ -294,7 +318,6 @@ internal class DefaultMultiNetworkStatusProducerTest {
         Truth.assertThat(actual2.first()).isEqualTo(expected2)
     }
 
-    @Disabled
     @Test
     fun `flow throws exception`() = runTest {
         // Arrange
@@ -338,30 +361,26 @@ internal class DefaultMultiNetworkStatusProducerTest {
         } returns statuses.last().network
         // endregion
 
-        val producerFlow = producer.produceWithFallback()
+        val producerFlow = createProducer().produceWithFallback()
 
-        // Act 1 (fallback)
+        producerFlow.test {
-        val actual1 = getEmittedValues(flow = producerFlow)
+            // first collection throws -> retryWhen emits the empty fallback, then waits 2s
+            Truth.assertThat(awaitItem()).isEqualTo(emptySet<NetworkStatus>())
 
-        // Assert
+            verify(inverse = true) {
-        val expected1 = emptySet<NetworkStatus>()
+                networkFactory.create(networkId = any(), derivationPath = any(), userWallet = any())
-        Truth.assertThat(actual1.size).isEqualTo(1)
+            }
-        Truth.assertThat(actual1.first()).isEqualTo(expected1)
 
-        verifyOrder(inverse = true) {
+            // recover the upstream and let the retry fire
-            userWalletsListRepository.getSyncOrNull(any())
+            innerFlow.value = true
-            networkFactory.create(networkId = any(), derivationPath = any(), userWallet = any())
+            advanceTimeBy(delayTimeMillis = 2001)
+            runCurrent()
+
+            Truth.assertThat(awaitItem()).isEqualTo(statuses)
+
+            cancelAndIgnoreRemainingEvents()
         }
 
-        // Act 2 (emit)
-        innerFlow.emit(value = true)
-        val actual2 = getEmittedValues(flow = producerFlow)
-
-        // Assert
-        val expected2 = statuses
-        Truth.assertThat(actual2.size).isEqualTo(1)
-        Truth.assertThat(actual2.first()).isEqualTo(expected2)
-
         verifyOrder {
             userWalletsListRepository.userWallets
             networkFactory.create(

data/networks/src/test/java/com/tangem/data/networks/single/DefaultSingleNetworkStatusProducerTest.kt

@@ -1,27 +1,33 @@
 package com.tangem.data.networks.single
 
+import app.cash.turbine.test
 import com.google.common.truth.Truth
 import com.tangem.common.test.domain.network.MockNetworkStatusFactory
 import com.tangem.common.test.domain.token.MockCryptoCurrencyFactory
-import com.tangem.domain.core.flow.FlowProducerTools
 import com.tangem.domain.models.network.NetworkStatus
 import com.tangem.domain.models.wallet.UserWalletId
 import com.tangem.domain.networks.multi.MultiNetworkStatusProducer
 import com.tangem.domain.networks.multi.MultiNetworkStatusSupplier
 import com.tangem.domain.networks.single.SingleNetworkStatusProducer
+import com.tangem.test.core.TestFlowProducerTools
 import com.tangem.test.core.getEmittedValues
 import com.tangem.utils.coroutines.TestingCoroutineDispatcherProvider
 import io.mockk.every
 import io.mockk.mockk
 import io.mockk.verify
+import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.flow.*
+import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.UnconfinedTestDispatcher
+import kotlinx.coroutines.test.advanceTimeBy
+import kotlinx.coroutines.test.runCurrent
 import kotlinx.coroutines.test.runTest
-import org.junit.jupiter.api.Disabled
 import org.junit.jupiter.api.Test
 
 /**
 [REDACTED_AUTHOR]
  */
+@OptIn(ExperimentalCoroutinesApi::class)
 internal class DefaultSingleNetworkStatusProducerTest {
 
     private val params = SingleNetworkStatusProducer.Params(
@@ -30,15 +36,22 @@ internal class DefaultSingleNetworkStatusProducerTest {
     )
 
     private val multiNetworkStatusSupplier = mockk<MultiNetworkStatusSupplier>()
-    private val dispatchers = TestingCoroutineDispatcherProvider()
-    private val flowProducerTools: FlowProducerTools = mockk()
 
-    private val producer = DefaultSingleNetworkStatusProducer(
+    private fun TestScope.createProducer(): DefaultSingleNetworkStatusProducer {
-        params = params,
+        val testDispatcher = UnconfinedTestDispatcher(testScheduler)
-        multiNetworkStatusSupplier = multiNetworkStatusSupplier,
+        return DefaultSingleNetworkStatusProducer(
-        dispatchers = dispatchers,
+            params = params,
-        flowProducerTools = flowProducerTools,
+            multiNetworkStatusSupplier = multiNetworkStatusSupplier,
-    )
+            dispatchers = TestingCoroutineDispatcherProvider(
+                main = testDispatcher,
+                mainImmediate = testDispatcher,
+                io = testDispatcher,
+                default = testDispatcher,
+                single = testDispatcher,
+            ),
+            flowProducerTools = TestFlowProducerTools(scope = backgroundScope, dispatcher = testDispatcher),
+        )
+    }
 
     @Test
     fun `test that flow is mapped for network from params`() = runTest {
@@ -53,7 +66,7 @@ internal class DefaultSingleNetworkStatusProducerTest {
         val multiParams = MultiNetworkStatusProducer.Params(userWalletId = params.userWalletId)
         every { multiNetworkStatusSupplier(multiParams) } returns expected
 
-        val actual = producer.produce()
+        val actual = createProducer().produce()
 
         verify { multiNetworkStatusSupplier(multiParams) }
 
@@ -63,11 +76,6 @@ internal class DefaultSingleNetworkStatusProducerTest {
         Truth.assertThat(values).isEqualTo(listOf(status))
     }
 
-    // TODO: rework for produceWithFallback() hot-SharedFlow semantics. These tests assert against
-    //  multiple cold collections, which is incompatible with shareIn(replay = 1) used in production.
-    //  Dormant under JUnit 4 (useJUnitPlatform without vintage); disabled to match
-    //  DefaultMultiNetworkStatusProducerTest.
-    @Disabled("Needs rework for produceWithFallback() hot-SharedFlow semantics")
     @Test
     fun `test that flow is updated if network status is updated`() = runTest {
         val expected = MutableSharedFlow<Set<NetworkStatus>>(replay = 2, extraBufferCapacity = 1)
@@ -75,30 +83,23 @@ internal class DefaultSingleNetworkStatusProducerTest {
         val multiParams = MultiNetworkStatusProducer.Params(userWalletId = params.userWalletId)
         every { multiNetworkStatusSupplier(multiParams) } returns expected
 
-        val actual = producer.produceWithFallback()
+        val actual = createProducer().produceWithFallback()
 
         verify { multiNetworkStatusSupplier(multiParams) }
 
-        // first emit
+        actual.test {
-        val status = MockNetworkStatusFactory.createMissedDerivation(params.network)
+            val status = MockNetworkStatusFactory.createMissedDerivation(params.network)
-        expected.emit(value = setOf(status))
+            expected.emit(value = setOf(status))
+            Truth.assertThat(awaitItem()).isEqualTo(status)
 
-        val values1 = getEmittedValues(flow = actual)
+            val updatedStatus = status.copy(value = NetworkStatus.Unreachable(null))
+            expected.emit(value = setOf(updatedStatus))
+            Truth.assertThat(awaitItem()).isEqualTo(updatedStatus)
 
-        Truth.assertThat(values1.size).isEqualTo(1)
+            cancelAndIgnoreRemainingEvents()
-        Truth.assertThat(values1).isEqualTo(listOf(status))
+        }
-
-        // second emit
-        val updatedStatus = status.copy(value = NetworkStatus.Unreachable(null))
-        expected.emit(value = setOf(updatedStatus))
-
-        val values2 = getEmittedValues(flow = actual)
-
-        Truth.assertThat(values2.size).isEqualTo(2)
-        Truth.assertThat(values2).isEqualTo(listOf(status, updatedStatus))
     }
 
-    @Disabled("Needs rework for produceWithFallback() hot-SharedFlow semantics")
     @Test
     fun `test that flow is filtered the same status`() = runTest {
         val expected = MutableSharedFlow<Set<NetworkStatus>>(replay = 2, extraBufferCapacity = 1)
@@ -106,29 +107,23 @@ internal class DefaultSingleNetworkStatusProducerTest {
         val multiParams = MultiNetworkStatusProducer.Params(userWalletId = params.userWalletId)
         every { multiNetworkStatusSupplier(multiParams) } returns expected
 
-        val actual = producer.produceWithFallback()
+        val actual = createProducer().produceWithFallback()
 
         verify { multiNetworkStatusSupplier(multiParams) }
 
-        // first emit
+        actual.test {
-        val status = MockNetworkStatusFactory.createMissedDerivation(params.network)
+            val status = MockNetworkStatusFactory.createMissedDerivation(params.network)
-        expected.emit(value = setOf(status))
+            expected.emit(value = setOf(status))
+            Truth.assertThat(awaitItem()).isEqualTo(status)
 
-        val values1 = getEmittedValues(flow = actual)
+            // same status again -> filtered out by distinctUntilChanged
+            expected.emit(value = setOf(status))
+            expectNoEvents()
 
-        Truth.assertThat(values1.size).isEqualTo(1)
+            cancelAndIgnoreRemainingEvents()
-        Truth.assertThat(values1).isEqualTo(listOf(status))
+        }
-
-        // second emit
-        expected.emit(value = setOf(status))
-
-        val values2 = getEmittedValues(flow = actual)
-
-        Truth.assertThat(values2.size).isEqualTo(1)
-        Truth.assertThat(values2).isEqualTo(listOf(status))
     }
 
-    @Disabled("Needs rework for produceWithFallback() infinite retryWhen + delay under virtual time")
     @Test
     fun `test if flow throws exception`() = runTest {
         val exception = IllegalStateException()
@@ -147,21 +142,24 @@ internal class DefaultSingleNetworkStatusProducerTest {
         val multiParams = MultiNetworkStatusProducer.Params(userWalletId = params.userWalletId)
         every { multiNetworkStatusSupplier(multiParams) } returns expected
 
-        val actual = producer.produceWithFallback()
+        val actual = createProducer().produceWithFallback()
 
         verify { multiNetworkStatusSupplier(multiParams) }
 
-        val values1 = getEmittedValues(flow = actual)
+        actual.test {
+            // first collection throws -> retryWhen emits the fallback, then waits 2s before retrying
+            val fallbackStatus = MockNetworkStatusFactory.createUnreachable(params.network)
+            Truth.assertThat(awaitItem()).isEqualTo(fallbackStatus)
 
-        Truth.assertThat(values1.size).isEqualTo(1)
+            // recover the upstream and let the retry fire
-        val fallbackStatus = MockNetworkStatusFactory.createUnreachable(params.network)
+            innerFlow.value = true
-        Truth.assertThat(values1).isEqualTo(listOf(fallbackStatus))
+            advanceTimeBy(delayTimeMillis = 2001)
+            runCurrent()
 
-        innerFlow.emit(value = true)
+            Truth.assertThat(awaitItem()).isEqualTo(status)
 
-        val values2 = getEmittedValues(flow = actual)
+            cancelAndIgnoreRemainingEvents()
-        Truth.assertThat(values2.size).isEqualTo(1)
+        }
-        Truth.assertThat(values2).isEqualTo(listOf(status))
     }
 
     @Test
@@ -173,7 +171,7 @@ internal class DefaultSingleNetworkStatusProducerTest {
         val multiParams = MultiNetworkStatusProducer.Params(userWalletId = params.userWalletId)
         every { multiNetworkStatusSupplier(multiParams) } returns expected
 
-        val actual = producer.produce()
+        val actual = createProducer().produce()
 
         verify { multiNetworkStatusSupplier(multiParams) }
 

data/quotes/src/test/java/com/tangem/data/quotes/single/DefaultSingleQuoteStatusProducerTest.kt

@@ -1,26 +1,32 @@
 package com.tangem.data.quotes.single
 
+import app.cash.turbine.test
 import com.google.common.truth.Truth
 import com.tangem.data.quotes.store.QuotesStatusesStore
-import com.tangem.domain.core.flow.FlowProducerTools
 import com.tangem.domain.models.StatusSource
 import com.tangem.domain.models.currency.CryptoCurrency
 import com.tangem.domain.models.quote.QuoteStatus
 import com.tangem.domain.quotes.single.SingleQuoteStatusProducer
+import com.tangem.test.core.TestFlowProducerTools
 import com.tangem.test.core.getEmittedValues
 import com.tangem.utils.coroutines.TestingCoroutineDispatcherProvider
 import io.mockk.every
 import io.mockk.mockk
 import io.mockk.verify
+import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.flow.*
+import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.UnconfinedTestDispatcher
+import kotlinx.coroutines.test.advanceTimeBy
+import kotlinx.coroutines.test.runCurrent
 import kotlinx.coroutines.test.runTest
-import org.junit.jupiter.api.Disabled
 import org.junit.jupiter.api.Test
 import java.math.BigDecimal
 
 /**
 [REDACTED_AUTHOR]
  */
+@OptIn(ExperimentalCoroutinesApi::class)
 internal class DefaultSingleQuoteStatusProducerTest {
 
     private val params = SingleQuoteStatusProducer.Params(
@@ -28,14 +34,22 @@ internal class DefaultSingleQuoteStatusProducerTest {
     )
 
     private val quotesStore = mockk<QuotesStatusesStore>()
-    private val flowProducerTools: FlowProducerTools = mockk()
 
-    private val producer = DefaultSingleQuoteStatusProducer(
+    private fun TestScope.createProducer(): DefaultSingleQuoteStatusProducer {
-        params = params,
+        val testDispatcher = UnconfinedTestDispatcher(testScheduler)
-        quotesStatusesStore = quotesStore,
+        return DefaultSingleQuoteStatusProducer(
-        flowProducerTools = flowProducerTools,
+            params = params,
-        dispatchers = TestingCoroutineDispatcherProvider(),
+            quotesStatusesStore = quotesStore,
-    )
+            flowProducerTools = TestFlowProducerTools(scope = backgroundScope, dispatcher = testDispatcher),
+            dispatchers = TestingCoroutineDispatcherProvider(
+                main = testDispatcher,
+                mainImmediate = testDispatcher,
+                io = testDispatcher,
+                default = testDispatcher,
+                single = testDispatcher,
+            ),
+        )
+    }
 
     @Test
     fun `test that flow is mapped for network from params`() = runTest {
@@ -49,7 +63,7 @@ internal class DefaultSingleQuoteStatusProducerTest {
 
         every { quotesStore.get() } returns storeQuote
 
-        val actual = producer.produce()
+        val actual = createProducer().produce()
 
         verify { quotesStore.get() }
 
@@ -59,74 +73,60 @@ internal class DefaultSingleQuoteStatusProducerTest {
         Truth.assertThat(values).isEqualTo(listOf(status))
     }
 
-    @Disabled("Needs rework for produceWithFallback() hot-SharedFlow semantics")
     @Test
     fun `test that flow is updated if quote is updated`() = runTest {
         val storeQuote = MutableSharedFlow<Set<QuoteStatus>>(replay = 2, extraBufferCapacity = 1)
 
         every { quotesStore.get() } returns storeQuote
 
-        val actual = producer.produceWithFallback()
+        val actual = createProducer().produceWithFallback()
 
         verify { quotesStore.get() }
 
-        // first emit
+        actual.test {
-        val status = QuoteStatus(rawCurrencyId = params.rawCurrencyId)
+            val status = QuoteStatus(rawCurrencyId = params.rawCurrencyId)
-        storeQuote.emit(value = setOf(status))
+            storeQuote.emit(value = setOf(status))
+            Truth.assertThat(awaitItem()).isEqualTo(status)
 
-        val values1 = getEmittedValues(flow = actual)
+            val updatedStatus = QuoteStatus(
+                rawCurrencyId = params.rawCurrencyId,
+                value = QuoteStatus.Data(
+                    fiatRate = BigDecimal.ONE,
+                    priceChange = BigDecimal.ZERO,
+                    fiatRateUSD = BigDecimal.ZERO,
+                    source = StatusSource.ACTUAL,
+                ),
+            )
+            storeQuote.emit(value = setOf(updatedStatus))
+            Truth.assertThat(awaitItem()).isEqualTo(updatedStatus)
 
-        Truth.assertThat(values1.size).isEqualTo(1)
+            cancelAndIgnoreRemainingEvents()
-        Truth.assertThat(values1).isEqualTo(listOf(status))
+        }
-
-        // second emit
-        val updatedStatus = QuoteStatus(
-            rawCurrencyId = params.rawCurrencyId,
-            value = QuoteStatus.Data(
-                fiatRate = BigDecimal.ONE,
-                priceChange = BigDecimal.ZERO,
-                fiatRateUSD = BigDecimal.ZERO,
-                source = StatusSource.ACTUAL,
-            ),
-        )
-        storeQuote.emit(value = setOf(updatedStatus))
-
-        val values2 = getEmittedValues(flow = actual)
-
-        Truth.assertThat(values2.size).isEqualTo(2)
-        Truth.assertThat(values2).isEqualTo(listOf(status, updatedStatus))
     }
 
-    @Disabled("Needs rework for produceWithFallback() hot-SharedFlow semantics")
     @Test
     fun `test that flow is filtered the same status`() = runTest {
         val storeQuote = MutableSharedFlow<Set<QuoteStatus>>(replay = 2, extraBufferCapacity = 1)
 
         every { quotesStore.get() } returns storeQuote
 
-        val actual = producer.produceWithFallback()
+        val actual = createProducer().produceWithFallback()
 
         verify { quotesStore.get() }
 
-        // first emit
+        actual.test {
-        val status = QuoteStatus(rawCurrencyId = params.rawCurrencyId)
+            val status = QuoteStatus(rawCurrencyId = params.rawCurrencyId)
-        storeQuote.emit(value = setOf(status))
+            storeQuote.emit(value = setOf(status))
+            Truth.assertThat(awaitItem()).isEqualTo(status)
 
-        val values1 = getEmittedValues(flow = actual)
+            // same status again -> filtered out by distinctUntilChanged
+            storeQuote.emit(value = setOf(status))
+            expectNoEvents()
 
-        Truth.assertThat(values1.size).isEqualTo(1)
+            cancelAndIgnoreRemainingEvents()
-        Truth.assertThat(values1).isEqualTo(listOf(status))
+        }
-
-        // second emit
-        storeQuote.emit(value = setOf(status))
-
-        val values2 = getEmittedValues(flow = actual)
-
-        Truth.assertThat(values2.size).isEqualTo(1)
-        Truth.assertThat(values2).isEqualTo(listOf(status))
     }
 
-    @Disabled("Needs rework for produceWithFallback() infinite retryWhen + delay under virtual time")
     @Test
     fun `test if flow throws exception`() = runTest {
         val exception = IllegalStateException()
@@ -152,24 +152,24 @@ internal class DefaultSingleQuoteStatusProducerTest {
 
         every { quotesStore.get() } returns storeQuote
 
-        val actual = producer.produceWithFallback()
+        val actual = createProducer().produceWithFallback()
 
         verify { quotesStore.get() }
 
-        val values1 = getEmittedValues(flow = actual)
+        actual.test {
+            val fallbackStatus = QuoteStatus(rawCurrencyId = params.rawCurrencyId)
+            Truth.assertThat(awaitItem()).isEqualTo(fallbackStatus)
 
-        Truth.assertThat(values1.size).isEqualTo(1)
+            innerFlow.value = true
-        val fallbackStatus = QuoteStatus(rawCurrencyId = params.rawCurrencyId)
+            advanceTimeBy(delayTimeMillis = 2001)
-        Truth.assertThat(values1).isEqualTo(listOf(fallbackStatus))
+            runCurrent()
 
-        innerFlow.emit(value = true)
+            Truth.assertThat(awaitItem()).isEqualTo(status)
 
-        val values2 = getEmittedValues(flow = actual)
+            cancelAndIgnoreRemainingEvents()
-        Truth.assertThat(values2.size).isEqualTo(1)
+        }
-        Truth.assertThat(values2).isEqualTo(listOf(status))
     }
 
-    @Disabled("Needs rework for produceWithFallback() hot-SharedFlow semantics")
     @Test
     fun `test if flow doesn't contain network from params`() = runTest {
         val storeFlow = flowOf(
@@ -180,12 +180,14 @@ internal class DefaultSingleQuoteStatusProducerTest {
 
         every { quotesStore.get() } returns storeFlow
 
-        val actual = producer.produceWithFallback()
+        val actual = createProducer().produceWithFallback()
 
         verify { quotesStore.get() }
 
-        val values = getEmittedValues(flow = actual)
+        actual.test {
-
+            // params currency (BTC) is not in the store -> nothing is emitted
-        Truth.assertThat(values.size).isEqualTo(0)
+            expectNoEvents()
+            cancelAndIgnoreRemainingEvents()
+        }
     }
 }

data/staking/src/test/kotlin/com/tangem/data/staking/multi/DefaultMultiStakingBalanceProducerTest.kt

@@ -12,19 +12,26 @@ import com.tangem.domain.models.staking.*
 import com.tangem.domain.models.wallet.UserWalletId
 import com.tangem.domain.staking.model.StakingIntegrationID
 import com.tangem.domain.staking.multi.MultiStakingBalanceProducer
+import app.cash.turbine.test
+import com.tangem.test.core.TestFlowProducerTools
 import com.tangem.test.core.getEmittedValues
 import com.tangem.utils.coroutines.TestingCoroutineDispatcherProvider
 import io.mockk.every
 import io.mockk.mockk
 import io.mockk.verify
+import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.flow.*
+import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.UnconfinedTestDispatcher
+import kotlinx.coroutines.test.advanceTimeBy
+import kotlinx.coroutines.test.runCurrent
 import kotlinx.coroutines.test.runTest
-import org.junit.jupiter.api.Disabled
 import org.junit.jupiter.api.Test
 
 /**
 [REDACTED_AUTHOR]
  */
+@OptIn(ExperimentalCoroutinesApi::class)
 internal class DefaultMultiStakingBalanceProducerTest {
 
     private val params = MultiStakingBalanceProducer.Params(userWalletId = UserWalletId("011"))
@@ -42,6 +49,25 @@ internal class DefaultMultiStakingBalanceProducerTest {
         dispatchers = dispatchers,
     )
 
+    // Producer wired with a real test FlowProducerTools (shareIn + retry + distinctUntilChanged)
+    // for produceWithFallback() cases.
+    private fun TestScope.createProducer(): DefaultMultiStakingBalanceProducer {
+        val testDispatcher = UnconfinedTestDispatcher(testScheduler)
+        return DefaultMultiStakingBalanceProducer(
+            params = params,
+            stakeKitBalancesStore = stakeKitBalancesStore,
+            p2PEthPoolBalancesStore = p2PEthPoolBalancesStore,
+            flowProducerTools = TestFlowProducerTools(scope = backgroundScope, dispatcher = testDispatcher),
+            dispatchers = TestingCoroutineDispatcherProvider(
+                main = testDispatcher,
+                mainImmediate = testDispatcher,
+                io = testDispatcher,
+                default = testDispatcher,
+                single = testDispatcher,
+            ),
+        )
+    }
+
     @Test
     fun `test that flow is mapped for user wallet id from params`() = runTest {
         val balances = setOf(
@@ -107,7 +133,6 @@ internal class DefaultMultiStakingBalanceProducerTest {
         Truth.assertThat(values2).isEqualTo(expected)
     }
 
-    @Disabled("Needs rework: distinctUntilChanged moved into produceWithFallback()/shareInProducer")
     @Test
     fun `test that flow is filtered the same balance`() = runTest {
         val networksStatusesFlow = MutableSharedFlow<Set<StakingBalance>>(replay = 2)
@@ -115,35 +140,28 @@ internal class DefaultMultiStakingBalanceProducerTest {
         every { stakeKitBalancesStore.get(params.userWalletId) } returns networksStatusesFlow
         every { p2PEthPoolBalancesStore.get(params.userWalletId) } returns flowOf(emptySet())
 
-        val actual = producer.produce()
+        val actual = createProducer().produceWithFallback()
 
-        // check after producer.produce()
         verify { stakeKitBalancesStore.get(params.userWalletId) }
         verify { p2PEthPoolBalancesStore.get(params.userWalletId) }
 
-        // first emit
+        actual.test {
-        val wrappers = setOf(
+            val wrappers = setOf(
-            MockYieldBalanceWrapperDTOFactory.createWithEmptyBalance(tonId).toDomain(),
+                MockYieldBalanceWrapperDTOFactory.createWithEmptyBalance(tonId).toDomain(),
-            MockYieldBalanceWrapperDTOFactory.createWithEmptyBalance(solanaId).toDomain(),
+                MockYieldBalanceWrapperDTOFactory.createWithEmptyBalance(solanaId).toDomain(),
-        )
+            )
 
-        networksStatusesFlow.emit(wrappers)
+            networksStatusesFlow.emit(wrappers)
+            Truth.assertThat(awaitItem()).isEqualTo(wrappers)
 
-        val values1 = getEmittedValues(flow = actual)
+            // same balances again -> filtered out by distinctUntilChanged
+            networksStatusesFlow.emit(wrappers)
+            expectNoEvents()
 
-        Truth.assertThat(values1.size).isEqualTo(1)
+            cancelAndIgnoreRemainingEvents()
-        Truth.assertThat(values1.first()).isEqualTo(wrappers)
+        }
-
-        // second emit
-        networksStatusesFlow.emit(wrappers)
-
-        val values2 = getEmittedValues(flow = actual)
-
-        Truth.assertThat(values2.size).isEqualTo(1)
-        Truth.assertThat(values2.first()).isEqualTo(wrappers)
     }
 
-    @Disabled("Needs rework for produceWithFallback() infinite retryWhen + delay under virtual time")
     @Test
     fun `test if flow throws exception`() = runTest {
         val exception = IllegalStateException()
@@ -165,23 +183,24 @@ internal class DefaultMultiStakingBalanceProducerTest {
         every { stakeKitBalancesStore.get(params.userWalletId) } returns networksStatusesFlow
         every { p2PEthPoolBalancesStore.get(params.userWalletId) } returns flowOf(emptySet())
 
-        val actual = producer.produceWithFallback()
+        val actual = createProducer().produceWithFallback()
 
-        // check after producer.produce()
         verify { stakeKitBalancesStore.get(params.userWalletId) }
         verify { p2PEthPoolBalancesStore.get(params.userWalletId) }
 
-        val values1 = getEmittedValues(flow = actual)
+        actual.test {
+            // first collection throws -> retryWhen emits the empty fallback, then waits 2s
+            Truth.assertThat(awaitItem()).isEqualTo(emptySet<StakingBalance>())
 
-        Truth.assertThat(values1.size).isEqualTo(1)
+            // recover the upstream and let the retry fire
-        Truth.assertThat(values1).isEqualTo(listOf(emptySet<StakingBalance>()))
+            innerFlow.value = true
+            advanceTimeBy(delayTimeMillis = 2001)
+            runCurrent()
 
-        innerFlow.emit(value = true)
+            Truth.assertThat(awaitItem()).isEqualTo(balances)
 
-        val values2 = getEmittedValues(flow = actual)
+            cancelAndIgnoreRemainingEvents()
-
+        }
-        Truth.assertThat(values2.size).isEqualTo(1)
-        Truth.assertThat(values2).isEqualTo(listOf(balances))
     }
 
     @Test

data/staking/src/test/kotlin/com/tangem/data/staking/single/DefaultSingleStakingBalanceProducerTest.kt

@@ -11,22 +11,29 @@ import com.tangem.domain.models.wallet.UserWalletId
 import com.tangem.domain.staking.multi.MultiStakingBalanceProducer
 import com.tangem.domain.staking.multi.MultiStakingBalanceSupplier
 import com.tangem.domain.staking.single.SingleStakingBalanceProducer
+import app.cash.turbine.test
+import com.tangem.test.core.TestFlowProducerTools
 import com.tangem.test.core.getEmittedValues
 import com.tangem.utils.coroutines.TestingCoroutineDispatcherProvider
 import io.mockk.clearMocks
 import io.mockk.every
 import io.mockk.mockk
 import io.mockk.verify
+import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.flow.*
+import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.UnconfinedTestDispatcher
+import kotlinx.coroutines.test.advanceTimeBy
+import kotlinx.coroutines.test.runCurrent
 import kotlinx.coroutines.test.runTest
 import org.junit.jupiter.api.BeforeEach
-import org.junit.jupiter.api.Disabled
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.TestInstance
 
 /**
 [REDACTED_AUTHOR]
  */
+@OptIn(ExperimentalCoroutinesApi::class)
 @TestInstance(TestInstance.Lifecycle.PER_CLASS)
 internal class DefaultSingleStakingBalanceProducerTest {
 
@@ -48,6 +55,23 @@ internal class DefaultSingleStakingBalanceProducerTest {
         flowProducerTools = flowProducerTools,
     )
 
+    private fun TestScope.createProducer(): DefaultSingleStakingBalanceProducer {
+        val testDispatcher = UnconfinedTestDispatcher(testScheduler)
+        return DefaultSingleStakingBalanceProducer(
+            params = params,
+            multiStakingBalanceSupplier = multiNetworkStatusSupplier,
+            analyticsExceptionHandler = analyticsExceptionHandler,
+            dispatchers = TestingCoroutineDispatcherProvider(
+                main = testDispatcher,
+                mainImmediate = testDispatcher,
+                io = testDispatcher,
+                default = testDispatcher,
+                single = testDispatcher,
+            ),
+            flowProducerTools = TestFlowProducerTools(scope = backgroundScope, dispatcher = testDispatcher),
+        )
+    }
+
     @BeforeEach
     fun resetMocks() {
         clearMocks(multiNetworkStatusSupplier, analyticsExceptionHandler)
@@ -77,7 +101,6 @@ internal class DefaultSingleStakingBalanceProducerTest {
         verify(exactly = 1) { multiNetworkStatusSupplier(multiParams) }
     }
 
-    @Disabled
     @Test
     fun `flow is updated if staking balance is updated`() = runTest {
         // Arrange
@@ -86,31 +109,23 @@ internal class DefaultSingleStakingBalanceProducerTest {
         val multiParams = MultiStakingBalanceProducer.Params(userWalletId = params.userWalletId)
         every { multiNetworkStatusSupplier(multiParams) } returns multiFlow
 
-        val producerFlow = producer.produceWithFallback()
+        val producerFlow = createProducer().produceWithFallback()
 
-        val balance = MockYieldBalanceWrapperDTOFactory.createWithBalance(tonId).toDomain()
+        producerFlow.test {
-        val updatedBalance = StakingBalance.Error(stakingId = tonId)
+            val balance = MockYieldBalanceWrapperDTOFactory.createWithBalance(tonId).toDomain()
+            multiFlow.emit(value = setOf(balance))
+            Truth.assertThat(awaitItem()).isEqualTo(balance)
 
-        // Act (first emit)
+            val updatedBalance = StakingBalance.Error(stakingId = tonId)
-        multiFlow.emit(value = setOf(balance))
+            multiFlow.emit(value = setOf(updatedBalance))
-        val actual1 = getEmittedValues(flow = producerFlow)
+            Truth.assertThat(awaitItem()).isEqualTo(updatedBalance)
 
-        // Assert (first emit)
+            cancelAndIgnoreRemainingEvents()
-        Truth.assertThat(actual1).hasSize(1)
+        }
-        Truth.assertThat(actual1).containsExactly(balance)
-
-        // Act (second emit)
-        multiFlow.emit(value = setOf(updatedBalance))
-        val actual2 = getEmittedValues(flow = producerFlow)
-
-        // Assert (second emit)
-        Truth.assertThat(actual2).hasSize(2)
-        Truth.assertThat(actual2).containsExactly(balance, updatedBalance)
 
         verify(exactly = 1) { multiNetworkStatusSupplier(multiParams) }
     }
 
-    @Disabled
     @Test
     fun `flow is filtered the same status`() = runTest {
         // Arrange
@@ -119,30 +134,23 @@ internal class DefaultSingleStakingBalanceProducerTest {
         val multiParams = MultiStakingBalanceProducer.Params(userWalletId = params.userWalletId)
         every { multiNetworkStatusSupplier(multiParams) } returns multiFlow
 
-        val producerFlow = producer.produceWithFallback()
+        val producerFlow = createProducer().produceWithFallback()
 
-        val balance = MockYieldBalanceWrapperDTOFactory.createWithBalance(tonId).toDomain()
+        producerFlow.test {
+            val balance = MockYieldBalanceWrapperDTOFactory.createWithBalance(tonId).toDomain()
+            multiFlow.emit(value = setOf(balance))
+            Truth.assertThat(awaitItem()).isEqualTo(balance)
 
-        // Act (first emit)
+            // same balance again -> filtered out by distinctUntilChanged
-        multiFlow.emit(value = setOf(balance))
+            multiFlow.emit(value = setOf(balance))
-        val actual1 = getEmittedValues(flow = producerFlow)
+            expectNoEvents()
 
-        // Assert (first emit)
+            cancelAndIgnoreRemainingEvents()
-        Truth.assertThat(actual1).hasSize(1)
+        }
-        Truth.assertThat(actual1).containsExactly(balance)
-
-        // Act (second emit)
-        multiFlow.emit(value = setOf(balance))
-        val actual2 = getEmittedValues(flow = producerFlow)
-
-        // Assert (second emit)
-        Truth.assertThat(actual2).hasSize(1)
-        Truth.assertThat(actual2).containsExactly(balance)
 
         verify(exactly = 1) { multiNetworkStatusSupplier(multiParams) }
     }
 
-    @Disabled
     @Test
     fun `flow throws exception`() = runTest {
         // Arrange
@@ -163,23 +171,22 @@ internal class DefaultSingleStakingBalanceProducerTest {
         val multiParams = MultiStakingBalanceProducer.Params(userWalletId = params.userWalletId)
         every { multiNetworkStatusSupplier(multiParams) } returns multiFlow
 
-        val producerFlow = producer.produceWithFallback()
+        val producerFlow = createProducer().produceWithFallback()
 
-        // Act (first emit)
+        producerFlow.test {
-        val actual1 = getEmittedValues(flow = producerFlow)
+            // first collection throws -> retryWhen emits the fallback, then waits 2s
+            val fallbackStatus = StakingBalance.Error(stakingId = tonId.copy(address = "0x1"))
+            Truth.assertThat(awaitItem()).isEqualTo(fallbackStatus)
 
-        // Assert (first emit)
+            // recover the upstream and let the retry fire
-        val fallbackStatus = StakingBalance.Error(stakingId = tonId.copy(address = "0x1"))
+            innerFlow.value = true
+            advanceTimeBy(delayTimeMillis = 2001)
+            runCurrent()
 
-        Truth.assertThat(actual1).hasSize(1)
+            Truth.assertThat(awaitItem()).isEqualTo(balance)
-        Truth.assertThat(actual1).containsExactly(fallbackStatus)
 
-        // Act (second emit)
+            cancelAndIgnoreRemainingEvents()
-        innerFlow.emit(value = true)
+        }
-        val actual2 = getEmittedValues(flow = producerFlow)
-
-        Truth.assertThat(actual2).hasSize(1)
-        Truth.assertThat(actual2).containsExactly(balance)
 
         verify(exactly = 1) { multiNetworkStatusSupplier(multiParams) }
     }

data/staking/src/test/kotlin/com/tangem/data/staking/store/StakingBalancesStoreUpdateMethodsTest.kt

@@ -13,7 +13,6 @@ import com.tangem.domain.models.staking.StakingID
 import com.tangem.domain.models.wallet.UserWalletId
 import kotlinx.coroutines.flow.firstOrNull
 import kotlinx.coroutines.test.runTest
-import org.junit.jupiter.api.Disabled
 import org.junit.jupiter.api.Test
 
 /**
@@ -137,9 +136,6 @@ internal class StakingBalancesStoreUpdateMethodsTest {
         Truth.assertThat(persistenceStore.data.firstOrNull()).isEqualTo(emptyMap<String, Set<YieldBalanceWrapperDTO>>())
     }
 
-    // TODO: revisit — expected is built via wrapper.toDomain(ONLY_CACHE) but that yields source=ACTUAL,
-    //  while storeError() applies ONLY_CACHE. Mock/toDomain vs production source handling needs review.
-    @Disabled("Source-mismatch between toDomain() expectation and storeError() output; needs domain review")
     @Test
     fun `store error if runtime store contains balance with this id`() = runTest {
         val wrapper = MockYieldBalanceWrapperDTOFactory.createWithBalance(stakingId)
@@ -152,8 +148,10 @@ internal class StakingBalancesStoreUpdateMethodsTest {
 
         store.storeError(userWalletId = userWalletId, stakingIds = setOf(stakingId))
 
+        // storeError keeps the existing (CACHE) balance and downgrades its source to ONLY_CACHE.
+        // toDomain(ONLY_CACHE) can't express this: the converter maps any non-CACHE source to ACTUAL.
         val runtimeExpected = mapOf(
-            userWalletId to setOf(wrapper.toDomain(source = StatusSource.ONLY_CACHE)),
+            userWalletId to setOf(wrapper.toDomain(source = StatusSource.CACHE).copySealed(source = StatusSource.ONLY_CACHE)),
         )
 
         Truth.assertThat(runtimeStore.getSyncOrNull()).isEqualTo(runtimeExpected)

test/core/build.gradle.kts

@@ -5,6 +5,7 @@ plugins {
 
 dependencies {
     implementation(projects.core.utils)
+    implementation(projects.domain.core)
     implementation(deps.arrow.core)
 
     api(deps.androidx.datastore.core)

test/core/src/main/java/com/tangem/test/core/TestFlowProducerTools.kt

@@ -0,0 +1,58 @@
+package com.tangem.test.core
+
+import com.tangem.domain.core.flow.FlowProducer
+import com.tangem.domain.core.flow.FlowProducerTools
+import kotlinx.coroutines.CoroutineDispatcher
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.SharedFlow
+import kotlinx.coroutines.flow.SharingStarted
+import kotlinx.coroutines.flow.distinctUntilChanged
+import kotlinx.coroutines.flow.flowOn
+import kotlinx.coroutines.flow.retryWhen
+import kotlinx.coroutines.flow.shareIn
+
+/**
+ * Test implementation of [FlowProducerTools] that mirrors the production `DefaultFlowProducerTools`
+ * behaviour (retryWhen + fallback + distinctUntilChanged + shareIn) on a caller-provided test
+ * scope/dispatcher, without analytics/logging.
+ *
+ * Pass a [TestScope.backgroundScope] and a `StandardTestDispatcher`/`UnconfinedTestDispatcher` built
+ * from the test scheduler so virtual time (e.g. the 2s retry delay) is controllable.
+ *
+[REDACTED_AUTHOR]
+ */
+class TestFlowProducerTools(
+    private val scope: CoroutineScope,
+    private val dispatcher: CoroutineDispatcher,
+) : FlowProducerTools {
+
+    override fun <T> shareInProducer(
+        flow: Flow<T>,
+        flowProducer: FlowProducer<T>,
+        withRetryWhen: Boolean,
+    ): SharedFlow<T> {
+        var upstream = flow
+
+        if (withRetryWhen) {
+            upstream = upstream.retryWhen { _, _ ->
+                flowProducer.fallback.onSome { emit(it) }
+                delay(timeMillis = 2000)
+                true
+            }
+        }
+
+        return upstream
+            .flowOn(dispatcher)
+            .distinctUntilChanged()
+            .shareIn(
+                scope = scope,
+                replay = 1,
+                started = SharingStarted.WhileSubscribed(
+                    stopTimeoutMillis = 0,
+                    replayExpirationMillis = 0,
+                ),
+            )
+    }
+}