Merge changes Ib0df2810,Ic8dcb7e5 into main am: 8a14358e16
Original change: https://android-review.googlesource.com/c/platform/packages/modules/Connectivity/+/3057526
Change-Id: I543c518e1b7c33969788020500a23098cbf9e87b
Signed-off-by: Automerger Merge Worker <android-build-automerger-merge-worker@system.gserviceaccount.com>
diff --git a/tests/cts/multidevices/Android.bp b/tests/cts/multidevices/Android.bp
index 6caf482..5f9b3ef 100644
--- a/tests/cts/multidevices/Android.bp
+++ b/tests/cts/multidevices/Android.bp
@@ -20,7 +20,10 @@
python_test_host {
name: "CtsConnectivityMultiDevicesTestCases",
main: "connectivity_multi_devices_test.py",
- srcs: ["connectivity_multi_devices_test.py"],
+ srcs: [
+ "connectivity_multi_devices_test.py",
+ "tether_utils.py",
+ ],
libs: [
"mobly",
],
diff --git a/tests/cts/multidevices/connectivity_multi_devices_test.py b/tests/cts/multidevices/connectivity_multi_devices_test.py
index ab88504..417db99 100644
--- a/tests/cts/multidevices/connectivity_multi_devices_test.py
+++ b/tests/cts/multidevices/connectivity_multi_devices_test.py
@@ -1,23 +1,16 @@
# Lint as: python3
"""Connectivity multi devices tests."""
-import base64
import sys
-import uuid
-
-from mobly import asserts
from mobly import base_test
from mobly import test_runner
from mobly import utils
from mobly.controllers import android_device
+import tether_utils
+from tether_utils import UpstreamType
CONNECTIVITY_MULTI_DEVICES_SNIPPET_PACKAGE = "com.google.snippet.connectivity"
-class UpstreamType:
- CELLULAR = 1
- WIFI = 2
-
-
class ConnectivityMultiDevicesTest(base_test.BaseTestClass):
def setup_class(self):
@@ -40,66 +33,27 @@
raise_on_exception=True,
)
- @staticmethod
- def generate_uuid32_base64():
- """Generates a UUID32 and encodes it in Base64.
-
- Returns:
- str: The Base64-encoded UUID32 string. Which is 22 characters.
- """
- return base64.b64encode(uuid.uuid1().bytes).decode("utf-8").strip("=")
-
- def _do_test_hotspot_for_upstream_type(self, upstream_type):
- """Test hotspot with the specified upstream type.
-
- This test create a hotspot, make the client connect
- to it, and verify the packet is forwarded by the hotspot.
- """
- server = self.serverDevice.connectivity_multi_devices_snippet
- client = self.clientDevice.connectivity_multi_devices_snippet
-
- # Assert pre-conditions specific to each upstream type.
- asserts.skip_if(not client.hasWifiFeature(), "Client requires Wifi feature")
- asserts.skip_if(
- not server.hasHotspotFeature(), "Server requires hotspot feature"
- )
- if upstream_type == UpstreamType.CELLULAR:
- asserts.skip_if(
- not server.hasTelephonyFeature(), "Server requires Telephony feature"
- )
- server.requestCellularAndEnsureDefault()
- elif upstream_type == UpstreamType.WIFI:
- asserts.skip_if(
- not server.isStaApConcurrencySupported(),
- "Server requires Wifi AP + STA concurrency",
- )
- server.ensureWifiIsDefault()
- else:
- raise ValueError(f"Invalid upstream type: {upstream_type}")
-
- # Generate ssid/passphrase with random characters to make sure nearby devices won't
- # connect unexpectedly. Note that total length of ssid cannot go over 32.
- testSsid = "HOTSPOT-" + self.generate_uuid32_base64()
- testPassphrase = self.generate_uuid32_base64()
-
- try:
- # Create a hotspot with fixed SSID and password.
- server.startHotspot(testSsid, testPassphrase)
-
- # Make the client connects to the hotspot.
- client.connectToWifi(testSsid, testPassphrase, True)
-
- finally:
- if upstream_type == UpstreamType.CELLULAR:
- server.unrequestCellular()
- # Teardown the hotspot.
- server.stopAllTethering()
-
def test_hotspot_upstream_wifi(self):
- self._do_test_hotspot_for_upstream_type(UpstreamType.WIFI)
+ try:
+ # Connectivity of the client verified by asserting the validated capability.
+ tether_utils.setup_hotspot_and_client_for_upstream_type(
+ self.serverDevice, self.clientDevice, UpstreamType.WIFI
+ )
+ finally:
+ tether_utils.cleanup_tethering_for_upstream_type(
+ self.serverDevice, UpstreamType.WIFI
+ )
def test_hotspot_upstream_cellular(self):
- self._do_test_hotspot_for_upstream_type(UpstreamType.CELLULAR)
+ try:
+ # Connectivity of the client verified by asserting the validated capability.
+ tether_utils.setup_hotspot_and_client_for_upstream_type(
+ self.serverDevice, self.clientDevice, UpstreamType.CELLULAR
+ )
+ finally:
+ tether_utils.cleanup_tethering_for_upstream_type(
+ self.serverDevice, UpstreamType.CELLULAR
+ )
if __name__ == "__main__":
diff --git a/tests/cts/multidevices/snippet/ConnectivityMultiDevicesSnippet.kt b/tests/cts/multidevices/snippet/ConnectivityMultiDevicesSnippet.kt
index 258648f..8805edd 100644
--- a/tests/cts/multidevices/snippet/ConnectivityMultiDevicesSnippet.kt
+++ b/tests/cts/multidevices/snippet/ConnectivityMultiDevicesSnippet.kt
@@ -16,11 +16,11 @@
package com.google.snippet.connectivity
+import android.Manifest.permission.NETWORK_SETTINGS
import android.Manifest.permission.OVERRIDE_WIFI_CONFIG
import android.content.pm.PackageManager.FEATURE_TELEPHONY
import android.content.pm.PackageManager.FEATURE_WIFI
import android.net.ConnectivityManager
-import android.net.Network
import android.net.NetworkCapabilities.NET_CAPABILITY_VALIDATED
import android.net.NetworkCapabilities.TRANSPORT_WIFI
import android.net.NetworkRequest
@@ -30,20 +30,24 @@
import android.net.wifi.SoftApConfiguration
import android.net.wifi.SoftApConfiguration.SECURITY_TYPE_WPA2_PSK
import android.net.wifi.WifiConfiguration
+import android.net.wifi.WifiInfo
import android.net.wifi.WifiManager
import android.net.wifi.WifiNetworkSpecifier
import android.net.wifi.WifiSsid
import androidx.test.platform.app.InstrumentationRegistry
+import com.android.testutils.AutoReleaseNetworkCallbackRule
import com.android.testutils.ConnectUtil
import com.android.testutils.NetworkCallbackHelper
-import com.android.testutils.RecorderCallback.CallbackEntry.Available
import com.android.testutils.RecorderCallback.CallbackEntry.CapabilitiesChanged
import com.android.testutils.TestableNetworkCallback
import com.android.testutils.runAsShell
import com.google.android.mobly.snippet.Snippet
import com.google.android.mobly.snippet.rpc.Rpc
+import org.junit.Rule
class ConnectivityMultiDevicesSnippet : Snippet {
+ @get:Rule
+ val networkCallbackRule = AutoReleaseNetworkCallbackRule()
private val context = InstrumentationRegistry.getInstrumentation().getTargetContext()
private val wifiManager = context.getSystemService(WifiManager::class.java)!!
private val cm = context.getSystemService(ConnectivityManager::class.java)!!
@@ -88,10 +92,8 @@
// Suppress warning because WifiManager methods to connect to a config are
// documented not to be deprecated for privileged users.
@Suppress("DEPRECATION")
- fun connectToWifi(ssid: String, passphrase: String, requireValidation: Boolean): Network {
+ fun connectToWifi(ssid: String, passphrase: String): Long {
val specifier = WifiNetworkSpecifier.Builder()
- .setSsid(ssid)
- .setWpa2Passphrase(passphrase)
.setBand(ScanResult.WIFI_BAND_24_GHZ)
.build()
val wifiConfig = WifiConfiguration()
@@ -102,27 +104,28 @@
wifiConfig.allowedPairwiseCiphers.set(WifiConfiguration.PairwiseCipher.TKIP)
wifiConfig.allowedPairwiseCiphers.set(WifiConfiguration.PairwiseCipher.CCMP)
- // Register network callback for the specific wifi.
+ // Add the test configuration and connect to it.
+ val connectUtil = ConnectUtil(context)
+ connectUtil.connectToWifiConfig(wifiConfig)
+
+ // Implement manual SSID matching. Specifying the SSID in
+ // NetworkSpecifier is ineffective
+ // (see WifiNetworkAgentSpecifier#canBeSatisfiedBy for details).
+ // Note that holding permission is necessary when waiting for
+ // the callbacks. The handler thread checks permission; if
+ // it's not present, the SSID will be redacted.
val networkCallback = TestableNetworkCallback()
- val wifiRequest = NetworkRequest.Builder().addTransportType(TRANSPORT_WIFI)
- .setNetworkSpecifier(specifier)
- .build()
- cm.registerNetworkCallback(wifiRequest, networkCallback)
-
- try {
- // Add the test configuration and connect to it.
- val connectUtil = ConnectUtil(context)
- connectUtil.connectToWifiConfig(wifiConfig)
-
- val event = networkCallback.expect<Available>()
- if (requireValidation) {
- networkCallback.eventuallyExpect<CapabilitiesChanged> {
- it.caps.hasCapability(NET_CAPABILITY_VALIDATED)
- }
- }
- return event.network
- } finally {
- cm.unregisterNetworkCallback(networkCallback)
+ val wifiRequest = NetworkRequest.Builder().addTransportType(TRANSPORT_WIFI).build()
+ return runAsShell(NETWORK_SETTINGS) {
+ // Register the network callback is needed here.
+ // This is to avoid the race condition where callback is fired before
+ // acquiring permission.
+ networkCallbackRule.registerNetworkCallback(wifiRequest, networkCallback)
+ return@runAsShell networkCallback.eventuallyExpect<CapabilitiesChanged> {
+ // Remove double quotes.
+ val ssidFromCaps = (WifiInfo::sanitizeSsid)(it.caps.ssid)
+ ssidFromCaps == ssid && it.caps.hasCapability(NET_CAPABILITY_VALIDATED)
+ }.network.networkHandle
}
}
diff --git a/tests/cts/multidevices/tether_utils.py b/tests/cts/multidevices/tether_utils.py
new file mode 100644
index 0000000..a2d703c
--- /dev/null
+++ b/tests/cts/multidevices/tether_utils.py
@@ -0,0 +1,103 @@
+# Copyright (C) 2024 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.
+#
+# 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.
+
+import base64
+import uuid
+
+from mobly import asserts
+from mobly.controllers import android_device
+
+
+class UpstreamType:
+ CELLULAR = 1
+ WIFI = 2
+
+
+def generate_uuid32_base64() -> str:
+ """Generates a UUID32 and encodes it in Base64.
+
+ Returns:
+ str: The Base64-encoded UUID32 string. Which is 22 characters.
+ """
+ # Strip padding characters to make it safer for hotspot name length limit.
+ return base64.b64encode(uuid.uuid1().bytes).decode("utf-8").strip("=")
+
+
+def setup_hotspot_and_client_for_upstream_type(
+ server_device: android_device,
+ client_device: android_device,
+ upstream_type: UpstreamType,
+) -> (str, int):
+ """Setup the hotspot with a connected client with the specified upstream type.
+
+ This creates a hotspot, make the client connect
+ to it, and verify the packet is forwarded by the hotspot.
+ And returns interface name of both if successful.
+ """
+ server = server_device.connectivity_multi_devices_snippet
+ client = client_device.connectivity_multi_devices_snippet
+
+ # Assert pre-conditions specific to each upstream type.
+ asserts.skip_if(not client.hasWifiFeature(), "Client requires Wifi feature")
+ asserts.skip_if(
+ not server.hasHotspotFeature(), "Server requires hotspot feature"
+ )
+ if upstream_type == UpstreamType.CELLULAR:
+ asserts.skip_if(
+ not server.hasTelephonyFeature(), "Server requires Telephony feature"
+ )
+ server.requestCellularAndEnsureDefault()
+ elif upstream_type == UpstreamType.WIFI:
+ asserts.skip_if(
+ not server.isStaApConcurrencySupported(),
+ "Server requires Wifi AP + STA concurrency",
+ )
+ server.ensureWifiIsDefault()
+ else:
+ raise ValueError(f"Invalid upstream type: {upstream_type}")
+
+ # Generate ssid/passphrase with random characters to make sure nearby devices won't
+ # connect unexpectedly. Note that total length of ssid cannot go over 32.
+ test_ssid = "HOTSPOT-" + generate_uuid32_base64()
+ test_passphrase = generate_uuid32_base64()
+
+ # Create a hotspot with fixed SSID and password.
+ hotspot_interface = server.startHotspot(test_ssid, test_passphrase)
+
+ # Make the client connects to the hotspot.
+ client_network = client.connectToWifi(test_ssid, test_passphrase)
+
+ return hotspot_interface, client_network
+
+
+def cleanup_tethering_for_upstream_type(
+ server_device: android_device, upstream_type: UpstreamType
+) -> None:
+ server = server_device.connectivity_multi_devices_snippet
+ if upstream_type == UpstreamType.CELLULAR:
+ server.unrequestCellular()
+ # Teardown the hotspot.
+ server.stopAllTethering()