diff --git a/tests/cts/multidevices/Android.bp b/tests/cts/multidevices/Android.bp index 6caf4826db..5f9b3ef38c 100644 --- a/tests/cts/multidevices/Android.bp +++ b/tests/cts/multidevices/Android.bp @@ -20,7 +20,10 @@ package { 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 ab88504994..417db99711 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 @@ class ConnectivityMultiDevicesTest(base_test.BaseTestClass): 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 258648f6b1..8805edd818 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.ScanResult 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 @@ class ConnectivityMultiDevicesSnippet : Snippet { // 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 @@ class ConnectivityMultiDevicesSnippet : Snippet { 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() - if (requireValidation) { - networkCallback.eventuallyExpect { - 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 { + // 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 0000000000..a2d703ca56 --- /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()