Merge changes Id57cc404,Ieaf30071,Idb2b182c,I5de259d8,I5d7762b7 into tm-qpr-dev am: f932e8e60f am: a4bc86d57a

Original change: https://googleplex-android-review.googlesource.com/c/platform/frameworks/base/+/21957487

Change-Id: I8025a694640bd7635f6b08778ba5eb81c8c1d2b0
Signed-off-by: Automerger Merge Worker <android-build-automerger-merge-worker@system.gserviceaccount.com>

4 files changed, 567 insertions, 141 deletions

diff --git a/packages/SystemUI/src/com/android/systemui/statusbar/pipeline/mobile/data/repository/prod/MobileConnectionRepositoryImpl.kt b/packages/SystemUI/src/com/android/systemui/statusbar/pipeline/mobile/data/repository/prod/MobileConnectionRepositoryImpl.kt
index f1fc3868d690..f866d65e2d2d 100644
--- a/packages/SystemUI/src/com/android/systemui/statusbar/pipeline/mobile/data/repository/prod/MobileConnectionRepositoryImpl.kt
+++ b/packages/SystemUI/src/com/android/systemui/statusbar/pipeline/mobile/data/repository/prod/MobileConnectionRepositoryImpl.kt
@@ -18,7 +18,6 @@ package com.android.systemui.statusbar.pipeline.mobile.data.repository.prod
 
 import android.content.Context
 import android.content.IntentFilter
-import android.telephony.CellSignalStrength
 import android.telephony.CellSignalStrength.SIGNAL_STRENGTH_NONE_OR_UNKNOWN
 import android.telephony.CellSignalStrengthCdma
 import android.telephony.ServiceState
@@ -28,12 +27,12 @@ import android.telephony.TelephonyCallback
 import android.telephony.TelephonyDisplayInfo
 import android.telephony.TelephonyDisplayInfo.OVERRIDE_NETWORK_TYPE_NONE
 import android.telephony.TelephonyManager
-import android.telephony.TelephonyManager.ERI_OFF
+import android.telephony.TelephonyManager.ERI_FLASH
+import android.telephony.TelephonyManager.ERI_ON
 import android.telephony.TelephonyManager.EXTRA_SUBSCRIPTION_ID
 import android.telephony.TelephonyManager.NETWORK_TYPE_UNKNOWN
 import com.android.settingslib.Utils
 import com.android.systemui.broadcast.BroadcastDispatcher
-import com.android.systemui.common.coroutine.ConflatedCallbackFlow.conflatedCallbackFlow
 import com.android.systemui.dagger.qualifiers.Application
 import com.android.systemui.dagger.qualifiers.Background
 import com.android.systemui.log.table.TableLogBuffer
@@ -59,16 +58,14 @@ import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.asExecutor
 import kotlinx.coroutines.channels.awaitClose
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.MutableSharedFlow
-import kotlinx.coroutines.flow.SharedFlow
 import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.callbackFlow
 import kotlinx.coroutines.flow.filter
-import kotlinx.coroutines.flow.filterIsInstance
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.mapLatest
 import kotlinx.coroutines.flow.mapNotNull
-import kotlinx.coroutines.flow.shareIn
+import kotlinx.coroutines.flow.scan
 import kotlinx.coroutines.flow.stateIn
 
 /**
@@ -100,8 +97,6 @@ class MobileConnectionRepositoryImpl(
         }
     }
 
-    private val telephonyCallbackEvent = MutableSharedFlow<Unit>(extraBufferCapacity = 1)
-
     /**
      * This flow defines the single shared connection to system_server via TelephonyCallback. Any
      * new callback should be added to this listener and funneled through callbackEvents via a data
@@ -109,9 +104,15 @@ class MobileConnectionRepositoryImpl(
      *
      * The reason we need to do this is because TelephonyManager limits the number of registered
      * listeners per-process, so we don't want to create a new listener for every callback.
+     *
+     * A note on the design for back pressure here: We use the [coalesce] operator here to change
+     * the backpressure strategy to store exactly the last callback event of _each type_ here, as
+     * opposed to the default strategy which is to drop the oldest event (regardless of type). This
+     * means that we should never miss any single event as long as the flow has been started.
      */
