joy/android_packages_modules_Connectivity
Watch
1
Fork
You've already forked android_packages_modules_Connectivity
 0
Code Issues Pull requests Projects Releases Wiki Activity

Merge changes Ib0df2810,Ic8dcb7e5 into main

Browse source 
* changes:
  Refactor tethering utility methods
  Fix NetworkCallback#onAvailable does not match SSID
This commit is contained in:
Junyu Lai 2024-05-31 08:41:06 +00:00 • committed by Gerrit Code Review
commit 8a14358e16
4 changed files with 153 additions and 90 deletions
Download patch file Download diff file Expand all files Collapse all files

5
tests/cts/multidevices/Android.bp
View file

@ -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",
],

86
tests/cts/multidevices/connectivity_multi_devices_test.py
View file

@ -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__":

53
tests/cts/multidevices/snippet/ConnectivityMultiDevicesSnippet.kt
View file

@ -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<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
}
}

103
tests/cts/multidevices/tether_utils.py Normal file
View file

@ -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()