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

Format multidevice tests with pyformat

Browse source 
Use the following commands to apply pyformat
pyformat -i -r --force_quote_type single staticlibs/
pyformat -i -r --force_quote_type single tests/

Flag: EXEMPT mainline
Bug: 419423721
Test: TH
Change-Id: I0d3914ec652664ed4ec5d44afc25d97ff8bb4626
This commit is contained in:
Jimi Chen 2025-06-02 06:31:34 +00:00
commit ead97d1ef3
8 changed files with 77 additions and 74 deletions
Download patch file Download diff file Expand all files Collapse all files

50
staticlibs/tests/unit/host/python/adb_utils_test.py
View file

@ -29,12 +29,12 @@ class TestAdbUtils(base_test.BaseTestClass, parameterized.TestCase):
def setup_test(self):
self.mock_ad = MagicMock() # Mock Android device object
self.mock_ad.log = MagicMock()
self.mock_ad.adb.shell.return_value = b"" # Default empty return for shell
self.mock_ad.adb.shell.return_value = b'' # Default empty return for shell
@patch(
"net_tests_utils.host.python.adb_utils.expect_dumpsys_state_with_retry"
'net_tests_utils.host.python.adb_utils.expect_dumpsys_state_with_retry'
)
@patch("net_tests_utils.host.python.adb_utils._set_screen_state")
@patch('net_tests_utils.host.python.adb_utils._set_screen_state')
def test_set_doze_mode_enable(
self, mock_set_screen_state, mock_expect_dumpsys_state
):
@ -42,29 +42,29 @@ class TestAdbUtils(base_test.BaseTestClass, parameterized.TestCase):
mock_set_screen_state.assert_called_once_with(self.mock_ad, False)
@patch(
"net_tests_utils.host.python.adb_utils.expect_dumpsys_state_with_retry"
'net_tests_utils.host.python.adb_utils.expect_dumpsys_state_with_retry'
)
def test_set_doze_mode_disable(self, mock_expect_dumpsys_state):
adb_utils.set_doze_mode(self.mock_ad, False)
@patch("net_tests_utils.host.python.adb_utils._get_screen_state")
@patch('net_tests_utils.host.python.adb_utils._get_screen_state')
def test_set_screen_state_success(self, mock_get_screen_state):
mock_get_screen_state.side_effect = [False, True] # Simulate toggle
adb_utils._set_screen_state(self.mock_ad, True)
@patch("net_tests_utils.host.python.adb_utils._get_screen_state")
@patch('net_tests_utils.host.python.adb_utils._get_screen_state')
def test_set_screen_state_failure(self, mock_get_screen_state):
mock_get_screen_state.return_value = False # State doesn't change
with asserts.assert_raises(UnexpectedBehaviorError):
adb_utils._set_screen_state(self.mock_ad, True)
@parameterized.parameters(
("Awake", True),
("Asleep", False),
("Dozing", False),
("SomeOtherState", False),
('Awake', True),
('Asleep', False),
('Dozing', False),
('SomeOtherState', False),
) # Declare inputs for state_str and expected_result.
@patch("net_tests_utils.host.python.adb_utils.get_value_of_key_from_dumpsys")
@patch('net_tests_utils.host.python.adb_utils.get_value_of_key_from_dumpsys')
def test_get_screen_state(self, state_str, expected_result, mock_get_value):
mock_get_value.return_value = state_str
asserts.assert_equal(
@ -73,44 +73,44 @@ class TestAdbUtils(base_test.BaseTestClass, parameterized.TestCase):
def test_get_value_of_key_from_dumpsys(self):
self.mock_ad.adb.shell.return_value = (
b"mWakefulness=Awake\nmOtherKey=SomeValue"
b'mWakefulness=Awake\nmOtherKey=SomeValue'
)
result = adb_utils.get_value_of_key_from_dumpsys(
self.mock_ad, "power", "mWakefulness"
self.mock_ad, 'power', 'mWakefulness'
)
asserts.assert_equal(result, "Awake")
asserts.assert_equal(result, 'Awake')
@parameterized.parameters(
(True, ["true"]),
(False, ["false"]),
(True, ['true']),
(False, ['false']),
(
True,
["false", "true"],
['false', 'true'],
), # Expect True, get False which is unexpected, then get True
(
False,
["true", "false"],
['true', 'false'],
), # Expect False, get True which is unexpected, then get False
) # Declare inputs for expected_state and returned_value
@patch("net_tests_utils.host.python.adb_utils.get_value_of_key_from_dumpsys")
@patch('net_tests_utils.host.python.adb_utils.get_value_of_key_from_dumpsys')
def test_expect_dumpsys_state_with_retry_success(
self, expected_state, returned_value, mock_get_value
):
mock_get_value.side_effect = returned_value
# Verify the method returns and does not throw.
adb_utils.expect_dumpsys_state_with_retry(
self.mock_ad, "service", "key", expected_state, 0
self.mock_ad, 'service', 'key', expected_state, 0
)
@patch("net_tests_utils.host.python.adb_utils.get_value_of_key_from_dumpsys")
@patch('net_tests_utils.host.python.adb_utils.get_value_of_key_from_dumpsys')
def test_expect_dumpsys_state_with_retry_failure(self, mock_get_value):
mock_get_value.return_value = "false"
mock_get_value.return_value = 'false'
with asserts.assert_raises(UnexpectedBehaviorError):
adb_utils.expect_dumpsys_state_with_retry(
self.mock_ad, "service", "key", True, 0
self.mock_ad, 'service', 'key', True, 0
)
@patch("net_tests_utils.host.python.adb_utils.get_value_of_key_from_dumpsys")
@patch('net_tests_utils.host.python.adb_utils.get_value_of_key_from_dumpsys')
def test_expect_dumpsys_state_with_retry_not_found(self, mock_get_value):
# Simulate the get_value_of_key_from_dumpsys cannot find the give key.
mock_get_value.return_value = None
@ -118,5 +118,5 @@ class TestAdbUtils(base_test.BaseTestClass, parameterized.TestCase):
# Expect the function to raise UnexpectedBehaviorError due to the exception
with asserts.assert_raises(UnexpectedBehaviorError):
adb_utils.expect_dumpsys_state_with_retry(
self.mock_ad, "service", "key", True
self.mock_ad, 'service', 'key', True
)

31
staticlibs/tests/unit/host/python/assert_utils_test.py
View file

@ -15,7 +15,10 @@
from mobly import asserts
from mobly import base_test
from net_tests_utils.host.python.assert_utils import (
UnexpectedBehaviorError, UnexpectedExceptionError, expect_with_retry, expect_throws
UnexpectedBehaviorError,
UnexpectedExceptionError,
expect_throws,
expect_with_retry,
)
@ -56,7 +59,7 @@ class TestAssertUtils(base_test.BaseTestClass):
retry_interval_sec=0,
)
asserts.assert_false(
retry_action_called, "retry_action called."
retry_action_called, 'retry_action called.'
) # Assert retry_action was NOT called
def test_retry_action_not_called_failed(self):
@ -75,7 +78,7 @@ class TestAssertUtils(base_test.BaseTestClass):
retry_interval_sec=0,
)
asserts.assert_false(
retry_action_called, "retry_action called."
retry_action_called, 'retry_action called.'
) # Assert retry_action was NOT called
def test_retry_action_called(self):
@ -93,23 +96,23 @@ class TestAssertUtils(base_test.BaseTestClass):
max_retries=2,
retry_interval_sec=0,
)
asserts.assert_true(retry_action_called, "retry_action not called.")
asserts.assert_true(retry_action_called, 'retry_action not called.')
def test_expect_exception_throws(self):
def raise_unexpected_behavior_error():
raise UnexpectedBehaviorError()
def raise_unexpected_behavior_error():
raise UnexpectedBehaviorError()
expect_throws(raise_unexpected_behavior_error, UnexpectedBehaviorError)
expect_throws(raise_unexpected_behavior_error, UnexpectedBehaviorError)
def test_unexpect_exception_throws(self):
def raise_value_error():
raise ValueError()
def raise_value_error():
raise ValueError()
with asserts.assert_raises(UnexpectedExceptionError):
expect_throws(raise_value_error, UnexpectedBehaviorError)
with asserts.assert_raises(UnexpectedExceptionError):
expect_throws(raise_value_error, UnexpectedBehaviorError)
def test_no_exception_throws(self):
def raise_no_error():
return
def raise_no_error():
return
expect_throws(raise_no_error, UnexpectedBehaviorError)
expect_throws(raise_no_error, UnexpectedBehaviorError)