-    private val callbackEvents: SharedFlow<CallbackEvent> =
-        conflatedCallbackFlow {
+    private val callbackEvents: StateFlow<TelephonyCallbackState> = run {
+        val initial = TelephonyCallbackState()
+        callbackFlow {
                 val callback =
                     object :
                         TelephonyCallback(),
@@ -165,48 +166,50 @@ class MobileConnectionRepositoryImpl(
                 telephonyManager.registerTelephonyCallback(bgDispatcher.asExecutor(), callback)
                 awaitClose { telephonyManager.unregisterTelephonyCallback(callback) }
             }
-            .shareIn(scope, SharingStarted.WhileSubscribed())
+            .scan(initial = initial) { state, event -> state.applyEvent(event) }
+            .stateIn(scope = scope, started = SharingStarted.WhileSubscribed(), initial)
+    }
 
     override val isEmergencyOnly =
         callbackEvents
-            .filterIsInstance<CallbackEvent.OnServiceStateChanged>()
+            .mapNotNull { it.onServiceStateChanged }
             .map { it.serviceState.isEmergencyOnly }
             .stateIn(scope, SharingStarted.WhileSubscribed(), false)
 
     override val isRoaming =
         callbackEvents
-            .filterIsInstance<CallbackEvent.OnServiceStateChanged>()
+            .mapNotNull { it.onServiceStateChanged }
             .map { it.serviceState.roaming }
             .stateIn(scope, SharingStarted.WhileSubscribed(), false)
 
     override val operatorAlphaShort =
         callbackEvents
-            .filterIsInstance<CallbackEvent.OnServiceStateChanged>()
+            .mapNotNull { it.onServiceStateChanged }
             .map { it.serviceState.operatorAlphaShort }
             .stateIn(scope, SharingStarted.WhileSubscribed(), null)
 
     override val isInService =
         callbackEvents
-            .filterIsInstance<CallbackEvent.OnServiceStateChanged>()
+            .mapNotNull { it.onServiceStateChanged }
             .map { Utils.isInService(it.serviceState) }
             .stateIn(scope, SharingStarted.WhileSubscribed(), false)
 
     override val isGsm =
         callbackEvents
-            .filterIsInstance<CallbackEvent.OnSignalStrengthChanged>()
+            .mapNotNull { it.onSignalStrengthChanged }
             .map { it.signalStrength.isGsm }
             .stateIn(scope, SharingStarted.WhileSubscribed(), false)
 
     override val cdmaLevel =
         callbackEvents
-            .filterIsInstance<CallbackEvent.OnSignalStrengthChanged>()
+            .mapNotNull { it.onSignalStrengthChanged }
             .map {
                 it.signalStrength.getCellSignalStrengths(CellSignalStrengthCdma::class.java).let {
                     strengths ->
                     if (strengths.isNotEmpty()) {
                         strengths[0].level
                     } else {
-                        CellSignalStrength.SIGNAL_STRENGTH_NONE_OR_UNKNOWN
+                        SIGNAL_STRENGTH_NONE_OR_UNKNOWN
                     }
                 }
             }
@@ -214,19 +217,19 @@ class MobileConnectionRepositoryImpl(
 
     override val primaryLevel =
         callbackEvents
-            .filterIsInstance<CallbackEvent.OnSignalStrengthChanged>()
+            .mapNotNull { it.onSignalStrengthChanged }
             .map { it.signalStrength.level }
             .stateIn(scope, SharingStarted.WhileSubscribed(), SIGNAL_STRENGTH_NONE_OR_UNKNOWN)
 
     override val dataConnectionState =
         callbackEvents
-            .filterIsInstance<CallbackEvent.OnDataConnectionStateChanged>()
+            .mapNotNull { it.onDataConnectionStateChanged }
             .map { it.dataState.toDataConnectionType() }
             .stateIn(scope, SharingStarted.WhileSubscribed(), Disconnected)
 
     override val dataActivityDirection =
         callbackEvents
-            .filterIsInstance<CallbackEvent.OnDataActivity>()
+            .mapNotNull { it.onDataActivity }
             .map { it.direction.toMobileDataActivityModel() }
             .stateIn(
                 scope,
@@ -236,28 +239,26 @@ class MobileConnectionRepositoryImpl(
 
     override val carrierNetworkChangeActive =
         callbackEvents
-            .filterIsInstance<CallbackEvent.OnCarrierNetworkChange>()
+            .mapNotNull { it.onCarrierNetworkChange }
             .map { it.active }
             .stateIn(scope, SharingStarted.WhileSubscribed(), false)
 
     override val resolvedNetworkType =
         callbackEvents
-            .filterIsInstance<CallbackEvent.OnDisplayInfoChanged>()
+            .mapNotNull { it.onDisplayInfoChanged }
             .map {
-                if (it.telephonyDisplayInfo.networkType == NETWORK_TYPE_UNKNOWN) {
-                    UnknownNetworkType
-                } else if (
-                    it.telephonyDisplayInfo.overrideNetworkType == OVERRIDE_NETWORK_TYPE_NONE
-                ) {
-                    DefaultNetworkType(
-                        mobileMappingsProxy.toIconKey(it.telephonyDisplayInfo.networkType)
-                    )
-                } else {
+                if (it.telephonyDisplayInfo.overrideNetworkType != OVERRIDE_NETWORK_TYPE_NONE) {
                     OverrideNetworkType(
                         mobileMappingsProxy.toIconKeyOverride(
                             it.telephonyDisplayInfo.overrideNetworkType
                         )
                     )
+                } else if (it.telephonyDisplayInfo.networkType != NETWORK_TYPE_UNKNOWN) {
+                    DefaultNetworkType(
+                        mobileMappingsProxy.toIconKey(it.telephonyDisplayInfo.networkType)
+                    )
+                } else {
+                    UnknownNetworkType
                 }
             }
             .stateIn(scope, SharingStarted.WhileSubscribed(), UnknownNetworkType)
@@ -282,7 +283,10 @@ class MobileConnectionRepositoryImpl(
 
     override val cdmaRoaming: StateFlow<Boolean> =
         telephonyPollingEvent
-            .mapLatest { telephonyManager.cdmaEnhancedRoamingIndicatorDisplayNumber != ERI_OFF }
+            .mapLatest {
+                val cdmaEri = telephonyManager.cdmaEnhancedRoamingIndicatorDisplayNumber
+                cdmaEri == ERI_ON || cdmaEri == ERI_FLASH
+            }
             .stateIn(scope, SharingStarted.WhileSubscribed(), false)
 
     override val networkName: StateFlow<NetworkNameModel> =
@@ -300,7 +304,8 @@ class MobileConnectionRepositoryImpl(
     override val dataEnabled = run {
         val initial = telephonyManager.isDataConnectionAllowed
         callbackEvents
-            .mapNotNull { (it as? CallbackEvent.OnDataEnabledChanged)?.enabled }
+            .mapNotNull { it.onDataEnabledChanged }
+            .map { it.enabled }
             .stateIn(scope, SharingStarted.WhileSubscribed(), initial)
     }
 
@@ -344,12 +349,41 @@ class MobileConnectionRepositoryImpl(
  * Wrap every [TelephonyCallback] we care about in a data class so we can accept them in a single
  * shared flow and then split them back out into other flows.
  */
-private sealed interface CallbackEvent {
-    data class OnServiceStateChanged(val serviceState: ServiceState) : CallbackEvent
-    data class OnSignalStrengthChanged(val signalStrength: SignalStrength) : CallbackEvent
-    data class OnDataConnectionStateChanged(val dataState: Int) : CallbackEvent
-    data class OnDataActivity(val direction: Int) : CallbackEvent
+sealed interface CallbackEvent {
     data class OnCarrierNetworkChange(val active: Boolean) : CallbackEvent
-    data class OnDisplayInfoChanged(val telephonyDisplayInfo: TelephonyDisplayInfo) : CallbackEvent
+    data class OnDataActivity(val direction: Int) : CallbackEvent
+    data class OnDataConnectionStateChanged(val dataState: Int) : CallbackEvent
     data class OnDataEnabledChanged(val enabled: Boolean) : CallbackEvent
+    data class OnDisplayInfoChanged(val telephonyDisplayInfo: TelephonyDisplayInfo) : CallbackEvent
+    data class OnServiceStateChanged(val serviceState: ServiceState) : CallbackEvent
+    data class OnSignalStrengthChanged(val signalStrength: SignalStrength) : CallbackEvent
+}
+
+/**
+ * A simple box type for 1-to-1 mapping of [CallbackEvent] to the batched event. Used in conjunction
+ * with [scan] to make sure we don't drop important callbacks due to late subscribers
+ */
+data class TelephonyCallbackState(
+    val onDataActivity: CallbackEvent.OnDataActivity? = null,
+    val onCarrierNetworkChange: CallbackEvent.OnCarrierNetworkChange? = null,
+    val onDataConnectionStateChanged: CallbackEvent.OnDataConnectionStateChanged? = null,
+    val onDataEnabledChanged: CallbackEvent.OnDataEnabledChanged? = null,
+    val onDisplayInfoChanged: CallbackEvent.OnDisplayInfoChanged? = null,
+    val onServiceStateChanged: CallbackEvent.OnServiceStateChanged? = null,
+    val onSignalStrengthChanged: CallbackEvent.OnSignalStrengthChanged? = null,
+) {
+    fun applyEvent(event: CallbackEvent): TelephonyCallbackState {
+        return when (event) {
+            is CallbackEvent.OnCarrierNetworkChange -> copy(onCarrierNetworkChange = event)
+            is CallbackEvent.OnDataActivity -> copy(onDataActivity = event)
+            is CallbackEvent.OnDataConnectionStateChanged ->
+                copy(onDataConnectionStateChanged = event)
+            is CallbackEvent.OnDataEnabledChanged -> copy(onDataEnabledChanged = event)
+            is CallbackEvent.OnDisplayInfoChanged -> copy(onDisplayInfoChanged = event)
+            is CallbackEvent.OnServiceStateChanged -> {
+                copy(onServiceStateChanged = event)
+            }
+            is CallbackEvent.OnSignalStrengthChanged -> copy(onSignalStrengthChanged = event)
+        }
+    }
 }
diff --git a/packages/SystemUI/tests/src/com/android/systemui/statusbar/pipeline/mobile/data/repository/prod/MobileConnectionRepositoryTest.kt b/packages/SystemUI/tests/src/com/android/systemui/statusbar/pipeline/mobile/data/repository/prod/MobileConnectionRepositoryTest.kt
index f6e595924f58..542b688b162d 100644
--- a/packages/SystemUI/tests/src/com/android/systemui/statusbar/pipeline/mobile/data/repository/prod/MobileConnectionRepositoryTest.kt
+++ b/packages/SystemUI/tests/src/com/android/systemui/statusbar/pipeline/mobile/data/repository/prod/MobileConnectionRepositoryTest.kt
@@ -18,17 +18,16 @@ package com.android.systemui.statusbar.pipeline.mobile.data.repository.prod
 
 import android.content.Intent
 import android.telephony.CarrierConfigManager.KEY_INFLATE_SIGNAL_STRENGTH_BOOL
-import android.telephony.CellSignalStrengthCdma
 import android.telephony.NetworkRegistrationInfo
 import android.telephony.ServiceState
 import android.telephony.ServiceState.STATE_IN_SERVICE
 import android.telephony.ServiceState.STATE_OUT_OF_SERVICE
-import android.telephony.SignalStrength
 import android.telephony.TelephonyCallback
 import android.telephony.TelephonyCallback.DataActivityListener
 import android.telephony.TelephonyCallback.ServiceStateListener
 import android.telephony.TelephonyDisplayInfo
 import android.telephony.TelephonyDisplayInfo.OVERRIDE_NETWORK_TYPE_LTE_CA
+import android.telephony.TelephonyDisplayInfo.OVERRIDE_NETWORK_TYPE_NONE
 import android.telephony.TelephonyManager
 import android.telephony.TelephonyManager.DATA_ACTIVITY_DORMANT
 import android.telephony.TelephonyManager.DATA_ACTIVITY_IN
@@ -68,6 +67,7 @@ import com.android.systemui.statusbar.pipeline.mobile.data.model.SystemUiCarrier
 import com.android.systemui.statusbar.pipeline.mobile.data.model.toNetworkNameModel
 import com.android.systemui.statusbar.pipeline.mobile.data.repository.FakeMobileConnectionsRepository
 import com.android.systemui.statusbar.pipeline.mobile.data.repository.MobileConnectionRepository.Companion.DEFAULT_NUM_LEVELS
+import com.android.systemui.statusbar.pipeline.mobile.data.repository.prod.MobileTelephonyHelpers.signalStrength
 import com.android.systemui.statusbar.pipeline.mobile.util.FakeMobileMappingsProxy
 import com.android.systemui.statusbar.pipeline.shared.data.model.DataActivityModel
 import com.android.systemui.statusbar.pipeline.shared.data.model.toMobileDataActivityModel
@@ -75,14 +75,12 @@ import com.android.systemui.util.mockito.any
 import com.android.systemui.util.mockito.mock
 import com.android.systemui.util.mockito.whenever
 import com.google.common.truth.Truth.assertThat
-import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.ExperimentalCoroutinesApi
-import kotlinx.coroutines.cancel
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.onEach
-import kotlinx.coroutines.runBlocking
-import org.junit.After
+import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.UnconfinedTestDispatcher
+import kotlinx.coroutines.test.runTest
 import org.junit.Before
 import org.junit.Test
 import org.mockito.Mock
@@ -99,7 +97,6 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
     @Mock private lateinit var logger: MobileInputLogger
     @Mock private lateinit var tableLogger: TableLogBuffer
 
-    private val scope = CoroutineScope(IMMEDIATE)
     private val mobileMappings = FakeMobileMappingsProxy()
     private val systemUiCarrierConfig =
         SystemUiCarrierConfig(
@@ -107,6 +104,9 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
             createTestConfig(),
         )
 
+    private val testDispatcher = UnconfinedTestDispatcher()
+    private val testScope = TestScope(testDispatcher)
+
     @Before
     fun setUp() {
         MockitoAnnotations.initMocks(this)
@@ -124,21 +124,16 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
                 systemUiCarrierConfig,
                 fakeBroadcastDispatcher,
                 mobileMappings,
-                IMMEDIATE,
+                testDispatcher,
                 logger,
                 tableLogger,
-                scope,
+                testScope.backgroundScope,
             )
     }
 
-    @After
-    fun tearDown() {
-        scope.cancel()
-    }
-
     @Test
     fun emergencyOnly() =
-        runBlocking(IMMEDIATE) {
+        testScope.runTest {
             var latest: Boolean? = null
             val job = underTest.isEmergencyOnly.onEach { latest = it }.launchIn(this)
 
@@ -154,18 +149,15 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
 
     @Test
     fun emergencyOnly_toggles() =
-        runBlocking(IMMEDIATE) {
+        testScope.runTest {
             var latest: Boolean? = null
             val job = underTest.isEmergencyOnly.onEach { latest = it }.launchIn(this)
 
             val callback = getTelephonyCallbackForType<ServiceStateListener>()
-            val serviceState = ServiceState()
-            serviceState.isEmergencyOnly = true
-            callback.onServiceStateChanged(serviceState)
+            callback.onServiceStateChanged(ServiceState().also { it.isEmergencyOnly = true })
             assertThat(latest).isTrue()
 
-            serviceState.isEmergencyOnly = false
-            callback.onServiceStateChanged(serviceState)
+            callback.onServiceStateChanged(ServiceState().also { it.isEmergencyOnly = false })
 
             assertThat(latest).isFalse()
 
@@ -174,7 +166,7 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
 
     @Test
     fun cdmaLevelUpdates() =
-        runBlocking(IMMEDIATE) {
+        testScope.runTest {
             var latest: Int? = null
             val job = underTest.cdmaLevel.onEach { latest = it }.launchIn(this)
 
@@ -194,7 +186,7 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
 
     @Test
     fun gsmLevelUpdates() =
-        runBlocking(IMMEDIATE) {
+        testScope.runTest {
             var latest: Int? = null
             val job = underTest.primaryLevel.onEach { latest = it }.launchIn(this)
 
@@ -214,7 +206,7 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
 
     @Test
     fun isGsm() =
-        runBlocking(IMMEDIATE) {
+        testScope.runTest {
             var latest: Boolean? = null
             val job = underTest.isGsm.onEach { latest = it }.launchIn(this)
 
@@ -234,7 +226,7 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
 
     @Test
     fun dataConnectionState_connected() =
-        runBlocking(IMMEDIATE) {
+        testScope.runTest {
             var latest: DataConnectionState? = null
             val job = underTest.dataConnectionState.onEach { latest = it }.launchIn(this)
 
@@ -249,7 +241,7 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
 
     @Test
     fun dataConnectionState_connecting() =
-        runBlocking(IMMEDIATE) {
+        testScope.runTest {
             var latest: DataConnectionState? = null
             val job = underTest.dataConnectionState.onEach { latest = it }.launchIn(this)
 
@@ -264,7 +256,7 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
 
     @Test
     fun dataConnectionState_disconnected() =
-        runBlocking(IMMEDIATE) {
+        testScope.runTest {
             var latest: DataConnectionState? = null
             val job = underTest.dataConnectionState.onEach { latest = it }.launchIn(this)
 
@@ -279,7 +271,7 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
 
     @Test
     fun dataConnectionState_disconnecting() =
-        runBlocking(IMMEDIATE) {
+        testScope.runTest {
             var latest: DataConnectionState? = null
             val job = underTest.dataConnectionState.onEach { latest = it }.launchIn(this)
 
@@ -294,7 +286,7 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
 
     @Test
     fun dataConnectionState_suspended() =
-        runBlocking(IMMEDIATE) {
+        testScope.runTest {
             var latest: DataConnectionState? = null
             val job = underTest.dataConnectionState.onEach { latest = it }.launchIn(this)
 
@@ -309,7 +301,7 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
 
     @Test
     fun dataConnectionState_handoverInProgress() =
-        runBlocking(IMMEDIATE) {
+        testScope.runTest {
             var latest: DataConnectionState? = null
             val job = underTest.dataConnectionState.onEach { latest = it }.launchIn(this)
 
@@ -324,7 +316,7 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
 
     @Test
     fun dataConnectionState_unknown() =
-        runBlocking(IMMEDIATE) {
+        testScope.runTest {
             var latest: DataConnectionState? = null
             val job = underTest.dataConnectionState.onEach { latest = it }.launchIn(this)
 
@@ -339,7 +331,7 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
 
     @Test
     fun dataConnectionState_invalid() =
-        runBlocking(IMMEDIATE) {
+        testScope.runTest {
             var latest: DataConnectionState? = null
             val job = underTest.dataConnectionState.onEach { latest = it }.launchIn(this)
 
@@ -354,7 +346,7 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
 
     @Test
     fun dataActivity() =
-        runBlocking(IMMEDIATE) {
+        testScope.runTest {
             var latest: DataActivityModel? = null
             val job = underTest.dataActivityDirection.onEach { latest = it }.launchIn(this)
 
@@ -368,7 +360,7 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
 
     @Test
     fun carrierNetworkChange() =
-        runBlocking(IMMEDIATE) {
+        testScope.runTest {
             var latest: Boolean? = null
             val job = underTest.carrierNetworkChangeActive.onEach { latest = it }.launchIn(this)
 
@@ -382,7 +374,7 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
 
     @Test
     fun networkType_default() =
-        runBlocking(IMMEDIATE) {
+        testScope.runTest {
             var latest: ResolvedNetworkType? = null
             val job = underTest.resolvedNetworkType.onEach { latest = it }.launchIn(this)
 
@@ -395,7 +387,7 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
 
     @Test
     fun networkType_unknown_hasCorrectKey() =
-        runBlocking(IMMEDIATE) {
+        testScope.runTest {
             var latest: ResolvedNetworkType? = null
             val job = underTest.resolvedNetworkType.onEach { latest = it }.launchIn(this)
 
@@ -413,14 +405,19 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
 
     @Test
     fun networkType_updatesUsingDefault() =
-        runBlocking(IMMEDIATE) {
+        testScope.runTest {
             var latest: ResolvedNetworkType? = null
             val job = underTest.resolvedNetworkType.onEach { latest = it }.launchIn(this)
 
             val callback = getTelephonyCallbackForType<TelephonyCallback.DisplayInfoListener>()
+            val overrideType = OVERRIDE_NETWORK_TYPE_NONE
             val type = NETWORK_TYPE_LTE
             val expected = DefaultNetworkType(mobileMappings.toIconKey(type))
-            val ti = mock<TelephonyDisplayInfo>().also { whenever(it.networkType).thenReturn(type) }
+            val ti =
+                mock<TelephonyDisplayInfo>().also {
+                    whenever(it.overrideNetworkType).thenReturn(overrideType)
+                    whenever(it.networkType).thenReturn(type)
+                }
             callback.onDisplayInfoChanged(ti)
 
             assertThat(latest).isEqualTo(expected)
@@ -430,7 +427,7 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
 
     @Test
     fun networkType_updatesUsingOverride() =
-        runBlocking(IMMEDIATE) {
+        testScope.runTest {
             var latest: ResolvedNetworkType? = null
             val job = underTest.resolvedNetworkType.onEach { latest = it }.launchIn(this)
 
@@ -450,16 +447,38 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
         }
 
     @Test
+    fun networkType_unknownNetworkWithOverride_usesOverrideKey() =
+        testScope.runTest {
+            var latest: ResolvedNetworkType? = null
+            val job = underTest.resolvedNetworkType.onEach { latest = it }.launchIn(this)
+
+            val callback = getTelephonyCallbackForType<TelephonyCallback.DisplayInfoListener>()
+            val unknown = NETWORK_TYPE_UNKNOWN
+            val type = OVERRIDE_NETWORK_TYPE_LTE_CA
+            val expected = OverrideNetworkType(mobileMappings.toIconKeyOverride(type))
+            val ti =
+                mock<TelephonyDisplayInfo>().also {
+                    whenever(it.networkType).thenReturn(unknown)
+                    whenever(it.overrideNetworkType).thenReturn(type)
+                }
+            callback.onDisplayInfoChanged(ti)
+
+            assertThat(latest).isEqualTo(expected)
+
+            job.cancel()
+        }
+
+    @Test
     fun dataEnabled_initial_false() =
-        runBlocking(IMMEDIATE) {
+        testScope.runTest {
             whenever(telephonyManager.isDataConnectionAllowed).thenReturn(false)
 
             assertThat(underTest.dataEnabled.value).isFalse()
         }
 
     @Test
-    fun `is data enabled - tracks telephony callback`() =
-        runBlocking(IMMEDIATE) {
+    fun isDataEnabled_tracksTelephonyCallback() =
+        testScope.runTest {
             var latest: Boolean? = null
             val job = underTest.dataEnabled.onEach { latest = it }.launchIn(this)
 
@@ -479,7 +498,7 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
 
     @Test
     fun numberOfLevels_isDefault() =
-        runBlocking(IMMEDIATE) {
+        testScope.runTest {
             var latest: Int? = null
             val job = underTest.numberOfLevels.onEach { latest = it }.launchIn(this)
 
@@ -489,51 +508,68 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
         }
 
     @Test
-    fun `roaming - cdma - queries telephony manager`() =
-        runBlocking(IMMEDIATE) {
+    fun roaming_cdma_queriesTelephonyManager() =
+        testScope.runTest {
             var latest: Boolean? = null
             val job = underTest.cdmaRoaming.onEach { latest = it }.launchIn(this)
 
             val cb = getTelephonyCallbackForType<ServiceStateListener>()
 
-            val serviceState = ServiceState()
-            serviceState.roaming = false
-
-            // CDMA roaming is off, GSM roaming is off
+            // CDMA roaming is off, GSM roaming is on
             whenever(telephonyManager.cdmaEnhancedRoamingIndicatorDisplayNumber).thenReturn(ERI_OFF)
-            cb.onServiceStateChanged(serviceState)
+            cb.onServiceStateChanged(ServiceState().also { it.roaming = true })
 
             assertThat(latest).isFalse()
 
-            // CDMA roaming is off, GSM roaming is on
+            // CDMA roaming is on, GSM roaming is off
             whenever(telephonyManager.cdmaEnhancedRoamingIndicatorDisplayNumber).thenReturn(ERI_ON)
-            cb.onServiceStateChanged(serviceState)
+            cb.onServiceStateChanged(ServiceState().also { it.roaming = false })
 
             assertThat(latest).isTrue()
 
             job.cancel()
         }
 
+    /**
+     * [TelephonyManager.getCdmaEnhancedRoamingIndicatorDisplayNumber] returns -1 if the service is
+     * not running or if there is an error while retrieving the cdma ERI
+     */
     @Test
-    fun `roaming - gsm - queries service state`() =
-        runBlocking(IMMEDIATE) {
+    fun cdmaRoaming_ignoresNegativeOne() =
+        testScope.runTest {
             var latest: Boolean? = null
-            val job = underTest.isRoaming.onEach { latest = it }.launchIn(this)
+            val job = underTest.cdmaRoaming.onEach { latest = it }.launchIn(this)
 
             val serviceState = ServiceState()
             serviceState.roaming = false
 
             val cb = getTelephonyCallbackForType<ServiceStateListener>()
 
+            // CDMA roaming is unavailable (-1), GSM roaming is off
+            whenever(telephonyManager.cdmaEnhancedRoamingIndicatorDisplayNumber).thenReturn(-1)
+            cb.onServiceStateChanged(serviceState)
+
+            assertThat(latest).isFalse()
+
+            job.cancel()
+        }
+
+    @Test
+    fun roaming_gsm_queriesServiceState() =
+        testScope.runTest {
+            var latest: Boolean? = null
+            val job = underTest.isRoaming.onEach { latest = it }.launchIn(this)
+
+            val cb = getTelephonyCallbackForType<ServiceStateListener>()
+
             // CDMA roaming is off, GSM roaming is off
             whenever(telephonyManager.cdmaEnhancedRoamingIndicatorDisplayNumber).thenReturn(ERI_OFF)
-            cb.onServiceStateChanged(serviceState)
+            cb.onServiceStateChanged(ServiceState().also { it.roaming = false })
 
             assertThat(latest).isFalse()
 
             // CDMA roaming is off, GSM roaming is on
-            serviceState.roaming = true
-            cb.onServiceStateChanged(serviceState)
+            cb.onServiceStateChanged(ServiceState().also { it.roaming = true })
 
             assertThat(latest).isTrue()
 
@@ -541,8 +577,8 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
         }
 
     @Test
-    fun `activity - updates from callback`() =
-        runBlocking(IMMEDIATE) {
+    fun activity_updatesFromCallback() =
+        testScope.runTest {
             var latest: DataActivityModel? = null
             val job = underTest.dataActivityDirection.onEach { latest = it }.launchIn(this)
 
@@ -578,8 +614,8 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
         }
 
     @Test
-    fun `network name - default`() =
-        runBlocking(IMMEDIATE) {
+    fun networkName_default() =
+        testScope.runTest {
             var latest: NetworkNameModel? = null
             val job = underTest.networkName.onEach { latest = it }.launchIn(this)
 
@@ -589,8 +625,8 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
         }
 
     @Test
-    fun `network name - uses broadcast info - returns derived`() =
-        runBlocking(IMMEDIATE) {
+    fun networkName_usesBroadcastInfo_returnsDerived() =
+        testScope.runTest {
             var latest: NetworkNameModel? = null
             val job = underTest.networkName.onEach { latest = it }.launchIn(this)
 
@@ -606,8 +642,8 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
         }
 
     @Test
-    fun `network name - broadcast not for this sub id - keeps old value`() =
-        runBlocking(IMMEDIATE) {
+    fun networkName_broadcastNotForThisSubId_keepsOldValue() =
+        testScope.runTest {
             var latest: NetworkNameModel? = null
             val job = underTest.networkName.onEach { latest = it }.launchIn(this)
 
@@ -631,8 +667,8 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
         }
 
     @Test
-    fun `network name - broadcast has no data - updates to default`() =
-        runBlocking(IMMEDIATE) {
+    fun networkName_broadcastHasNoData_updatesToDefault() =
+        testScope.runTest {
             var latest: NetworkNameModel? = null
             val job = underTest.networkName.onEach { latest = it }.launchIn(this)
 
@@ -658,8 +694,8 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
         }
 
     @Test
-    fun `operatorAlphaShort - tracked`() =
-        runBlocking(IMMEDIATE) {
+    fun operatorAlphaShort_tracked() =
+        testScope.runTest {
             var latest: String? = null
 
             val job = underTest.operatorAlphaShort.onEach { latest = it }.launchIn(this)
@@ -680,33 +716,45 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
         }
 
     @Test
-    fun `connection model - isInService - not iwlan`() =
-        runBlocking(IMMEDIATE) {
+    fun isInService_notIwlan() =
+        testScope.runTest {
             var latest: Boolean? = null
             val job = underTest.isInService.onEach { latest = it }.launchIn(this)
 
-            val serviceState = ServiceState()
-            serviceState.voiceRegState = STATE_IN_SERVICE
-            serviceState.dataRegState = STATE_IN_SERVICE
-
-            getTelephonyCallbackForType<ServiceStateListener>().onServiceStateChanged(serviceState)
+            getTelephonyCallbackForType<ServiceStateListener>()
+                .onServiceStateChanged(
+                    ServiceState().also {
+                        it.voiceRegState = STATE_IN_SERVICE
+                        it.dataRegState = STATE_IN_SERVICE
+                    }
+                )
 
             assertThat(latest).isTrue()
 
-            serviceState.voiceRegState = STATE_OUT_OF_SERVICE
-            getTelephonyCallbackForType<ServiceStateListener>().onServiceStateChanged(serviceState)
+            getTelephonyCallbackForType<ServiceStateListener>()
+                .onServiceStateChanged(
+                    ServiceState().also {
+                        it.dataRegState = STATE_IN_SERVICE
+                        it.voiceRegState = STATE_OUT_OF_SERVICE
+                    }
+                )
             assertThat(latest).isTrue()
 
-            serviceState.dataRegState = STATE_OUT_OF_SERVICE
-            getTelephonyCallbackForType<ServiceStateListener>().onServiceStateChanged(serviceState)
+            getTelephonyCallbackForType<ServiceStateListener>()
+                .onServiceStateChanged(
+                    ServiceState().also {
+                        it.voiceRegState = STATE_OUT_OF_SERVICE
+                        it.dataRegState = STATE_OUT_OF_SERVICE
+                    }
+                )
             assertThat(latest).isFalse()
 
             job.cancel()
         }
 
     @Test
-    fun `connection model - isInService - is iwlan - voice out of service - data in service`() =
-        runBlocking(IMMEDIATE) {
+    fun isInService_isIwlan_voiceOutOfService_dataInService() =
+        testScope.runTest {
             var latest: Boolean? = null
             val job = underTest.isInService.onEach { latest = it }.launchIn(this)
 
@@ -730,8 +778,8 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
         }
 
     @Test
-    fun `number of levels - uses carrier config`() =
-        runBlocking(IMMEDIATE) {
+    fun numberOfLevels_usesCarrierConfig() =
+        testScope.runTest {
             var latest: Int? = null
             val job = underTest.numberOfLevels.onEach { latest = it }.launchIn(this)
 
@@ -756,19 +804,6 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
         return MobileTelephonyHelpers.getTelephonyCallbackForType(telephonyManager)
     }
 
-    /** Convenience constructor for SignalStrength */
-    private fun signalStrength(gsmLevel: Int, cdmaLevel: Int, isGsm: Boolean): SignalStrength {
-        val signalStrength = mock<SignalStrength>()
-        whenever(signalStrength.isGsm).thenReturn(isGsm)
-        whenever(signalStrength.level).thenReturn(gsmLevel)
-        val cdmaStrength =
-            mock<CellSignalStrengthCdma>().also { whenever(it.level).thenReturn(cdmaLevel) }
-        whenever(signalStrength.getCellSignalStrengths(CellSignalStrengthCdma::class.java))
-            .thenReturn(listOf(cdmaStrength))
-
-        return signalStrength
-    }
-
     private fun spnIntent(
         subId: Int = SUB_1_ID,
         showSpn: Boolean = true,
@@ -785,7 +820,6 @@ class MobileConnectionRepositoryTest : SysuiTestCase() {
         }
 
     companion object {
-        private val IMMEDIATE = Dispatchers.Main.immediate
         private const val SUB_1_ID = 1
 
         private val DEFAULT_NAME = NetworkNameModel.Default("default name")
diff --git a/packages/SystemUI/tests/src/com/android/systemui/statusbar/pipeline/mobile/data/repository/prod/MobileConnectionTelephonySmokeTests.kt b/packages/SystemUI/tests/src/com/android/systemui/statusbar/pipeline/mobile/data/repository/prod/MobileConnectionTelephonySmokeTests.kt
new file mode 100644
index 000000000000..bbf04ed28fd7
--- /dev/null
+++ b/packages/SystemUI/tests/src/com/android/systemui/statusbar/pipeline/mobile/data/repository/prod/MobileConnectionTelephonySmokeTests.kt
@@ -0,0 +1,341 @@
+/*
+ * Copyright (C) 2023 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.android.systemui.statusbar.pipeline.mobile.data.repository.prod
+
+import android.telephony.ServiceState
+import android.telephony.TelephonyCallback
+import android.telephony.TelephonyCallback.CarrierNetworkListener
+import android.telephony.TelephonyCallback.DataActivityListener
+import android.telephony.TelephonyCallback.DataConnectionStateListener
+import android.telephony.TelephonyCallback.DataEnabledListener
+import android.telephony.TelephonyCallback.DisplayInfoListener
+import android.telephony.TelephonyCallback.ServiceStateListener
+import android.telephony.TelephonyDisplayInfo
+import android.telephony.TelephonyManager
+import android.telephony.TelephonyManager.DATA_ACTIVITY_INOUT
+import android.telephony.TelephonyManager.NETWORK_TYPE_LTE
+import android.telephony.TelephonyManager.NETWORK_TYPE_UNKNOWN
+import androidx.test.filters.SmallTest
+import com.android.systemui.SysuiTestCase
+import com.android.systemui.log.table.TableLogBuffer
+import com.android.systemui.statusbar.pipeline.mobile.data.MobileInputLogger
+import com.android.systemui.statusbar.pipeline.mobile.data.model.DataConnectionState
+import com.android.systemui.statusbar.pipeline.mobile.data.model.NetworkNameModel
+import com.android.systemui.statusbar.pipeline.mobile.data.model.ResolvedNetworkType
+import com.android.systemui.statusbar.pipeline.mobile.data.model.SystemUiCarrierConfig
+import com.android.systemui.statusbar.pipeline.mobile.data.model.SystemUiCarrierConfigTest
+import com.android.systemui.statusbar.pipeline.mobile.data.repository.FakeMobileConnectionsRepository
+import com.android.systemui.statusbar.pipeline.mobile.data.repository.prod.MobileTelephonyHelpers.getTelephonyCallbackForType
+import com.android.systemui.statusbar.pipeline.mobile.data.repository.prod.MobileTelephonyHelpers.signalStrength
+import com.android.systemui.statusbar.pipeline.mobile.util.FakeMobileMappingsProxy
+import com.android.systemui.statusbar.pipeline.shared.data.model.DataActivityModel
+import com.android.systemui.util.mockito.mock
+import com.android.systemui.util.mockito.whenever
+import com.google.common.truth.Truth.assertThat
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.UnconfinedTestDispatcher
+import kotlinx.coroutines.test.runTest
+import org.junit.Before
+import org.junit.Test
+import org.mockito.Mock
+import org.mockito.MockitoAnnotations
+
+/**
+ * Test class to stress test the TelephonyCallbacks that we listen to. In particular, the callbacks
+ * all come back in on a single listener (for reasons defined in the system). This test is built to
+ * ensure that we don't miss any important callbacks.
+ *
+ * Kind of like an interaction test case build just for [TelephonyCallback]
+ *
+ * The list of telephony callbacks we use is: [TelephonyCallback.CarrierNetworkListener]
+ * [TelephonyCallback.DataActivityListener] [TelephonyCallback.DataConnectionStateListener]
+ * [TelephonyCallback.DataEnabledListener] [TelephonyCallback.DisplayInfoListener]
+ * [TelephonyCallback.ServiceStateListener] [TelephonyCallback.SignalStrengthsListener]
+ *
+ * Because each of these callbacks comes in on the same callbackFlow, collecting on a field backed
+ * by only a single callback can immediately create backpressure on the other fields related to a
+ * mobile connection.
+ *
+ * This test should be designed to test _at least_ each individual callback in a smoke-test fashion.
+ * The way we will achieve this is as follows:
+ * 1. Start up a listener (A) collecting on a field which is _not under test_
+ * 2. Send a single event to a telephony callback which supports the field under test (B)
+ * 3. Send many (may be as few as 2) events to the callback backing A to ensure we start seeing
+ *    backpressure on other fields NOTE: poor handling of backpressure here would normally cause B
+ *    to get dropped
+ * 4. Start up a new collector for B
+ * 5. Assert that B has the state sent in step #2
+ */
+@Suppress("EXPERIMENTAL_IS_NOT_ENABLED")
+@OptIn(ExperimentalCoroutinesApi::class)
+@SmallTest
+class MobileConnectionTelephonySmokeTests : SysuiTestCase() {
+    private lateinit var underTest: MobileConnectionRepositoryImpl
+    private lateinit var connectionsRepo: FakeMobileConnectionsRepository
+
+    @Mock private lateinit var telephonyManager: TelephonyManager
+    @Mock private lateinit var logger: MobileInputLogger
+    @Mock private lateinit var tableLogger: TableLogBuffer
+
+    private val mobileMappings = FakeMobileMappingsProxy()
+    private val systemUiCarrierConfig =
+        SystemUiCarrierConfig(
+            SUB_1_ID,
+            SystemUiCarrierConfigTest.createTestConfig(),
+        )
+
+    private val testDispatcher = UnconfinedTestDispatcher()
+    private val testScope = TestScope(testDispatcher)
+
+    @Before
+    fun setUp() {
+        MockitoAnnotations.initMocks(this)
+        whenever(telephonyManager.subscriptionId).thenReturn(SUB_1_ID)
+
+        connectionsRepo = FakeMobileConnectionsRepository(mobileMappings, tableLogger)
+
+        underTest =
+            MobileConnectionRepositoryImpl(
+                context,
+                SUB_1_ID,
+                DEFAULT_NAME,
+                SEP,
+                telephonyManager,
+                systemUiCarrierConfig,
+                fakeBroadcastDispatcher,
+                mobileMappings,
+                testDispatcher,
+                logger,
+                tableLogger,
+                testScope.backgroundScope,
+            )
+    }
+
+    @Test
+    fun carrierNetworkChangeListener_noisyActivity() =
+        testScope.runTest {
+            var latest: Boolean? = null
+
+            // Start collecting data activity; don't care about the result
+            val activityJob = underTest.dataActivityDirection.launchIn(this)
+            val activityCallback = getTelephonyCallbackForType<DataActivityListener>()
+
+            val callback = getTelephonyCallbackForType<CarrierNetworkListener>()
+            callback.onCarrierNetworkChange(true)
+
+            flipActivity(100, activityCallback)
+
+            val job = underTest.carrierNetworkChangeActive.onEach { latest = it }.launchIn(this)
+
+            assertThat(latest).isTrue()
+
+            activityJob.cancel()
+            job.cancel()
+        }
+
+    @Test
+    fun dataActivityLate_noisyDisplayInfo() =
+        testScope.runTest {
+            var latest: DataActivityModel? = null
+
+            // start collecting displayInfo; don't care about the result
+            val displayInfoJob = underTest.resolvedNetworkType.launchIn(this)
+
+            val activityCallback = getTelephonyCallbackForType<DataActivityListener>()
+            activityCallback.onDataActivity(DATA_ACTIVITY_INOUT)
+
+            val displayInfoCallback = getTelephonyCallbackForType<DisplayInfoListener>()
+            val type1 = NETWORK_TYPE_UNKNOWN
+            val type2 = NETWORK_TYPE_LTE
+            val t1 =
+                mock<TelephonyDisplayInfo>().also { whenever(it.networkType).thenReturn(type1) }
+            val t2 =
+                mock<TelephonyDisplayInfo>().also { whenever(it.networkType).thenReturn(type2) }
+
+            flipDisplayInfo(100, listOf(t1, t2), displayInfoCallback)
+
+            val job = underTest.dataActivityDirection.onEach { latest = it }.launchIn(this)
+
+            assertThat(latest)
+                .isEqualTo(
+                    DataActivityModel(
+                        hasActivityIn = true,
+                        hasActivityOut = true,
+                    )
+                )
+
+            displayInfoJob.cancel()
+            job.cancel()
+        }
+
+    @Test
+    fun dataConnectionStateListener_noisyActivity() =
+        testScope.runTest {
+            var latest: DataConnectionState? = null
+
+            // Start collecting data activity; don't care about the result
+            val activityJob = underTest.dataActivityDirection.launchIn(this)
+
+            val connectionCallback = getTelephonyCallbackForType<DataConnectionStateListener>()
+            val activityCallback = getTelephonyCallbackForType<DataActivityListener>()
+
+            connectionCallback.onDataConnectionStateChanged(
+                TelephonyManager.DATA_CONNECTED,
+                200 /* unused */
+            )
+
+            // Send a bunch of events that we don't care about, to overrun the replay buffer
+            flipActivity(100, activityCallback)
+
+            val connectionJob = underTest.dataConnectionState.onEach { latest = it }.launchIn(this)
+
+            assertThat(latest).isEqualTo(DataConnectionState.Connected)
+
+            activityJob.cancel()
+            connectionJob.cancel()
+        }
+
+    @Test
+    fun dataEnabledLate_noisyActivity() =
+        testScope.runTest {
+            var latest: Boolean? = null
+
+            // Start collecting data activity; don't care about the result
+            val activityJob = underTest.dataActivityDirection.launchIn(this)
+
+            val enabledCallback = getTelephonyCallbackForType<DataEnabledListener>()
+            val activityCallback = getTelephonyCallbackForType<DataActivityListener>()
+
+            enabledCallback.onDataEnabledChanged(true, 1 /* unused */)
+
+            // Send a bunch of events that we don't care about, to overrun the replay buffer
+            flipActivity(100, activityCallback)
+
+            val job = underTest.dataEnabled.onEach { latest = it }.launchIn(this)
+
+            assertThat(latest).isTrue()
+
+            activityJob.cancel()
+            job.cancel()
+        }
+
+    @Test
+    fun displayInfoLate_noisyActivity() =
+        testScope.runTest {
+            var latest: ResolvedNetworkType? = null
+
+            // Start collecting data activity; don't care about the result
+            val activityJob = underTest.dataActivityDirection.launchIn(this)
+
+            val displayInfoCallback = getTelephonyCallbackForType<DisplayInfoListener>()
+            val activityCallback = getTelephonyCallbackForType<DataActivityListener>()
+
+            val type = NETWORK_TYPE_LTE
+            val expected = ResolvedNetworkType.DefaultNetworkType(mobileMappings.toIconKey(type))
+            val ti = mock<TelephonyDisplayInfo>().also { whenever(it.networkType).thenReturn(type) }
+            displayInfoCallback.onDisplayInfoChanged(ti)
+
+            // Send a bunch of events that we don't care about, to overrun the replay buffer
+            flipActivity(100, activityCallback)
+
+            val job = underTest.resolvedNetworkType.onEach { latest = it }.launchIn(this)
+
+            assertThat(latest).isEqualTo(expected)
+
+            activityJob.cancel()
+            job.cancel()
+        }
+
+    @Test
+    fun serviceStateListener_noisyActivity() =
+        testScope.runTest {
+            var latest: Boolean? = null
+
+            // Start collecting data activity; don't care about the result
+            val activityJob = underTest.dataActivityDirection.launchIn(this)
+
+            val serviceStateCallback = getTelephonyCallbackForType<ServiceStateListener>()
+            val activityCallback = getTelephonyCallbackForType<DataActivityListener>()
+
+            // isEmergencyOnly comes in
+            val serviceState = ServiceState()
+            serviceState.isEmergencyOnly = true
+            serviceStateCallback.onServiceStateChanged(serviceState)
+
+            flipActivity(100, activityCallback)
+
+            val job = underTest.isEmergencyOnly.onEach { latest = it }.launchIn(this)
+
+            assertThat(latest).isTrue()
+
+            activityJob.cancel()
+            job.cancel()
+        }
+
+    @Test
+    fun signalStrengthsListenerLate_noisyActivity() =
+        testScope.runTest {
+            var latest: Int? = null
+
+            // Start collecting data activity; don't care about the result
+            val activityJob = underTest.dataActivityDirection.launchIn(this)
+            val activityCallback = getTelephonyCallbackForType<DataActivityListener>()
+
+            val callback = getTelephonyCallbackForType<TelephonyCallback.SignalStrengthsListener>()
+            val strength = signalStrength(gsmLevel = 1, cdmaLevel = 2, isGsm = true)
+            callback.onSignalStrengthsChanged(strength)
+
+            flipActivity(100, activityCallback)
+
+            val job = underTest.cdmaLevel.onEach { latest = it }.launchIn(this)
+
+            assertThat(latest).isEqualTo(2)
+
+            activityJob.cancel()
+            job.cancel()
+        }
+
+    private fun flipActivity(
+        times: Int,
+        callback: DataActivityListener,
+    ) {
+        repeat(times) { index -> callback.onDataActivity(index % 4) }
+    }
+
+    private fun flipDisplayInfo(
+        times: Int,
+        infos: List<TelephonyDisplayInfo>,
+        callback: DisplayInfoListener,
+    ) {
+        val len = infos.size
+        repeat(times) { index -> callback.onDisplayInfoChanged(infos[index % len]) }
+    }
+
+    private inline fun <reified T> getTelephonyCallbackForType(): T {
+        return getTelephonyCallbackForType(telephonyManager)
+    }
+
+    companion object {
+        private const val SUB_1_ID = 1
+
+        private val DEFAULT_NAME = NetworkNameModel.Default("default name")
+        private const val SEP = "-"
+    }
+}
diff --git a/packages/SystemUI/tests/src/com/android/systemui/statusbar/pipeline/mobile/data/repository/prod/MobileTelephonyHelpers.kt b/packages/SystemUI/tests/src/com/android/systemui/statusbar/pipeline/mobile/data/repository/prod/MobileTelephonyHelpers.kt
index 621f79307e49..d07b96f6609e 100644
--- a/packages/SystemUI/tests/src/com/android/systemui/statusbar/pipeline/mobile/data/repository/prod/MobileTelephonyHelpers.kt
+++ b/packages/SystemUI/tests/src/com/android/systemui/statusbar/pipeline/mobile/data/repository/prod/MobileTelephonyHelpers.kt
@@ -16,10 +16,14 @@
 
 package com.android.systemui.statusbar.pipeline.mobile.data.repository.prod
 
+import android.telephony.CellSignalStrengthCdma
+import android.telephony.SignalStrength
 import android.telephony.TelephonyCallback
 import android.telephony.TelephonyManager
 import com.android.systemui.util.mockito.any
 import com.android.systemui.util.mockito.argumentCaptor
+import com.android.systemui.util.mockito.mock
+import com.android.systemui.util.mockito.whenever
 import com.google.common.truth.Truth.assertThat
 import org.mockito.Mockito.verify
 
@@ -31,6 +35,19 @@ object MobileTelephonyHelpers {
         return callbackCaptor.allValues
     }
 
+    /** Convenience constructor for SignalStrength */
+    fun signalStrength(gsmLevel: Int, cdmaLevel: Int, isGsm: Boolean): SignalStrength {
+        val signalStrength = mock<SignalStrength>()
+        whenever(signalStrength.isGsm).thenReturn(isGsm)
+        whenever(signalStrength.level).thenReturn(gsmLevel)
+        val cdmaStrength =
+            mock<CellSignalStrengthCdma>().also { whenever(it.level).thenReturn(cdmaLevel) }
+        whenever(signalStrength.getCellSignalStrengths(CellSignalStrengthCdma::class.java))
+            .thenReturn(listOf(cdmaStrength))
+
+        return signalStrength
+    }
+
     inline fun <reified T> getTelephonyCallbackForType(mockTelephonyManager: TelephonyManager): T {
         val cbs = getTelephonyCallbacks(mockTelephonyManager).filterIsInstance<T>()
         assertThat(cbs.size).isEqualTo(1)