/* * Copyright (C) 2020 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.networkstack.tethering import android.app.Notification import android.app.NotificationManager import android.app.PendingIntent import android.app.PendingIntent.FLAG_IMMUTABLE import android.content.Context import android.content.Intent import android.content.pm.ActivityInfo import android.content.pm.ApplicationInfo import android.content.pm.PackageManager import android.content.pm.ResolveInfo import android.content.res.Resources import android.net.ConnectivityManager.TETHERING_WIFI import android.net.NetworkCapabilities import android.net.NetworkCapabilities.NET_CAPABILITY_NOT_ROAMING import android.os.Handler import android.os.HandlerThread import android.os.Looper import android.os.UserHandle import android.provider.Settings import android.telephony.TelephonyManager import androidx.test.filters.SmallTest import androidx.test.platform.app.InstrumentationRegistry import androidx.test.runner.AndroidJUnit4 import com.android.internal.util.test.BroadcastInterceptingContext import com.android.networkstack.tethering.TetheringNotificationUpdater.ACTION_DISABLE_TETHERING import com.android.networkstack.tethering.TetheringNotificationUpdater.DOWNSTREAM_NONE import com.android.networkstack.tethering.TetheringNotificationUpdater.EVENT_SHOW_NO_UPSTREAM import com.android.networkstack.tethering.TetheringNotificationUpdater.NO_UPSTREAM_NOTIFICATION_ID import com.android.networkstack.tethering.TetheringNotificationUpdater.RESTRICTED_NOTIFICATION_ID import com.android.networkstack.tethering.TetheringNotificationUpdater.ROAMING_NOTIFICATION_ID import com.android.networkstack.tethering.TetheringNotificationUpdater.VERIZON_CARRIER_ID import com.android.networkstack.tethering.TetheringNotificationUpdater.getSettingsPackageName import com.android.testutils.waitForIdle import org.junit.After import org.junit.Assert.assertEquals import org.junit.Assert.assertNotNull import org.junit.Assert.fail import org.junit.Before import org.junit.Test import org.junit.runner.RunWith import org.mockito.ArgumentCaptor import org.mockito.ArgumentMatchers.any import org.mockito.ArgumentMatchers.anyInt import org.mockito.ArgumentMatchers.eq import org.mockito.Mock import org.mockito.Mockito.doReturn import org.mockito.Mockito.mock import org.mockito.Mockito.never import org.mockito.Mockito.reset import org.mockito.Mockito.times import org.mockito.Mockito.verify import org.mockito.Mockito.verifyZeroInteractions import org.mockito.MockitoAnnotations const val TEST_SUBID = 1 const val WIFI_MASK = 1 shl TETHERING_WIFI const val TEST_DISALLOW_TITLE = "Tether function is disallowed" const val TEST_DISALLOW_MESSAGE = "Please contact your admin" const val TEST_NO_UPSTREAM_TITLE = "Hotspot has no internet access" const val TEST_NO_UPSTREAM_MESSAGE = "Device cannot connect to internet." const val TEST_NO_UPSTREAM_BUTTON = "Turn off hotspot" const val TEST_ROAMING_TITLE = "Hotspot is on" const val TEST_ROAMING_MESSAGE = "Additional charges may apply while roaming." @RunWith(AndroidJUnit4::class) @SmallTest class TetheringNotificationUpdaterTest { // lateinit used here for mocks as they need to be reinitialized between each test and the test // should crash if they are used before being initialized. @Mock private lateinit var mockContext: Context @Mock private lateinit var notificationManager: NotificationManager @Mock private lateinit var telephonyManager: TelephonyManager @Mock private lateinit var testResources: Resources // lateinit for these classes under test, as they should be reset to a different instance for // every test but should always be initialized before use (or the test should crash). private lateinit var context: TestContext private lateinit var notificationUpdater: TetheringNotificationUpdater // Initializing the following members depends on initializing some of the mocks and // is more logically done in setup(). private lateinit var fakeTetheringThread: HandlerThread private val ROAMING_CAPABILITIES = NetworkCapabilities() private val HOME_CAPABILITIES = NetworkCapabilities().addCapability(NET_CAPABILITY_NOT_ROAMING) private val NOTIFICATION_ICON_ID = R.drawable.stat_sys_tether_general private val TIMEOUT_MS = 500L private val ACTIVITY_PENDING_INTENT = 0 private val BROADCAST_PENDING_INTENT = 1 private inner class TestContext(c: Context) : BroadcastInterceptingContext(c) { override fun createContextAsUser(user: UserHandle, flags: Int) = if (user == UserHandle.ALL) mockContext else this override fun getSystemService(name: String) = if (name == Context.TELEPHONY_SERVICE) telephonyManager else super.getSystemService(name) } private inner class WrappedNotificationUpdater(c: Context, looper: Looper) : TetheringNotificationUpdater(c, looper) { override fun getResourcesForSubId(c: Context, subId: Int) = if (subId == TEST_SUBID) testResources else super.getResourcesForSubId(c, subId) } private fun setupResources() { doReturn(5).`when`(testResources) .getInteger(R.integer.delay_to_show_no_upstream_after_no_backhaul) doReturn(true).`when`(testResources) .getBoolean(R.bool.config_upstream_roaming_notification) doReturn(TEST_DISALLOW_TITLE).`when`(testResources) .getString(R.string.disable_tether_notification_title) doReturn(TEST_DISALLOW_MESSAGE).`when`(testResources) .getString(R.string.disable_tether_notification_message) doReturn(TEST_NO_UPSTREAM_TITLE).`when`(testResources) .getString(R.string.no_upstream_notification_title) doReturn(TEST_NO_UPSTREAM_MESSAGE).`when`(testResources) .getString(R.string.no_upstream_notification_message) doReturn(TEST_NO_UPSTREAM_BUTTON).`when`(testResources) .getString(R.string.no_upstream_notification_disable_button) doReturn(TEST_ROAMING_TITLE).`when`(testResources) .getString(R.string.upstream_roaming_notification_title) doReturn(TEST_ROAMING_MESSAGE).`when`(testResources) .getString(R.string.upstream_roaming_notification_message) } @Before fun setUp() { MockitoAnnotations.initMocks(this) context = TestContext(InstrumentationRegistry.getInstrumentation().context) doReturn(notificationManager).`when`(mockContext) .getSystemService(Context.NOTIFICATION_SERVICE) fakeTetheringThread = HandlerThread(this::class.java.simpleName) fakeTetheringThread.start() notificationUpdater = WrappedNotificationUpdater(context, fakeTetheringThread.looper) setupResources() } @After fun tearDown() { fakeTetheringThread.quitSafely() } private fun verifyActivityPendingIntent(intent: Intent, flags: Int) { // Use FLAG_NO_CREATE to verify whether PendingIntent has FLAG_IMMUTABLE flag(forcefully add // the flag in creating arguments). If the described PendingIntent does not already exist, // getActivity() will return null instead of PendingIntent object. val pi = PendingIntent.getActivity( context.createContextAsUser(UserHandle.CURRENT, 0 /* flags */), 0 /* requestCode */, intent, flags or FLAG_IMMUTABLE or PendingIntent.FLAG_NO_CREATE, null /* options */) assertNotNull("Activity PendingIntent with FLAG_IMMUTABLE does not exist.", pi) } private fun verifyBroadcastPendingIntent(intent: Intent, flags: Int) { // Use FLAG_NO_CREATE to verify whether PendingIntent has FLAG_IMMUTABLE flag(forcefully add // the flag in creating arguments). If the described PendingIntent does not already exist, // getBroadcast() will return null instead of PendingIntent object. val pi = PendingIntent.getBroadcast( context.createContextAsUser(UserHandle.CURRENT, 0 /* flags */), 0 /* requestCode */, intent, flags or FLAG_IMMUTABLE or PendingIntent.FLAG_NO_CREATE) assertNotNull("Broadcast PendingIntent with FLAG_IMMUTABLE does not exist.", pi) } private fun Notification.title() = this.extras.getString(Notification.EXTRA_TITLE) private fun Notification.text() = this.extras.getString(Notification.EXTRA_TEXT) private fun verifyNotification( iconId: Int, title: String, text: String, id: Int, intentSenderType: Int, intent: Intent, flags: Int ) { verify(notificationManager, never()).cancel(any(), eq(id)) val notificationCaptor = ArgumentCaptor.forClass(Notification::class.java) verify(notificationManager, times(1)) .notify(any(), eq(id), notificationCaptor.capture()) val notification = notificationCaptor.getValue() assertEquals(iconId, notification.smallIcon.resId) assertEquals(title, notification.title()) assertEquals(text, notification.text()) when (intentSenderType) { ACTIVITY_PENDING_INTENT -> verifyActivityPendingIntent(intent, flags) BROADCAST_PENDING_INTENT -> verifyBroadcastPendingIntent(intent, flags) } reset(notificationManager) } private fun verifyNotificationCancelled( notificationIds: List, resetAfterVerified: Boolean = true ) { notificationIds.forEach { verify(notificationManager, times(1)).cancel(any(), eq(it)) } if (resetAfterVerified) reset(notificationManager) } @Test fun testRestrictedNotification() { val settingsIntent = Intent(Settings.ACTION_TETHER_SETTINGS) .setPackage(getSettingsPackageName(context.packageManager)) .addFlags(Intent.FLAG_ACTIVITY_NEW_TASK) // Set test sub id. notificationUpdater.onActiveDataSubscriptionIdChanged(TEST_SUBID) verifyNotificationCancelled(listOf(NO_UPSTREAM_NOTIFICATION_ID, ROAMING_NOTIFICATION_ID)) // User restrictions on. Show restricted notification. notificationUpdater.notifyTetheringDisabledByRestriction() verifyNotification(NOTIFICATION_ICON_ID, TEST_DISALLOW_TITLE, TEST_DISALLOW_MESSAGE, RESTRICTED_NOTIFICATION_ID, ACTIVITY_PENDING_INTENT, settingsIntent, FLAG_IMMUTABLE) // User restrictions off. Clear notification. notificationUpdater.tetheringRestrictionLifted() verifyNotificationCancelled(listOf(RESTRICTED_NOTIFICATION_ID)) // No downstream. notificationUpdater.onDownstreamChanged(DOWNSTREAM_NONE) verifyZeroInteractions(notificationManager) // User restrictions on again. Show restricted notification. notificationUpdater.notifyTetheringDisabledByRestriction() verifyNotification(NOTIFICATION_ICON_ID, TEST_DISALLOW_TITLE, TEST_DISALLOW_MESSAGE, RESTRICTED_NOTIFICATION_ID, ACTIVITY_PENDING_INTENT, settingsIntent, FLAG_IMMUTABLE) } val MAX_BACKOFF_MS = 200L /** * Waits for all messages, including delayed ones, to be processed. * * This will wait until the handler has no more messages to be processed including * delayed ones, or the timeout has expired. It uses an exponential backoff strategy * to wait longer and longer to consume less CPU, with the max granularity being * MAX_BACKOFF_MS. * * @return true if all messages have been processed including delayed ones, false if timeout * * TODO: Move this method to com.android.testutils.HandlerUtils.kt. */ private fun Handler.waitForDelayedMessage(what: Int?, timeoutMs: Long) { fun hasMatchingMessages() = if (what == null) hasMessagesOrCallbacks() else hasMessages(what) val expiry = System.currentTimeMillis() + timeoutMs var delay = 5L while (System.currentTimeMillis() < expiry && hasMatchingMessages()) { // None of Handler, Looper, Message and MessageQueue expose any way to retrieve // the time when the next (let alone the last) message will be processed, so // short of examining the internals with reflection sleep() is the only solution. Thread.sleep(delay) delay = (delay * 2) .coerceAtMost(expiry - System.currentTimeMillis()) .coerceAtMost(MAX_BACKOFF_MS) } val timeout = expiry - System.currentTimeMillis() if (timeout <= 0) fail("Delayed message did not process yet after ${timeoutMs}ms") waitForIdle(timeout) } @Test fun testNoUpstreamNotification() { val disableIntent = Intent(ACTION_DISABLE_TETHERING).setPackage(context.packageName) // Set test sub id. notificationUpdater.onActiveDataSubscriptionIdChanged(TEST_SUBID) verifyNotificationCancelled(listOf(NO_UPSTREAM_NOTIFICATION_ID, ROAMING_NOTIFICATION_ID)) // Wifi downstream. notificationUpdater.onDownstreamChanged(WIFI_MASK) verifyNotificationCancelled(listOf(NO_UPSTREAM_NOTIFICATION_ID, ROAMING_NOTIFICATION_ID)) // There is no upstream. Show no upstream notification. notificationUpdater.onUpstreamCapabilitiesChanged(null) notificationUpdater.handler.waitForDelayedMessage(EVENT_SHOW_NO_UPSTREAM, TIMEOUT_MS) verifyNotification(NOTIFICATION_ICON_ID, TEST_NO_UPSTREAM_TITLE, TEST_NO_UPSTREAM_MESSAGE, NO_UPSTREAM_NOTIFICATION_ID, BROADCAST_PENDING_INTENT, disableIntent, FLAG_IMMUTABLE) // Same capabilities changed. Nothing happened. notificationUpdater.onUpstreamCapabilitiesChanged(null) verifyZeroInteractions(notificationManager) // Upstream come back. Clear no upstream notification. notificationUpdater.onUpstreamCapabilitiesChanged(HOME_CAPABILITIES) verifyNotificationCancelled(listOf(NO_UPSTREAM_NOTIFICATION_ID)) // No upstream again. Show no upstream notification. notificationUpdater.onUpstreamCapabilitiesChanged(null) notificationUpdater.handler.waitForDelayedMessage(EVENT_SHOW_NO_UPSTREAM, TIMEOUT_MS) verifyNotification(NOTIFICATION_ICON_ID, TEST_NO_UPSTREAM_TITLE, TEST_NO_UPSTREAM_MESSAGE, NO_UPSTREAM_NOTIFICATION_ID, BROADCAST_PENDING_INTENT, disableIntent, FLAG_IMMUTABLE) // No downstream. notificationUpdater.onDownstreamChanged(DOWNSTREAM_NONE) verifyNotificationCancelled(listOf(NO_UPSTREAM_NOTIFICATION_ID, ROAMING_NOTIFICATION_ID)) // Wifi downstream and home capabilities. notificationUpdater.onDownstreamChanged(WIFI_MASK) notificationUpdater.onUpstreamCapabilitiesChanged(HOME_CAPABILITIES) verifyNotificationCancelled(listOf(NO_UPSTREAM_NOTIFICATION_ID, ROAMING_NOTIFICATION_ID)) // Set R.integer.delay_to_show_no_upstream_after_no_backhaul to -1 and change to no upstream // again. Don't put up no upstream notification. doReturn(-1).`when`(testResources) .getInteger(R.integer.delay_to_show_no_upstream_after_no_backhaul) notificationUpdater.onUpstreamCapabilitiesChanged(null) notificationUpdater.handler.waitForDelayedMessage(EVENT_SHOW_NO_UPSTREAM, TIMEOUT_MS) verifyNotificationCancelled(listOf(NO_UPSTREAM_NOTIFICATION_ID)) } @Test fun testGetResourcesForSubId() { doReturn(telephonyManager).`when`(telephonyManager).createForSubscriptionId(anyInt()) doReturn(1234).`when`(telephonyManager).getSimCarrierId() doReturn("000000").`when`(telephonyManager).getSimOperator() val subId = -2 // Use invalid subId to avoid getting resource from cache or real subId. val config = context.resources.configuration var res = notificationUpdater.getResourcesForSubId(context, subId) assertEquals(config.mcc, res.configuration.mcc) assertEquals(config.mnc, res.configuration.mnc) doReturn(VERIZON_CARRIER_ID).`when`(telephonyManager).getSimCarrierId() res = notificationUpdater.getResourcesForSubId(context, subId) assertEquals(config.mcc, res.configuration.mcc) assertEquals(config.mnc, res.configuration.mnc) doReturn("20404").`when`(telephonyManager).getSimOperator() res = notificationUpdater.getResourcesForSubId(context, subId) assertEquals(311, res.configuration.mcc) assertEquals(480, res.configuration.mnc) } @Test fun testRoamingNotification() { val disableIntent = Intent(ACTION_DISABLE_TETHERING).setPackage(context.packageName) val settingsIntent = Intent(Settings.ACTION_TETHER_SETTINGS) .setPackage(getSettingsPackageName(context.packageManager)) .addFlags(Intent.FLAG_ACTIVITY_NEW_TASK) // Set test sub id. notificationUpdater.onActiveDataSubscriptionIdChanged(TEST_SUBID) verifyNotificationCancelled(listOf(NO_UPSTREAM_NOTIFICATION_ID, ROAMING_NOTIFICATION_ID)) // Wifi downstream. notificationUpdater.onDownstreamChanged(WIFI_MASK) verifyNotificationCancelled(listOf(NO_UPSTREAM_NOTIFICATION_ID, ROAMING_NOTIFICATION_ID)) // Upstream capabilities changed to roaming state. Show roaming notification. notificationUpdater.onUpstreamCapabilitiesChanged(ROAMING_CAPABILITIES) verifyNotification(NOTIFICATION_ICON_ID, TEST_ROAMING_TITLE, TEST_ROAMING_MESSAGE, ROAMING_NOTIFICATION_ID, ACTIVITY_PENDING_INTENT, settingsIntent, FLAG_IMMUTABLE) // Same capabilities change. Nothing happened. notificationUpdater.onUpstreamCapabilitiesChanged(ROAMING_CAPABILITIES) verifyZeroInteractions(notificationManager) // Upstream capabilities changed to home state. Clear roaming notification. notificationUpdater.onUpstreamCapabilitiesChanged(HOME_CAPABILITIES) verifyNotificationCancelled(listOf(ROAMING_NOTIFICATION_ID)) // Upstream capabilities changed to roaming state again. Show roaming notification. notificationUpdater.onUpstreamCapabilitiesChanged(ROAMING_CAPABILITIES) verifyNotification(NOTIFICATION_ICON_ID, TEST_ROAMING_TITLE, TEST_ROAMING_MESSAGE, ROAMING_NOTIFICATION_ID, ACTIVITY_PENDING_INTENT, settingsIntent, FLAG_IMMUTABLE) // No upstream. Clear roaming notification and show no upstream notification. notificationUpdater.onUpstreamCapabilitiesChanged(null) notificationUpdater.handler.waitForDelayedMessage(EVENT_SHOW_NO_UPSTREAM, TIMEOUT_MS) verifyNotificationCancelled(listOf(ROAMING_NOTIFICATION_ID), false) verifyNotification(NOTIFICATION_ICON_ID, TEST_NO_UPSTREAM_TITLE, TEST_NO_UPSTREAM_MESSAGE, NO_UPSTREAM_NOTIFICATION_ID, BROADCAST_PENDING_INTENT, disableIntent, FLAG_IMMUTABLE) // No downstream. notificationUpdater.onDownstreamChanged(DOWNSTREAM_NONE) verifyNotificationCancelled(listOf(NO_UPSTREAM_NOTIFICATION_ID, ROAMING_NOTIFICATION_ID)) // Wifi downstream again. notificationUpdater.onDownstreamChanged(WIFI_MASK) notificationUpdater.handler.waitForDelayedMessage(EVENT_SHOW_NO_UPSTREAM, TIMEOUT_MS) verifyNotificationCancelled(listOf(ROAMING_NOTIFICATION_ID), false) verifyNotification(NOTIFICATION_ICON_ID, TEST_NO_UPSTREAM_TITLE, TEST_NO_UPSTREAM_MESSAGE, NO_UPSTREAM_NOTIFICATION_ID, BROADCAST_PENDING_INTENT, disableIntent, FLAG_IMMUTABLE) // Set R.bool.config_upstream_roaming_notification to false and change upstream // network to roaming state again. No roaming notification. doReturn(false).`when`(testResources) .getBoolean(R.bool.config_upstream_roaming_notification) notificationUpdater.onUpstreamCapabilitiesChanged(ROAMING_CAPABILITIES) verifyNotificationCancelled(listOf(NO_UPSTREAM_NOTIFICATION_ID, ROAMING_NOTIFICATION_ID)) } @Test fun testGetSettingsPackageName() { val defaultSettingsPackageName = "com.android.settings" val testSettingsPackageName = "com.android.test.settings" val pm = mock(PackageManager::class.java) doReturn(null).`when`(pm).resolveActivity(any(), anyInt()) assertEquals(defaultSettingsPackageName, getSettingsPackageName(pm)) val resolveInfo = ResolveInfo().apply { activityInfo = ActivityInfo().apply { name = "test" applicationInfo = ApplicationInfo().apply { packageName = testSettingsPackageName } } } doReturn(resolveInfo).`when`(pm).resolveActivity(any(), anyInt()) assertEquals(testSettingsPackageName, getSettingsPackageName(pm)) } }