2
staticlibs/tests/unit/host/python/run_tests.py
View file

@ -21,7 +21,7 @@ from host.python.assert_utils_test import TestAssertUtils
from mobly import suite_runner
if __name__ == "__main__":
if __name__ == '__main__':
# For MoblyBinaryHostTest, this entry point will be called twice:
# 1. List tests.
# <mobly-par-file-name> -- --list_tests

34
staticlibs/testutils/host/python/adb_utils.py
View file

@ -16,32 +16,32 @@ import re
from mobly.controllers import android_device
from net_tests_utils.host.python import assert_utils
BYTE_DECODE_UTF_8 = "utf-8"
BYTE_DECODE_UTF_8 = 'utf-8'
def set_doze_mode(ad: android_device.AndroidDevice, enable: bool) -> None:
if enable:
adb_shell(ad, "cmd battery unplug")
adb_shell(ad, 'cmd battery unplug')
expect_dumpsys_state_with_retry(
ad, "deviceidle", key="mCharging", expected_state=False
ad, 'deviceidle', key='mCharging', expected_state=False
)
_set_screen_state(ad, False)
adb_shell(ad, "dumpsys deviceidle enable deep")
adb_shell(ad, 'dumpsys deviceidle enable deep')
expect_dumpsys_state_with_retry(
ad, "deviceidle", key="mDeepEnabled", expected_state=True
ad, 'deviceidle', key='mDeepEnabled', expected_state=True
)
adb_shell(ad, "dumpsys deviceidle force-idle deep")
adb_shell(ad, 'dumpsys deviceidle force-idle deep')
expect_dumpsys_state_with_retry(
ad, "deviceidle", key="mForceIdle", expected_state=True
ad, 'deviceidle', key='mForceIdle', expected_state=True
)
else:
adb_shell(ad, "cmd battery reset")
adb_shell(ad, 'cmd battery reset')
expect_dumpsys_state_with_retry(
ad, "deviceidle", key="mCharging", expected_state=True
ad, 'deviceidle', key='mCharging', expected_state=True
)
adb_shell(ad, "dumpsys deviceidle unforce")
adb_shell(ad, 'dumpsys deviceidle unforce')
expect_dumpsys_state_with_retry(
ad, "deviceidle", key="mForceIdle", expected_state=False
ad, 'deviceidle', key='mForceIdle', expected_state=False
)
@ -51,13 +51,13 @@ def _set_screen_state(
assert_utils.expect_with_retry(
predicate=lambda: _get_screen_state(ad) == target_state,
retry_action=lambda: adb_shell(
ad, "input keyevent KEYCODE_POWER"
ad, 'input keyevent KEYCODE_POWER'
), # Toggle power key again when retry.
)
def _get_screen_state(ad: android_device.AndroidDevice) -> bool:
return get_value_of_key_from_dumpsys(ad, "power", "mWakefulness") == "Awake"
return get_value_of_key_from_dumpsys(ad, 'power', 'mWakefulness') == 'Awake'
def get_value_of_key_from_dumpsys(
@ -66,12 +66,12 @@ def get_value_of_key_from_dumpsys(
output = get_dumpsys_for_service(ad, service)
# Search for key=value pattern from the dumpsys output.
# e.g. mWakefulness=Awake
pattern = rf"{key}=(.*)"
pattern = rf'{key}=(.*)'
# Only look for the first occurrence.
match = re.search(pattern, output)
if match:
ad.log.debug(
"Getting key-value from dumpsys: " + key + "=" + match.group(1)
'Getting key-value from dumpsys: ' + key + '=' + match.group(1)
)
return match.group(1)
else:
@ -100,7 +100,7 @@ def expect_dumpsys_state_with_retry(
def get_dumpsys_for_service(
ad: android_device.AndroidDevice, service: str
) -> str:
return adb_shell(ad, "dumpsys " + service)
return adb_shell(ad, 'dumpsys ' + service)
def adb_shell(ad: android_device.AndroidDevice, shell_cmd: str) -> str:
@ -113,6 +113,6 @@ def adb_shell(ad: android_device.AndroidDevice, shell_cmd: str) -> str:
Returns:
string, replies from adb shell command.
"""
ad.log.debug("Executing adb shell %s", shell_cmd)
ad.log.debug('Executing adb shell %s', shell_cmd)
data = ad.adb.shell(shell_cmd)
return data.decode(BYTE_DECODE_UTF_8).strip()

4
staticlibs/testutils/host/python/mdns_utils.py
View file

@ -23,10 +23,10 @@ def assume_mdns_test_preconditions(
discovery = discovery_device.connectivity_multi_devices_snippet
asserts.skip_if(
not advertising.isAtLeastT(), "Advertising device SDK is lower than T."
not advertising.isAtLeastT(), 'Advertising device SDK is lower than T.'
)
asserts.skip_if(
not discovery.isAtLeastT(), "Discovery device SDK is lower than T."
not discovery.isAtLeastT(), 'Discovery device SDK is lower than T.'
)

4
staticlibs/testutils/host/python/multi_devices_test_base.py
View file

@ -28,7 +28,7 @@ from mobly import base_test
from mobly import utils
from mobly.controllers import android_device
CONNECTIVITY_MULTI_DEVICES_SNIPPET_PACKAGE = "com.google.snippet.connectivity"
CONNECTIVITY_MULTI_DEVICES_SNIPPET_PACKAGE = 'com.google.snippet.connectivity'
class MultiDevicesTestBase(base_test.BaseTestClass):
@ -41,7 +41,7 @@ class MultiDevicesTestBase(base_test.BaseTestClass):
def setup_device(device):
device.load_snippet(
"connectivity_multi_devices_snippet",
'connectivity_multi_devices_snippet',
CONNECTIVITY_MULTI_DEVICES_SNIPPET_PACKAGE,
)

16
staticlibs/testutils/host/python/tether_utils.py
View file

@ -32,7 +32,7 @@ def generate_uuid32_base64() -> str:
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("=")
return base64.b64encode(uuid.uuid1().bytes).decode('utf-8').strip('=')
def assume_hotspot_test_preconditions(
@ -44,23 +44,23 @@ def assume_hotspot_test_preconditions(
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 client.hasWifiFeature(), 'Client requires Wifi feature')
asserts.skip_if(
not server.hasHotspotFeature(), "Server requires hotspot feature"
not server.hasHotspotFeature(), 'Server requires hotspot feature'
)
if upstream_type == UpstreamType.CELLULAR:
asserts.skip_if(
not server.hasTelephonyFeature(), "Server requires Telephony feature"
not server.hasTelephonyFeature(), 'Server requires Telephony feature'
)
elif upstream_type == UpstreamType.WIFI:
asserts.skip_if(
not server.isStaApConcurrencySupported(),
"Server requires Wifi AP + STA concurrency",
'Server requires Wifi AP + STA concurrency',
)
elif upstream_type == UpstreamType.NONE:
pass
else:
raise ValueError(f"Invalid upstream type: {upstream_type}")
raise ValueError(f'Invalid upstream type: {upstream_type}')
def setup_hotspot_and_client_for_upstream_type(
@ -84,11 +84,11 @@ def setup_hotspot_and_client_for_upstream_type(
elif upstream_type == UpstreamType.NONE:
pass
else:
raise ValueError(f"Invalid upstream type: {upstream_type}")
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_ssid = 'HOTSPOT-' + generate_uuid32_base64()
test_passphrase = generate_uuid32_base64()
# Create a hotspot with fixed SSID and password.

10
staticlibs/testutils/host/python/wifip2p_utils.py
View file

@ -25,13 +25,13 @@ def assume_wifi_p2p_test_preconditions(
client = client_device.connectivity_multi_devices_snippet
# Assert pre-conditions
asserts.skip_if(not server.hasWifiFeature(), "Server requires Wifi feature")
asserts.skip_if(not client.hasWifiFeature(), "Client requires Wifi feature")
asserts.skip_if(not server.hasWifiFeature(), 'Server requires Wifi feature')
asserts.skip_if(not client.hasWifiFeature(), 'Client requires Wifi feature')
asserts.skip_if(
not server.isP2pSupported(), "Server requires Wi-fi P2P feature"
not server.isP2pSupported(), 'Server requires Wi-fi P2P feature'
)
asserts.skip_if(
not client.isP2pSupported(), "Client requires Wi-fi P2P feature"
not client.isP2pSupported(), 'Client requires Wi-fi P2P feature'
)
@ -54,7 +54,7 @@ def setup_wifi_p2p_server_and_client(
# network name must starts with prefix "DIRECT-", followed by any two
# random chars from the set ('A' - 'Z', 'a' - 'z', '0' - '9')
# This follows the documentation on WifiP2pConfig.Builder#setNetworkName.
group_name = "DIRECT-XY" + tether_utils.generate_uuid32_base64()
group_name = 'DIRECT-XY' + tether_utils.generate_uuid32_base64()
group_passphrase = tether_utils.generate_uuid32_base64()
# Server creates a Wi-Fi P2P group