From dcbe9b403d64bd3d27fd2cfdb4f2bf76d7536b2c Mon Sep 17 00:00:00 2001 From: Jimi Chen Date: Wed, 28 May 2025 09:33:18 +0000 Subject: [PATCH] Refactor: apply pyformat for coding style consistency add pyformat for presubmit hook Flag: EXEMPT mainline Bug: 419423721 Test: atest CtsConnectivityMultiDevicesTestCases Change-Id: I9abc5cee454e8c8e21ce7e3c134145857680251f --- PREUPLOAD.cfg | 2 + preupload_hook_script.py | 93 ++++ .../tests/unit/host/python/apf_utils_test.py | 420 ++++++++---------- .../testutils/host/python/apf_test_base.py | 57 +-- staticlibs/testutils/host/python/apf_utils.py | 158 ++++--- .../testutils/host/python/assert_utils.py | 15 +- tests/cts/multidevices/apfv4_test.py | 27 +- tests/cts/multidevices/apfv6_test.py | 372 ++++++++-------- tests/cts/multidevices/run_tests.py | 10 +- 9 files changed, 638 insertions(+), 516 deletions(-) create mode 100755 preupload_hook_script.py diff --git a/PREUPLOAD.cfg b/PREUPLOAD.cfg index cbe321c5f9..97a78cf4be 100644 --- a/PREUPLOAD.cfg +++ b/PREUPLOAD.cfg @@ -13,3 +13,5 @@ checkstyle_hook = ${REPO_ROOT}/prebuilts/checkstyle/checkstyle.py --sha ${PREUPL ktlint_hook = ${REPO_ROOT}/prebuilts/ktlint/ktlint.py --disabled-rules comment-wrapping -f ${PREUPLOAD_FILES} hidden_api_txt_checksorted_hook = ${REPO_ROOT}/tools/platform-compat/hiddenapi/checksorted_sha.sh ${PREUPLOAD_COMMIT} ${REPO_ROOT} + +python_hook = ${REPO_ROOT}/packages/modules/Connectivity/preupload_hook_script.py ${PREUPLOAD_FILES} diff --git a/preupload_hook_script.py b/preupload_hook_script.py new file mode 100755 index 0000000000..e2a6401080 --- /dev/null +++ b/preupload_hook_script.py @@ -0,0 +1,93 @@ +#!/usr/bin/env python3 +# Copyright 2025, 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. + +"""Preupload hook to perform necessary checks and formatting on python files.""" + +import argparse +import concurrent.futures +import multiprocessing +import pathlib +import shlex +import subprocess +import sys + + +def _filter_python_files(files: list[pathlib.Path]) -> list[pathlib.Path]: + """Filter a list of files and return a new list of python files only.""" + return [file for file in files if file.suffix == '.py'] + + +def _check_run_shell_command(cmd: str, cwd: str = None) -> None: + """Run a shell command and raise error if failed.""" + if subprocess.run(shlex.split(cmd), cwd=cwd, check=False).returncode: + print('Preupload files did not pass Connectivity preupload hook script.') + sys.exit(1) + + +def _run_pyformat(files: list[pathlib.Path]) -> None: + """Run pyformat on certain projects.""" + if subprocess.run( + shlex.split('which pyformat'), + check=False, + ).returncode: + print('pyformat not available. Please run sudo apt install pyformat.') + sys.exit(1) + + def _run_pyformat_on_file(file): + completed_process = subprocess.run( + shlex.split('pyformat --force_quote_type single ' + file.as_posix()), + capture_output=True, + check=False, + ) + + if completed_process.stdout: + subprocess.run( + shlex.split( + 'pyformat -i --force_quote_type single ' + file.as_posix() + ), + check=False, + ) + return True + return False + + cpu_count = multiprocessing.cpu_count() + with concurrent.futures.ThreadPoolExecutor(max_workers=cpu_count) as executor: + need_reformat = executor.map( + _run_pyformat_on_file, _filter_python_files(files) + ) + + if any(need_reformat): + print( + 'Reformatting completed. Please add the modified files to git and rerun' + ' the repo preupload hook.' + ) + sys.exit(1) + + +def get_preupload_files() -> list[pathlib.Path]: + """Get the list of files to be uploaded.""" + parser = argparse.ArgumentParser() + parser.add_argument('preupload_files', nargs='*', help='Files to upload.') + args = parser.parse_args() + files_to_upload = args.preupload_files + file_paths_to_upload = [ + pathlib.Path(file).resolve() for file in files_to_upload + ] + return [file for file in file_paths_to_upload if file.exists()] + + +if __name__ == '__main__': + preupload_files = get_preupload_files() + _run_pyformat(preupload_files) diff --git a/staticlibs/tests/unit/host/python/apf_utils_test.py b/staticlibs/tests/unit/host/python/apf_utils_test.py index b0a70ae40b..6bf7a962a5 100644 --- a/staticlibs/tests/unit/host/python/apf_utils_test.py +++ b/staticlibs/tests/unit/host/python/apf_utils_test.py @@ -25,22 +25,22 @@ from net_tests_utils.host.python.apf_utils import ( get_apf_capabilities, get_apf_counter, get_apf_counters_from_cmd, - get_ipv4_addresses, - get_non_tentative_ipv6_addresses, get_exclude_all_host_ipv6_multicast_addresses, get_hardware_address, - is_send_raw_packet_downstream_supported, - is_packet_capture_supported, + get_ipv4_addresses, + get_matched_packet_counts, + get_non_tentative_ipv6_addresses, is_apf_dump_counters_supported, + is_packet_capture_supported, + is_send_raw_packet_downstream_supported, + send_raw_packet_downstream, start_capture_packets, stop_capture_packets, - get_matched_packet_counts, - send_raw_packet_downstream, ) from net_tests_utils.host.python.assert_utils import UnexpectedBehaviorError -TEST_IFACE_NAME = "eth0" -TEST_PACKET_IN_HEX = "AABBCCDDEEFF" +TEST_IFACE_NAME = 'eth0' +TEST_PACKET_IN_HEX = 'AABBCCDDEEFF' class TestApfUtils(base_test.BaseTestClass, parameterized.TestCase): @@ -51,27 +51,27 @@ class TestApfUtils(base_test.BaseTestClass, parameterized.TestCase): def setup_test(self): self.mock_ad = MagicMock() # Mock Android device object - @patch("net_tests_utils.host.python.apf_utils.get_apf_counters_from_cmd") + @patch('net_tests_utils.host.python.apf_utils.get_apf_counters_from_cmd') def test_get_apf_counter(self, mock_get_counters: MagicMock) -> None: - iface = "wlan0" + iface = 'wlan0' mock_get_counters.return_value = { - "COUNTER_NAME1": "123", - "COUNTER_NAME2": "456", - "COUNTER_NAME3": "[ERROR: impossible]", + 'COUNTER_NAME1': '123', + 'COUNTER_NAME2': '456', + 'COUNTER_NAME3': '[ERROR: impossible]', } asserts.assert_equal( - get_apf_counter(self.mock_ad, iface, "COUNTER_NAME1"), 123 + get_apf_counter(self.mock_ad, iface, 'COUNTER_NAME1'), 123 ) # Invalid counter value asserts.assert_equal( - get_apf_counter(self.mock_ad, iface, "COUNTER_NAME3"), -1 + get_apf_counter(self.mock_ad, iface, 'COUNTER_NAME3'), -1 ) # Not found asserts.assert_equal( - get_apf_counter(self.mock_ad, iface, "COUNTER_NAME4"), 0 + get_apf_counter(self.mock_ad, iface, 'COUNTER_NAME4'), 0 ) - @patch("net_tests_utils.host.python.adb_utils.adb_shell") + @patch('net_tests_utils.host.python.adb_utils.adb_shell') def test_get_hardware_address_success( self, mock_adb_shell: MagicMock ) -> None: @@ -83,21 +83,19 @@ class TestApfUtils(base_test.BaseTestClass, parameterized.TestCase): 47: wlan1: mtu 1500 qdisc ... link/ether 6a:16:81:ff:82:9b brd ff:ff:ff:ff:ff:ff """ - mac_address = get_hardware_address(self.mock_ad, "wlan0") - asserts.assert_equal(mac_address, "72:05:77:82:21:E0") + mac_address = get_hardware_address(self.mock_ad, 'wlan0') + asserts.assert_equal(mac_address, '72:05:77:82:21:E0') - @patch("net_tests_utils.host.python.adb_utils.adb_shell") + @patch('net_tests_utils.host.python.adb_utils.adb_shell') def test_get_hardware_address_not_found( self, mock_adb_shell: MagicMock ) -> None: - mock_adb_shell.return_value = "Some output without MAC address" + mock_adb_shell.return_value = 'Some output without MAC address' with asserts.assert_raises(PatternNotFoundException): - get_hardware_address(self.mock_ad, "wlan0") + get_hardware_address(self.mock_ad, 'wlan0') - @patch("net_tests_utils.host.python.adb_utils.adb_shell") - def test_get_ipv4_addresses_success( - self, mock_adb_shell: MagicMock - ) -> None: + @patch('net_tests_utils.host.python.adb_utils.adb_shell') + def test_get_ipv4_addresses_success(self, mock_adb_shell: MagicMock) -> None: mock_adb_shell.return_value = """ 54: wlan0: mtu 1500 state UP qlen 1000 inet 192.168.195.162/24 brd 192.168.195.255 scope global wlan0 @@ -105,18 +103,18 @@ class TestApfUtils(base_test.BaseTestClass, parameterized.TestCase): inet 192.168.200.1/24 brd 192.168.200.255 scope global wlan0 valid_lft forever preferred_lft forever """ - ip_addresses = get_ipv4_addresses(self.mock_ad, "wlan0") - asserts.assert_equal(ip_addresses, ["192.168.195.162", "192.168.200.1"]) + ip_addresses = get_ipv4_addresses(self.mock_ad, 'wlan0') + asserts.assert_equal(ip_addresses, ['192.168.195.162', '192.168.200.1']) - @patch("net_tests_utils.host.python.adb_utils.adb_shell") + @patch('net_tests_utils.host.python.adb_utils.adb_shell') def test_get_ipv4_addresses_not_found( self, mock_adb_shell: MagicMock ) -> None: - mock_adb_shell.return_value = "" - ip_addresses = get_ipv4_addresses(self.mock_ad, "wlan0") + mock_adb_shell.return_value = '' + ip_addresses = get_ipv4_addresses(self.mock_ad, 'wlan0') asserts.assert_equal(ip_addresses, []) - @patch("net_tests_utils.host.python.adb_utils.adb_shell") + @patch('net_tests_utils.host.python.adb_utils.adb_shell') def test_get_non_tentative_ipv6_addresses_success( self, mock_adb_shell: MagicMock ) -> None: @@ -129,22 +127,23 @@ class TestApfUtils(base_test.BaseTestClass, parameterized.TestCase): inet6 fe80::3aff:2199:2d8e:20d1/64 scope link noprefixroute valid_lft forever preferred_lft forever """ - ip_addresses = get_non_tentative_ipv6_addresses(self.mock_ad, "wlan0") - asserts.assert_equal(ip_addresses, - ["fe80::10a3:5dff:fe52:de32", - "2001:b400:e53f:164e:9c1e:780e:d1:4658", - "fe80::3aff:2199:2d8e:20d1"]) + ip_addresses = get_non_tentative_ipv6_addresses(self.mock_ad, 'wlan0') + asserts.assert_equal( + ip_addresses, + [ + 'fe80::10a3:5dff:fe52:de32', + '2001:b400:e53f:164e:9c1e:780e:d1:4658', + 'fe80::3aff:2199:2d8e:20d1', + ], + ) - @patch("net_tests_utils.host.python.adb_utils.adb_shell") - def test_get_ipv6_address_not_found( - self, mock_adb_shell: MagicMock - ) -> None: - mock_adb_shell.return_value = "" - ip_addresses = get_non_tentative_ipv6_addresses(self.mock_ad, "wlan0") + @patch('net_tests_utils.host.python.adb_utils.adb_shell') + def test_get_ipv6_address_not_found(self, mock_adb_shell: MagicMock) -> None: + mock_adb_shell.return_value = '' + ip_addresses = get_non_tentative_ipv6_addresses(self.mock_ad, 'wlan0') asserts.assert_equal(ip_addresses, []) - - @patch("net_tests_utils.host.python.adb_utils.adb_shell") + @patch('net_tests_utils.host.python.adb_utils.adb_shell') def test_get_exclude_all_host_ipv6_multicast_addresses_success( self, mock_adb_shell: MagicMock ) -> None: @@ -155,39 +154,43 @@ class TestApfUtils(base_test.BaseTestClass, parameterized.TestCase): inet6 ff02::1 inet6 ff01::1 """ - ip_addresses = get_exclude_all_host_ipv6_multicast_addresses(self.mock_ad, "wlan0") - asserts.assert_equal(ip_addresses, - ["ff02::1:ff99:37b0", - "ff02::1:ffb7:cba2"]) + ip_addresses = get_exclude_all_host_ipv6_multicast_addresses( + self.mock_ad, 'wlan0' + ) + asserts.assert_equal( + ip_addresses, ['ff02::1:ff99:37b0', 'ff02::1:ffb7:cba2'] + ) - @patch("net_tests_utils.host.python.adb_utils.adb_shell") + @patch('net_tests_utils.host.python.adb_utils.adb_shell') def test_get_exclude_all_host_ipv6_multicast_addresses_not_found( - self, mock_adb_shell: MagicMock + self, mock_adb_shell: MagicMock ) -> None: - mock_adb_shell.return_value = "" - ip_addresses = get_exclude_all_host_ipv6_multicast_addresses(self.mock_ad, "wlan0") + mock_adb_shell.return_value = '' + ip_addresses = get_exclude_all_host_ipv6_multicast_addresses( + self.mock_ad, 'wlan0' + ) asserts.assert_equal(ip_addresses, []) - @patch("net_tests_utils.host.python.adb_utils.adb_shell") + @patch('net_tests_utils.host.python.adb_utils.adb_shell') def test_send_raw_packet_downstream_success( self, mock_adb_shell: MagicMock ) -> None: - mock_adb_shell.return_value = "" # Successful command output + mock_adb_shell.return_value = '' # Successful command output send_raw_packet_downstream( self.mock_ad, TEST_IFACE_NAME, TEST_PACKET_IN_HEX ) mock_adb_shell.assert_called_once_with( self.mock_ad, - "cmd network_stack send-raw-packet-downstream" - f" {TEST_IFACE_NAME} {TEST_PACKET_IN_HEX}", + 'cmd network_stack send-raw-packet-downstream' + f' {TEST_IFACE_NAME} {TEST_PACKET_IN_HEX}', ) - @patch("net_tests_utils.host.python.adb_utils.adb_shell") + @patch('net_tests_utils.host.python.adb_utils.adb_shell') def test_send_raw_packet_downstream_failure( self, mock_adb_shell: MagicMock ) -> None: mock_adb_shell.return_value = ( # Unexpected command output - "Any Unexpected Output" + 'Any Unexpected Output' ) with asserts.assert_raises(UnexpectedBehaviorError): send_raw_packet_downstream( @@ -195,15 +198,15 @@ class TestApfUtils(base_test.BaseTestClass, parameterized.TestCase): ) asserts.assert_true( is_send_raw_packet_downstream_supported(self.mock_ad), - "Send raw packet should be supported.", + 'Send raw packet should be supported.', ) - @patch("net_tests_utils.host.python.adb_utils.adb_shell") + @patch('net_tests_utils.host.python.adb_utils.adb_shell') def test_send_raw_packet_downstream_unsupported( self, mock_adb_shell: MagicMock ) -> None: mock_adb_shell.side_effect = AdbError( - cmd="", stdout="Unknown command", stderr="", ret_code=3 + cmd='', stdout='Unknown command', stderr='', ret_code=3 ) with asserts.assert_raises(UnsupportedOperationException): send_raw_packet_downstream( @@ -211,152 +214,122 @@ class TestApfUtils(base_test.BaseTestClass, parameterized.TestCase): ) asserts.assert_false( is_send_raw_packet_downstream_supported(self.mock_ad), - "Send raw packet should not be supported.", + 'Send raw packet should not be supported.', ) - @patch("net_tests_utils.host.python.adb_utils.adb_shell") - def test_start_capture_success( - self, mock_adb_shell: MagicMock - ) -> None: - mock_adb_shell.return_value = "success" # Successful command output - start_capture_packets( - self.mock_ad, TEST_IFACE_NAME - ) - mock_adb_shell.assert_called_once_with( - self.mock_ad, - "cmd network_stack capture start" - f" {TEST_IFACE_NAME}" - ) + @patch('net_tests_utils.host.python.adb_utils.adb_shell') + def test_start_capture_success(self, mock_adb_shell: MagicMock) -> None: + mock_adb_shell.return_value = 'success' # Successful command output + start_capture_packets(self.mock_ad, TEST_IFACE_NAME) + mock_adb_shell.assert_called_once_with( + self.mock_ad, f'cmd network_stack capture start {TEST_IFACE_NAME}' + ) - @patch("net_tests_utils.host.python.adb_utils.adb_shell") - def test_start_capture_failure( - self, mock_adb_shell: MagicMock - ) -> None: - mock_adb_shell.return_value = ( # Unexpected command output - "Any Unexpected Output" - ) - with asserts.assert_raises(UnexpectedBehaviorError): - start_capture_packets( - self.mock_ad, TEST_IFACE_NAME - ) - asserts.assert_true( - is_packet_capture_supported(self.mock_ad), - "Start capturing packets should be supported.", - ) + @patch('net_tests_utils.host.python.adb_utils.adb_shell') + def test_start_capture_failure(self, mock_adb_shell: MagicMock) -> None: + mock_adb_shell.return_value = ( # Unexpected command output + 'Any Unexpected Output' + ) + with asserts.assert_raises(UnexpectedBehaviorError): + start_capture_packets(self.mock_ad, TEST_IFACE_NAME) + asserts.assert_true( + is_packet_capture_supported(self.mock_ad), + 'Start capturing packets should be supported.', + ) - @patch("net_tests_utils.host.python.adb_utils.adb_shell") - def test_start_capture_unsupported( - self, mock_adb_shell: MagicMock - ) -> None: - mock_adb_shell.side_effect = AdbError( - cmd="", stdout="Unknown command", stderr="", ret_code=3 - ) - with asserts.assert_raises(UnsupportedOperationException): - start_capture_packets( - self.mock_ad, TEST_IFACE_NAME - ) - asserts.assert_false( - is_packet_capture_supported(self.mock_ad), - "Start capturing packets should not be supported.", - ) + @patch('net_tests_utils.host.python.adb_utils.adb_shell') + def test_start_capture_unsupported(self, mock_adb_shell: MagicMock) -> None: + mock_adb_shell.side_effect = AdbError( + cmd='', stdout='Unknown command', stderr='', ret_code=3 + ) + with asserts.assert_raises(UnsupportedOperationException): + start_capture_packets(self.mock_ad, TEST_IFACE_NAME) + asserts.assert_false( + is_packet_capture_supported(self.mock_ad), + 'Start capturing packets should not be supported.', + ) - @patch("net_tests_utils.host.python.adb_utils.adb_shell") - def test_stop_capture_success( - self, mock_adb_shell: MagicMock - ) -> None: - mock_adb_shell.return_value = "success" # Successful command output - stop_capture_packets( - self.mock_ad, TEST_IFACE_NAME - ) - mock_adb_shell.assert_called_once_with( - self.mock_ad, - "cmd network_stack capture stop" - f" {TEST_IFACE_NAME}" - ) + @patch('net_tests_utils.host.python.adb_utils.adb_shell') + def test_stop_capture_success(self, mock_adb_shell: MagicMock) -> None: + mock_adb_shell.return_value = 'success' # Successful command output + stop_capture_packets(self.mock_ad, TEST_IFACE_NAME) + mock_adb_shell.assert_called_once_with( + self.mock_ad, f'cmd network_stack capture stop {TEST_IFACE_NAME}' + ) - @patch("net_tests_utils.host.python.adb_utils.adb_shell") - def test_stop_capture_failure( - self, mock_adb_shell: MagicMock - ) -> None: - mock_adb_shell.return_value = ( # Unexpected command output - "Any Unexpected Output" - ) - with asserts.assert_raises(UnexpectedBehaviorError): - stop_capture_packets( - self.mock_ad, TEST_IFACE_NAME - ) - asserts.assert_true( - is_packet_capture_supported(self.mock_ad), - "Stop capturing packets should be supported.", - ) + @patch('net_tests_utils.host.python.adb_utils.adb_shell') + def test_stop_capture_failure(self, mock_adb_shell: MagicMock) -> None: + mock_adb_shell.return_value = ( # Unexpected command output + 'Any Unexpected Output' + ) + with asserts.assert_raises(UnexpectedBehaviorError): + stop_capture_packets(self.mock_ad, TEST_IFACE_NAME) + asserts.assert_true( + is_packet_capture_supported(self.mock_ad), + 'Stop capturing packets should be supported.', + ) - @patch("net_tests_utils.host.python.adb_utils.adb_shell") - def test_stop_capture_unsupported( - self, mock_adb_shell: MagicMock - ) -> None: - mock_adb_shell.side_effect = AdbError( - cmd="", stdout="Unknown command", stderr="", ret_code=3 - ) - with asserts.assert_raises(UnsupportedOperationException): - stop_capture_packets( - self.mock_ad, TEST_IFACE_NAME - ) - asserts.assert_false( - is_packet_capture_supported(self.mock_ad), - "Stop capturing packets should not be supported.", - ) + @patch('net_tests_utils.host.python.adb_utils.adb_shell') + def test_stop_capture_unsupported(self, mock_adb_shell: MagicMock) -> None: + mock_adb_shell.side_effect = AdbError( + cmd='', stdout='Unknown command', stderr='', ret_code=3 + ) + with asserts.assert_raises(UnsupportedOperationException): + stop_capture_packets(self.mock_ad, TEST_IFACE_NAME) + asserts.assert_false( + is_packet_capture_supported(self.mock_ad), + 'Stop capturing packets should not be supported.', + ) - @patch("net_tests_utils.host.python.adb_utils.adb_shell") + @patch('net_tests_utils.host.python.adb_utils.adb_shell') def test_get_matched_packet_counts_success( - self, mock_adb_shell: MagicMock + self, mock_adb_shell: MagicMock ) -> None: - mock_adb_shell.return_value = "10" # Successful command output + mock_adb_shell.return_value = '10' # Successful command output + get_matched_packet_counts(self.mock_ad, TEST_IFACE_NAME, TEST_PACKET_IN_HEX) + mock_adb_shell.assert_called_once_with( + self.mock_ad, + 'cmd network_stack capture matched-packet-counts' + f' {TEST_IFACE_NAME} {TEST_PACKET_IN_HEX}', + ) + + @patch('net_tests_utils.host.python.adb_utils.adb_shell') + def test_get_matched_packet_counts_failure( + self, mock_adb_shell: MagicMock + ) -> None: + mock_adb_shell.return_value = ( # Unexpected command output + 'Any Unexpected Output' + ) + with asserts.assert_raises(UnexpectedBehaviorError): get_matched_packet_counts( self.mock_ad, TEST_IFACE_NAME, TEST_PACKET_IN_HEX ) - mock_adb_shell.assert_called_once_with( - self.mock_ad, - "cmd network_stack capture matched-packet-counts" - f" {TEST_IFACE_NAME} {TEST_PACKET_IN_HEX}" - ) + asserts.assert_true( + is_packet_capture_supported(self.mock_ad), + 'Get matched packet counts should be supported.', + ) - @patch("net_tests_utils.host.python.adb_utils.adb_shell") - def test_get_matched_packet_counts_failure( - self, mock_adb_shell: MagicMock - ) -> None: - mock_adb_shell.return_value = ( # Unexpected command output - "Any Unexpected Output" - ) - with asserts.assert_raises(UnexpectedBehaviorError): - get_matched_packet_counts( - self.mock_ad, TEST_IFACE_NAME, TEST_PACKET_IN_HEX - ) - asserts.assert_true( - is_packet_capture_supported(self.mock_ad), - "Get matched packet counts should be supported.", - ) - - @patch("net_tests_utils.host.python.adb_utils.adb_shell") + @patch('net_tests_utils.host.python.adb_utils.adb_shell') def test_get_matched_packet_counts_unsupported( - self, mock_adb_shell: MagicMock + self, mock_adb_shell: MagicMock ) -> None: - mock_adb_shell.side_effect = AdbError( - cmd="", stdout="Unknown command", stderr="", ret_code=3 - ) - with asserts.assert_raises(UnsupportedOperationException): - get_matched_packet_counts( - self.mock_ad, TEST_IFACE_NAME, TEST_PACKET_IN_HEX - ) - asserts.assert_false( - is_packet_capture_supported(self.mock_ad), - "Get matched packet counts should not be supported.", + mock_adb_shell.side_effect = AdbError( + cmd='', stdout='Unknown command', stderr='', ret_code=3 + ) + with asserts.assert_raises(UnsupportedOperationException): + get_matched_packet_counts( + self.mock_ad, TEST_IFACE_NAME, TEST_PACKET_IN_HEX ) + asserts.assert_false( + is_packet_capture_supported(self.mock_ad), + 'Get matched packet counts should not be supported.', + ) - @patch("net_tests_utils.host.python.adb_utils.adb_shell") + @patch('net_tests_utils.host.python.adb_utils.adb_shell') def test_get_apf_counters_from_cmd_success( - self, mock_adb_shell: MagicMock + self, mock_adb_shell: MagicMock ) -> None: - mock_adb_shell.return_value = """ + mock_adb_shell.return_value = """ NDIANNESS: 305419896 TOTAL_PACKETS: 157553 FILTER_AGE_SECONDS: 356 @@ -380,53 +353,46 @@ class TestApfUtils(base_test.BaseTestClass, parameterized.TestCase): DROPPED_ARP_OTHER_HOST: 89 DROPPED_ARP_REQUEST_REPLIED: 13 """ - get_apf_counters_from_cmd( - self.mock_ad, TEST_IFACE_NAME - ) - mock_adb_shell.assert_called_once_with( - self.mock_ad, - f"cmd network_stack apf {TEST_IFACE_NAME} dump-counters" - ) + get_apf_counters_from_cmd(self.mock_ad, TEST_IFACE_NAME) + mock_adb_shell.assert_called_once_with( + self.mock_ad, f'cmd network_stack apf {TEST_IFACE_NAME} dump-counters' + ) - @patch("net_tests_utils.host.python.adb_utils.adb_shell") + @patch('net_tests_utils.host.python.adb_utils.adb_shell') def test_get_apf_counters_from_cmd_failure( - self, mock_adb_shell: MagicMock + self, mock_adb_shell: MagicMock ) -> None: - mock_adb_shell.return_value = ( # Unexpected command output - "Any Unexpected Output" - ) - with asserts.assert_raises(UnexpectedBehaviorError): - get_apf_counters_from_cmd( - self.mock_ad, TEST_IFACE_NAME - ) - asserts.assert_true( - is_packet_capture_supported(self.mock_ad), - "Dump APF packet counters should be supported.", - ) + mock_adb_shell.return_value = ( # Unexpected command output + 'Any Unexpected Output' + ) + with asserts.assert_raises(UnexpectedBehaviorError): + get_apf_counters_from_cmd(self.mock_ad, TEST_IFACE_NAME) + asserts.assert_true( + is_packet_capture_supported(self.mock_ad), + 'Dump APF packet counters should be supported.', + ) - @patch("net_tests_utils.host.python.adb_utils.adb_shell") + @patch('net_tests_utils.host.python.adb_utils.adb_shell') def test_get_apf_counters_from_cmd_unsupported( - self, mock_adb_shell: MagicMock + self, mock_adb_shell: MagicMock ) -> None: - mock_adb_shell.side_effect = AdbError( - cmd="", stdout="Unknown command", stderr="", ret_code=3 - ) - with asserts.assert_raises(UnsupportedOperationException): - get_apf_counters_from_cmd( - self.mock_ad, TEST_IFACE_NAME - ) - asserts.assert_false( - is_apf_dump_counters_supported(self.mock_ad), - "Dump APF packet counters should not be supported.", - ) + mock_adb_shell.side_effect = AdbError( + cmd='', stdout='Unknown command', stderr='', ret_code=3 + ) + with asserts.assert_raises(UnsupportedOperationException): + get_apf_counters_from_cmd(self.mock_ad, TEST_IFACE_NAME) + asserts.assert_false( + is_apf_dump_counters_supported(self.mock_ad), + 'Dump APF packet counters should not be supported.', + ) @parameterized.parameters( - ("2,2048,1", ApfCapabilities(2, 2048, 1)), # Valid input - ("3,1024,0", ApfCapabilities(3, 1024, 0)), # Valid input - ("invalid,output", ApfCapabilities(0, 0, 0)), # Invalid input - ("", ApfCapabilities(0, 0, 0)), # Empty input + ('2,2048,1', ApfCapabilities(2, 2048, 1)), # Valid input + ('3,1024,0', ApfCapabilities(3, 1024, 0)), # Valid input + ('invalid,output', ApfCapabilities(0, 0, 0)), # Invalid input + ('', ApfCapabilities(0, 0, 0)), # Empty input ) - @patch("net_tests_utils.host.python.adb_utils.adb_shell") + @patch('net_tests_utils.host.python.adb_utils.adb_shell') def test_get_apf_capabilities( self, mock_output, expected_result, mock_adb_shell ): @@ -435,7 +401,7 @@ class TestApfUtils(base_test.BaseTestClass, parameterized.TestCase): mock_adb_shell.return_value = mock_output # Call the function under test - result = get_apf_capabilities(self.mock_ad, "wlan0") + result = get_apf_capabilities(self.mock_ad, 'wlan0') # Assert that the result matches the expected result asserts.assert_equal(result, expected_result) diff --git a/staticlibs/testutils/host/python/apf_test_base.py b/staticlibs/testutils/host/python/apf_test_base.py index c9cda417a3..f77dd42603 100644 --- a/staticlibs/testutils/host/python/apf_test_base.py +++ b/staticlibs/testutils/host/python/apf_test_base.py @@ -12,10 +12,11 @@ # See the License for the specific language governing permissions and # limitations under the License. +import time from mobly import asserts from net_tests_utils.host.python import adb_utils, apf_utils, assert_utils, multi_devices_test_base, tether_utils from net_tests_utils.host.python.tether_utils import UpstreamType -import time + class ApfTestBase(multi_devices_test_base.MultiDevicesTestBase): @@ -25,8 +26,8 @@ class ApfTestBase(multi_devices_test_base.MultiDevicesTestBase): # Check test preconditions. asserts.abort_class_if( not self.client.isAtLeastV(), - "Do not enforce the test until V+ since chipset potential bugs are" - " expected to be fixed on V+ releases.", + 'Do not enforce the test until V+ since chipset potential bugs are' + ' expected to be fixed on V+ releases.', ) tether_utils.assume_hotspot_test_preconditions( self.serverDevice, self.clientDevice, UpstreamType.NONE @@ -35,7 +36,7 @@ class ApfTestBase(multi_devices_test_base.MultiDevicesTestBase): not apf_utils.is_send_raw_packet_downstream_supported( self.serverDevice ), - "NetworkStack is too old to support send raw packet, skip test.", + 'NetworkStack is too old to support send raw packet, skip test.', ) # Fetch device properties and storing them locally for later use. @@ -104,35 +105,35 @@ class ApfTestBase(multi_devices_test_base.MultiDevicesTestBase): self, send_packet: str, counter_name: str, receive_packet: str ) -> None: try: - apf_utils.start_capture_packets(self.serverDevice, self.server_iface_name) + apf_utils.start_capture_packets(self.serverDevice, self.server_iface_name) - count_before_test = apf_utils.get_apf_counter( - self.clientDevice, - self.client_iface_name, - counter_name, - ) + count_before_test = apf_utils.get_apf_counter( + self.clientDevice, + self.client_iface_name, + counter_name, + ) - apf_utils.send_raw_packet_downstream( - self.serverDevice, self.server_iface_name, send_packet - ) + apf_utils.send_raw_packet_downstream( + self.serverDevice, self.server_iface_name, send_packet + ) + assert_utils.expect_with_retry( + lambda: apf_utils.get_matched_packet_counts( + self.serverDevice, self.server_iface_name, receive_packet + ) + == 1 + ) + + # TODO: re-enable once the test passes reliably. + if False: assert_utils.expect_with_retry( - lambda: apf_utils.get_matched_packet_counts( - self.serverDevice, self.server_iface_name, receive_packet + lambda: apf_utils.get_apf_counter( + self.clientDevice, + self.client_iface_name, + counter_name, ) - == 1 + > count_before_test ) - # TODO: re-enable once the test passes reliably. - if False: - assert_utils.expect_with_retry( - lambda: apf_utils.get_apf_counter( - self.clientDevice, - self.client_iface_name, - counter_name, - ) - > count_before_test - ) - finally: - apf_utils.stop_capture_packets(self.serverDevice, self.server_iface_name) + apf_utils.stop_capture_packets(self.serverDevice, self.server_iface_name) diff --git a/staticlibs/testutils/host/python/apf_utils.py b/staticlibs/testutils/host/python/apf_utils.py index 1f91a5e715..3a1c8f5e3e 100644 --- a/staticlibs/testutils/host/python/apf_utils.py +++ b/staticlibs/testutils/host/python/apf_utils.py @@ -13,12 +13,12 @@ # limitations under the License. from dataclasses import dataclass +import functools import re from mobly import asserts from mobly.controllers import android_device from mobly.controllers.android_device_lib.adb import AdbError from net_tests_utils.host.python import adb_utils, assert_utils -import functools class PatternNotFoundException(Exception): @@ -38,6 +38,7 @@ def get_apf_counter( except ValueError: return -1 + def get_apf_counters_from_cmd( ad: android_device.AndroidDevice, iface_name: str ) -> dict: @@ -45,16 +46,19 @@ def get_apf_counters_from_cmd( output = AdbOutputHandler(ad, cmd).get_output() data_dict = {} for line in output.strip().split('\n'): - if line: # Ensure the line is not empty + if line: # Ensure the line is not empty # split only at the first colon parts = line.split(':', 1) if len(parts) == 2: data_dict[parts[0]] = parts[1] else: - raise assert_utils.UnexpectedBehaviorError(f"Got unexpected output: {output}.") + raise assert_utils.UnexpectedBehaviorError( + f'Got unexpected output: {output}.' + ) return data_dict + def get_ipv4_addresses( ad: android_device.AndroidDevice, iface_name: str ) -> list[str]: @@ -78,8 +82,8 @@ def get_ipv4_addresses( # valid_lft forever preferred_lft forever # inet 192.168.1.1/24 brd 192.168.1.255 scope global wlan2 # valid_lft forever preferred_lft forever - output = adb_utils.adb_shell(ad, f"ip -4 address show {iface_name}") - pattern = r"inet\s+(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})\/\d+" + output = adb_utils.adb_shell(ad, f'ip -4 address show {iface_name}') + pattern = r'inet\s+(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})\/\d+' matches = re.findall(pattern, output) if matches: @@ -87,12 +91,14 @@ def get_ipv4_addresses( else: return [] + def get_non_tentative_ipv6_addresses( ad: android_device.AndroidDevice, iface_name: str ) -> list[str]: """Retrieves the non-tentative IPv6 addresses of a given interface on an Android device. - This function executes an ADB shell command (`ip -6 address show -tentative`) to get the + This function executes an ADB shell command (`ip -6 address show -tentative`) + to get the network interface information and extracts the IPv6 address from the output. If devices have no IPv6 address, raise PatternNotFoundException. @@ -110,8 +116,10 @@ def get_non_tentative_ipv6_addresses( # valid_lft forever preferred_lft forever # inet6 fe80::1233:aadb:3d32:1234/64 scope link # valid_lft forever preferred_lft forever - output = adb_utils.adb_shell(ad, f"ip -6 address show -tentative {iface_name}") - pattern = r"inet6\s+([0-9a-fA-F:]+)\/\d+" + output = adb_utils.adb_shell( + ad, f'ip -6 address show -tentative {iface_name}' + ) + pattern = r'inet6\s+([0-9a-fA-F:]+)\/\d+' matches = re.findall(pattern, output) if matches: @@ -119,13 +127,15 @@ def get_non_tentative_ipv6_addresses( else: return [] + def get_exclude_all_host_ipv6_multicast_addresses( ad: android_device.AndroidDevice, iface_name: str ) -> list[str]: """Retrieves the IPv6 multicast addresses of a given interface on an Android device. This function executes an ADB shell command (`ip -6 maddr show`) to get the - network interface information and extracts the IPv6 multicast address from the output. + network interface information and extracts the IPv6 multicast address from the + output. If devices have no IPv6 multicast address, raise PatternNotFoundException. Args: @@ -142,15 +152,16 @@ def get_exclude_all_host_ipv6_multicast_addresses( # inet6 ff02::1:ffb7:cba2 users 2 # inet6 ff02::1 # inet6 ff01::1 - output = adb_utils.adb_shell(ad, f"ip -6 maddr show {iface_name}") - pattern = r"inet6\s+([a-fA-F0-9:]+)(?:\s+users\s+\d+)?" + output = adb_utils.adb_shell(ad, f'ip -6 maddr show {iface_name}') + pattern = r'inet6\s+([a-fA-F0-9:]+)(?:\s+users\s+\d+)?' matches = re.findall(pattern, output) if matches: - return [addr for addr in matches if addr not in ("ff02::1", "ff01::1")] + return [addr for addr in matches if addr not in ('ff02::1', 'ff01::1')] else: return [] + def get_hardware_address( ad: android_device.AndroidDevice, iface_name: str ) -> str: @@ -166,7 +177,7 @@ def get_hardware_address( """ # Run the "ip link" command and get its output. - ip_link_output = adb_utils.adb_shell(ad, f"ip link") + ip_link_output = adb_utils.adb_shell(ad, f'ip link') # Regular expression to extract the MAC address. # Parse hardware address from ip link output like below: @@ -176,34 +187,37 @@ def get_hardware_address( # link/ether 72:05:77:82:21:e0 brd ff:ff:ff:ff:ff:ff # 47: wlan1: mtu 1500 qdisc ... # link/ether 6a:16:81:ff:82:9b brd ff:ff:ff:ff:ff:ff" - pattern = rf"{iface_name}:.*?link/ether (([0-9a-fA-F]{{2}}:){{5}}[0-9a-fA-F]{{2}})" + pattern = ( + rf'{iface_name}:.*?link/ether (([0-9a-fA-F]{{2}}:){{5}}[0-9a-fA-F]{{2}})' + ) match = re.search(pattern, ip_link_output, re.DOTALL) if match: return match.group(1).upper() # Extract the MAC address string. else: raise PatternNotFoundException( - "Cannot get hardware address for " + iface_name + 'Cannot get hardware address for ' + iface_name ) + def is_packet_capture_supported( - ad: android_device.AndroidDevice, + ad: android_device.AndroidDevice, ) -> bool: try: # Invoke the shell command with empty argument and see how NetworkStack respond. # If supported, an IllegalArgumentException with help page will be printed. assert_utils.expect_throws( - lambda: start_capture_packets(ad, ""), - assert_utils.UnexpectedBehaviorError + lambda: start_capture_packets(ad, ''), + assert_utils.UnexpectedBehaviorError, ) assert_utils.expect_throws( - lambda: stop_capture_packets(ad, ""), - assert_utils.UnexpectedBehaviorError + lambda: stop_capture_packets(ad, ''), + assert_utils.UnexpectedBehaviorError, ) assert_utils.expect_throws( - lambda: get_matched_packet_counts(ad, "", ""), - assert_utils.UnexpectedBehaviorError + lambda: get_matched_packet_counts(ad, '', ''), + assert_utils.UnexpectedBehaviorError, ) except assert_utils.UnexpectedExceptionError: return False @@ -211,15 +225,14 @@ def is_packet_capture_supported( # If no UnsupportOperationException is thrown, regard it as supported return True -def is_apf_dump_counters_supported( - ad: android_device.AndroidDevice -) -> bool: + +def is_apf_dump_counters_supported(ad: android_device.AndroidDevice) -> bool: try: # Invoke the shell command with empty argument and see how NetworkStack respond. # If supported, an IllegalArgumentException with help page will be printed. assert_utils.expect_throws( - lambda: get_apf_counters_from_cmd(ad, ""), - assert_utils.UnexpectedBehaviorError + lambda: get_apf_counters_from_cmd(ad, ''), + assert_utils.UnexpectedBehaviorError, ) except assert_utils.UnexpectedExceptionError: return False @@ -227,13 +240,14 @@ def is_apf_dump_counters_supported( # If no UnsupportOperationException is thrown, regard it as supported return True + def is_send_raw_packet_downstream_supported( ad: android_device.AndroidDevice, ) -> bool: try: # Invoke the shell command with empty argument and see how NetworkStack respond. # If supported, an IllegalArgumentException with help page will be printed. - send_raw_packet_downstream(ad, "", "") + send_raw_packet_downstream(ad, '', '') except assert_utils.UnexpectedBehaviorError: return True except UnsupportedOperationException: @@ -272,18 +286,21 @@ def send_raw_packet_downstream( representation of a packet starting from L2 header. """ - cmd = f"cmd network_stack send-raw-packet-downstream {iface_name} {packet_in_hex}" + cmd = ( + 'cmd network_stack send-raw-packet-downstream' + f' {iface_name} {packet_in_hex}' + ) # Expect no output or Unknown command if NetworkStack is too old. Throw otherwise. adb_output = AdbOutputHandler(ad, cmd).get_output() if adb_output: raise assert_utils.UnexpectedBehaviorError( - f"Got unexpected output: {adb_output} for command: {cmd}." + f'Got unexpected output: {adb_output} for command: {cmd}.' ) + def start_capture_packets( - ad: android_device.AndroidDevice, - iface_name: str + ad: android_device.AndroidDevice, iface_name: str ) -> None: """Starts packet capturing on a specified network interface. @@ -296,18 +313,18 @@ def start_capture_packets( ad: The Android device object. iface_name: The name of the network interface (e.g., "wlan0"). """ - cmd = f"cmd network_stack capture start {iface_name}" + cmd = f'cmd network_stack capture start {iface_name}' # Expect no output or Unknown command if NetworkStack is too old. Throw otherwise. adb_output = AdbOutputHandler(ad, cmd).get_output() - if adb_output != "success": + if adb_output != 'success': raise assert_utils.UnexpectedBehaviorError( - f"Got unexpected output: {adb_output} for command: {cmd}." + f'Got unexpected output: {adb_output} for command: {cmd}.' ) + def stop_capture_packets( - ad: android_device.AndroidDevice, - iface_name: str + ad: android_device.AndroidDevice, iface_name: str ) -> None: """Stops packet capturing on a specified network interface. @@ -319,19 +336,18 @@ def stop_capture_packets( ad: The Android device object. iface_name: The name of the network interface (e.g., "wlan0"). """ - cmd = f"cmd network_stack capture stop {iface_name}" + cmd = f'cmd network_stack capture stop {iface_name}' # Expect no output or Unknown command if NetworkStack is too old. Throw otherwise. adb_output = AdbOutputHandler(ad, cmd).get_output() - if adb_output != "success": + if adb_output != 'success': raise assert_utils.UnexpectedBehaviorError( - f"Got unexpected output: {adb_output} for command: {cmd}." + f'Got unexpected output: {adb_output} for command: {cmd}.' ) + def get_matched_packet_counts( - ad: android_device.AndroidDevice, - iface_name: str, - packet_in_hex: str + ad: android_device.AndroidDevice, iface_name: str, packet_in_hex: str ) -> int: """Gets the number of captured packets matching a specific hexadecimal pattern. @@ -348,7 +364,10 @@ def get_matched_packet_counts( Returns: The number of matched packets as an integer. """ - cmd = f"cmd network_stack capture matched-packet-counts {iface_name} {packet_in_hex}" + cmd = ( + 'cmd network_stack capture matched-packet-counts' + f' {iface_name} {packet_in_hex}' + ) # Expect no output or Unknown command if NetworkStack is too old. Throw otherwise. adb_output = AdbOutputHandler(ad, cmd).get_output() @@ -356,9 +375,10 @@ def get_matched_packet_counts( return int(adb_output) except ValueError as e: raise assert_utils.UnexpectedBehaviorError( - f"Got unexpected exception: {e} for command: {cmd}." + f'Got unexpected exception: {e} for command: {cmd}.' ) + @dataclass class ApfCapabilities: """APF program support capabilities. @@ -391,9 +411,9 @@ class ApfCapabilities: def __str__(self): """Returns a user-friendly string representation of the APF capabilities.""" return ( - f"APF Version: {self.apf_version_supported}\n" - f"Ram Size: {self.apf_ram_size} bytes\n" - f"Packet Format: {self.apf_packet_format}" + f'APF Version: {self.apf_version_supported}\n' + f'Ram Size: {self.apf_ram_size} bytes\n' + f'Packet Format: {self.apf_packet_format}' ) @@ -401,10 +421,10 @@ def get_apf_capabilities( ad: android_device.AndroidDevice, iface_name: str ) -> ApfCapabilities: output = adb_utils.adb_shell( - ad, f"cmd network_stack apf {iface_name} capabilities" + ad, f'cmd network_stack apf {iface_name} capabilities' ) try: - values = [int(value_str) for value_str in output.split(",")] + values = [int(value_str) for value_str in output.split(',')] except ValueError: return ApfCapabilities(0, 0, 0) # Conversion to integer failed return ApfCapabilities(values[0], values[1], values[2]) @@ -416,43 +436,55 @@ def assume_apf_version_support_at_least( caps = get_apf_capabilities(ad, iface_name) asserts.abort_class_if( caps.apf_version_supported < expected_version, - f"Supported apf version {caps.apf_version_supported} < expected version" - f" {expected_version}", + f'Supported apf version {caps.apf_version_supported} < expected version' + f' {expected_version}', ) + def at_least_B(): def decorator(test_function): @functools.wraps(test_function) def wrapper(self, *args, **kwargs): asserts.abort_class_if( - (not hasattr(self, 'client')) or (not hasattr(self.client, 'isAtLeastB')), - "no valid client attribute" + (not hasattr(self, 'client')) + or (not hasattr(self.client, 'isAtLeastB')), + 'no valid client attribute', ) - asserts.abort_class_if(not self.client.isAtLeastB(), "client device is not Android B+") + asserts.abort_class_if( + not self.client.isAtLeastB(), 'client device is not Android B+' + ) return test_function(self, *args, **kwargs) + return wrapper + return decorator + def apf_ram_at_least(size): def decorator(test_function): @functools.wraps(test_function) def wrapper(self, *args, **kwargs): asserts.abort_class_if( - (not hasattr(self, 'clientDevice')) or (not hasattr(self, 'client_iface_name')), - "no valid client attribute" + (not hasattr(self, 'clientDevice')) + or (not hasattr(self, 'client_iface_name')), + 'no valid client attribute', ) caps = get_apf_capabilities(self.clientDevice, self.client_iface_name) asserts.skip_if( - caps.apf_ram_size < size, - f'APF rame size {caps.apf_ram_size} < {size}' + caps.apf_ram_size < size, + f'APF rame size {caps.apf_ram_size} < {size}', ) return test_function(self, *args, **kwargs) + return wrapper + return decorator + class AdbOutputHandler: + def __init__(self, ad, cmd): self._ad = ad self._cmd = cmd @@ -462,8 +494,6 @@ class AdbOutputHandler: return adb_utils.adb_shell(self._ad, self._cmd) except AdbError as e: output = str(e.stdout) - if "Unknown command" in output: - raise UnsupportedOperationException( - f"{self._cmd} is not supported." - ) - return output \ No newline at end of file + if 'Unknown command' in output: + raise UnsupportedOperationException(f'{self._cmd} is not supported.') + return output diff --git a/staticlibs/testutils/host/python/assert_utils.py b/staticlibs/testutils/host/python/assert_utils.py index 40094a2946..12cf4f50d7 100644 --- a/staticlibs/testutils/host/python/assert_utils.py +++ b/staticlibs/testutils/host/python/assert_utils.py @@ -19,9 +19,11 @@ from typing import Callable class UnexpectedBehaviorError(Exception): """Raised when there is an unexpected behavior during applying a procedure.""" + class UnexpectedExceptionError(Exception): """Raised when there is an unexpected exception throws during applying a procedure""" + def expect_with_retry( predicate: Callable[[], bool], retry_action: Callable[[], None] = None, @@ -41,19 +43,20 @@ def expect_with_retry( time.sleep(retry_interval_sec) raise UnexpectedBehaviorError( - "Predicate didn't become true after " + str(max_retries) + " retries." + "Predicate didn't become true after " + str(max_retries) + ' retries.' ) + def expect_throws(runnable: callable, exception_class) -> None: try: runnable() - raise UnexpectedBehaviorError("Expected an exception, but none was thrown") + raise UnexpectedBehaviorError('Expected an exception, but none was thrown') except exception_class: pass except UnexpectedBehaviorError as e: raise e except Exception as e: - raise UnexpectedExceptionError( - f"Expected exception of type {exception_class.__name__}, " - f"but got {type(e).__name__}: {e}" - ) \ No newline at end of file + raise UnexpectedExceptionError( + f'Expected exception of type {exception_class.__name__}, ' + f'but got {type(e).__name__}: {e}' + ) diff --git a/tests/cts/multidevices/apfv4_test.py b/tests/cts/multidevices/apfv4_test.py index 2f212fcdc9..6071d40769 100644 --- a/tests/cts/multidevices/apfv4_test.py +++ b/tests/cts/multidevices/apfv4_test.py @@ -12,19 +12,20 @@ # See the License for the specific language governing permissions and # limitations under the License. +import time from absl.testing import parameterized from mobly import asserts from net_tests_utils.host.python import apf_test_base, apf_utils from scapy.layers.l2 import Ether -import time # Constants. -COUNTER_DROPPED_ETHERTYPE_NOT_ALLOWED = "DROPPED_ETHERTYPE_NOT_ALLOWED" -ETHER_BROADCAST_ADDR = "FFFFFFFFFFFF" +COUNTER_DROPPED_ETHERTYPE_NOT_ALLOWED = 'DROPPED_ETHERTYPE_NOT_ALLOWED' +ETHER_BROADCAST_ADDR = 'FFFFFFFFFFFF' MIN_PACKET_SIZE = 60 class ApfV4Test(apf_test_base.ApfTestBase, parameterized.TestCase): + def setup_class(self): super().setup_class() # Check apf version preconditions. @@ -35,7 +36,7 @@ class ApfV4Test(apf_test_base.ApfTestBase, parameterized.TestCase): # Enforce APFv4 support for Android 14+ VSR. asserts.assert_true( caps.apf_version_supported >= 4, - "APFv4 became mandatory in Android 14 VSR.", + 'APFv4 became mandatory in Android 14 VSR.', ) else: # Skip tests for APF version < 4 before Android 14 VSR. @@ -48,18 +49,22 @@ class ApfV4Test(apf_test_base.ApfTestBase, parameterized.TestCase): # Tests can use any disallowed packet type. Currently, # several ethertypes from the legacy ApfFilter denylist are used. @parameterized.parameters( - 0x88a2, # ATA over Ethernet - 0x88a4, # EtherCAT - 0x88b8, # GOOSE (Generic Object Oriented Substation event) - 0x88cd, # SERCOS III - 0x88e3, # Media Redundancy Protocol (IEC62439-2) + 0x88A2, # ATA over Ethernet + 0x88A4, # EtherCAT + 0x88B8, # GOOSE (Generic Object Oriented Substation event) + 0x88CD, # SERCOS III + 0x88E3, # Media Redundancy Protocol (IEC62439-2) ) # Declare inputs for state_str and expected_result. def test_apf_drop_ethertype_not_allowed(self, blocked_ether_type): - eth = Ether(src=self.server_mac_address, dst=self.client_mac_address, type=blocked_ether_type) + eth = Ether( + src=self.server_mac_address, + dst=self.client_mac_address, + type=blocked_ether_type, + ) packet = bytes(eth).hex() # Add zero padding up to minimum ethernet frame length - packet = packet.ljust(MIN_PACKET_SIZE * 2, "0") + packet = packet.ljust(MIN_PACKET_SIZE * 2, '0') # Pause packet sending between tests to avoid APF disablement due to high throughput. time.sleep(3) diff --git a/tests/cts/multidevices/apfv6_test.py b/tests/cts/multidevices/apfv6_test.py index e06af65339..6b71c48bd3 100644 --- a/tests/cts/multidevices/apfv6_test.py +++ b/tests/cts/multidevices/apfv6_test.py @@ -13,198 +13,218 @@ # limitations under the License. from mobly import asserts -from scapy.layers.inet import IP, ICMP, IPOption_Router_Alert +from net_tests_utils.host.python import adb_utils, apf_test_base, apf_utils, assert_utils +from scapy.contrib.igmpv3 import IGMPv3, IGMPv3gr, IGMPv3mq, IGMPv3mr +from scapy.layers.inet import ICMP, IP, IPOption_Router_Alert from scapy.layers.inet6 import ( - IPv6, - IPv6ExtHdrHopByHop, - ICMPv6EchoRequest, ICMPv6EchoReply, + ICMPv6EchoRequest, + ICMPv6MLDMultAddrRec, ICMPv6MLQuery2, ICMPv6MLReport2, - ICMPv6MLDMultAddrRec, - ICMPv6NDOptSrcLLAddr, ICMPv6NDOptDstLLAddr, - ICMPv6ND_NS, + ICMPv6NDOptSrcLLAddr, ICMPv6ND_NA, - RouterAlert + ICMPv6ND_NS, + IPv6, + IPv6ExtHdrHopByHop, + RouterAlert, ) from scapy.layers.l2 import ARP, Ether -from scapy.contrib.igmpv3 import IGMPv3, IGMPv3mq, IGMPv3mr, IGMPv3gr -from net_tests_utils.host.python import apf_test_base, apf_utils, adb_utils, assert_utils APFV6_VERSION = 6000 ARP_OFFLOAD_REPLY_LEN = 60 + class ApfV6Test(apf_test_base.ApfTestBase): - def setup_class(self): - super().setup_class() - # Skip tests for APF version < 6000 - apf_utils.assume_apf_version_support_at_least( - self.clientDevice, self.client_iface_name, APFV6_VERSION + def setup_class(self): + super().setup_class() + + # Skip tests for APF version < 6000 + apf_utils.assume_apf_version_support_at_least( + self.clientDevice, self.client_iface_name, APFV6_VERSION + ) + + def teardown_class(self): + # force to stop capture on the server device if any test case failed + try: + apf_utils.stop_capture_packets(self.serverDevice, self.server_iface_name) + except assert_utils.UnexpectedBehaviorError: + pass + super().teardown_class() + + def test_unicast_arp_request_offload(self): + eth = Ether(src=self.server_mac_address, dst=self.client_mac_address) + arp = ARP( + op=1, + psrc=self.server_ipv4_addresses[0], + pdst=self.client_ipv4_addresses[0], + hwsrc=self.server_mac_address, + ) + arp_request = bytes(eth / arp).hex() + + eth = Ether(src=self.client_mac_address, dst=self.server_mac_address) + arp = ARP( + op=2, + psrc=self.client_ipv4_addresses[0], + pdst=self.server_ipv4_addresses[0], + hwsrc=self.client_mac_address, + hwdst=self.server_mac_address, + ) + expected_arp_reply = bytes(eth / arp).hex() + + # Add zero padding up to 60 bytes, since APFv6 ARP offload always sent out 60 bytes reply + expected_arp_reply = expected_arp_reply.ljust( + ARP_OFFLOAD_REPLY_LEN * 2, '0' + ) + + self.send_packet_and_expect_reply_received( + arp_request, 'DROPPED_ARP_REQUEST_REPLIED', expected_arp_reply + ) + + def test_non_dad_ipv6_neighbor_solicitation_offload(self): + eth = Ether(src=self.server_mac_address, dst=self.client_mac_address) + ip = IPv6( + src=self.server_ipv6_addresses[0], dst=self.client_ipv6_addresses[0] + ) + icmpv6 = ICMPv6ND_NS(tgt=self.client_ipv6_addresses[0]) + opt = ICMPv6NDOptSrcLLAddr(lladdr=self.server_mac_address) + neighbor_solicitation = bytes(eth / ip / icmpv6 / opt).hex() + + eth = Ether(src=self.client_mac_address, dst=self.server_mac_address) + ip = IPv6( + src=self.client_ipv6_addresses[0], dst=self.server_ipv6_addresses[0] + ) + icmpv6 = ICMPv6ND_NA(tgt=self.client_ipv6_addresses[0], R=1, S=1, O=1) + opt = ICMPv6NDOptDstLLAddr(lladdr=self.client_mac_address) + expected_neighbor_advertisement = bytes(eth / ip / icmpv6 / opt).hex() + self.send_packet_and_expect_reply_received( + neighbor_solicitation, + 'DROPPED_IPV6_NS_REPLIED_NON_DAD', + expected_neighbor_advertisement, + ) + + @apf_utils.at_least_B() + def test_ipv4_icmp_echo_request_offload(self): + eth = Ether(src=self.server_mac_address, dst=self.client_mac_address) + ip = IP( + src=self.server_ipv4_addresses[0], dst=self.client_ipv4_addresses[0] + ) + icmp = ICMP(id=1, seq=123) + echo_request = bytes(eth / ip / icmp / b'hello').hex() + + eth = Ether(src=self.client_mac_address, dst=self.server_mac_address) + ip = IP( + src=self.client_ipv4_addresses[0], dst=self.server_ipv4_addresses[0] + ) + icmp = ICMP(type=0, id=1, seq=123) + expected_echo_reply = bytes(eth / ip / icmp / b'hello').hex() + self.send_packet_and_expect_reply_received( + echo_request, 'DROPPED_IPV4_PING_REQUEST_REPLIED', expected_echo_reply + ) + + @apf_utils.at_least_B() + @apf_utils.apf_ram_at_least(3000) + def test_ipv6_icmp_echo_request_offload(self): + eth = Ether(src=self.server_mac_address, dst=self.client_mac_address) + ip = IPv6( + src=self.server_ipv6_addresses[0], dst=self.client_ipv6_addresses[0] + ) + icmp = ICMPv6EchoRequest(id=1, seq=123) + echo_request = bytes(eth / ip / icmp / b'hello').hex() + + eth = Ether(src=self.client_mac_address, dst=self.server_mac_address) + ip = IPv6( + src=self.client_ipv6_addresses[0], dst=self.server_ipv6_addresses[0] + ) + icmp = ICMPv6EchoReply(id=1, seq=123) + expected_echo_reply = bytes(eth / ip / icmp / b'hello').hex() + + self.send_packet_and_expect_reply_received( + echo_request, + 'DROPPED_IPV6_ICMP6_ECHO_REQUEST_REPLIED', + expected_echo_reply, + ) + + def is_running_as_root(self): + uid = adb_utils.adb_shell(self.clientDevice, 'id -u') + return uid == '0' + + @apf_utils.at_least_B() + def test_igmpv3_general_query_offload(self): + if not self.is_running_as_root(): + return + # use unicast to replace multicast ether dst to prevent flaky due to DTIM skip + ether = Ether(src=self.server_mac_address, dst=self.client_mac_address) + ip = IP( + src=self.server_ipv4_addresses[0], + dst='224.0.0.1', + options=[IPOption_Router_Alert()], + ) + igmp = IGMPv3(type=0x11) / IGMPv3mq() + igmpv3_general_query = bytes(ether / ip / igmp).hex() + + mcast_addrs = ['239.0.0.1', '239.0.0.2', '239.0.0.3'] + + for addr in mcast_addrs: + adb_utils.adb_shell( + self.clientDevice, + f'ip addr add {addr}/32 dev {self.client_iface_name} autojoin', + ) + + ether = Ether(src=self.client_mac_address, dst='01:00:5e:00:00:16') + ip = IP( + src=self.client_ipv4_addresses[0], + dst='224.0.0.22', + options=[IPOption_Router_Alert()], + id=0, + flags='DF', + ) + igmpv3_hdr = IGMPv3(type=0x22) + mcast_records = [] + for addr in mcast_addrs: + mcast_records.append(IGMPv3gr(rtype=2, maddr=addr)) + + igmp = IGMPv3mr(records=mcast_records) + expected_igmpv3_report = bytes(ether / ip / igmpv3_hdr / igmp).hex() + try: + self.send_packet_and_expect_reply_received( + igmpv3_general_query, + 'DROPPED_IGMP_V3_GENERAL_QUERY_REPLIED', + expected_igmpv3_report, + ) + finally: + for addr in mcast_addrs: + adb_utils.adb_shell( + self.clientDevice, + f'ip addr del {addr}/32 dev {self.client_iface_name}', ) - def teardown_class(self): - # force to stop capture on the server device if any test case failed - try: - apf_utils.stop_capture_packets(self.serverDevice, self.server_iface_name) - except assert_utils.UnexpectedBehaviorError: - pass - super().teardown_class() + @apf_utils.at_least_B() + @apf_utils.apf_ram_at_least(3000) + def test_mldv2_general_query_offload(self): + # use unicast to replace multicast ether dst to prevent flaky due to DTIM skip + ether = Ether(src=self.server_mac_address, dst=self.client_mac_address) + ip = IPv6(src=self.server_ipv6_addresses[0], dst='ff02::1', hlim=1) + hopOpts = IPv6ExtHdrHopByHop(options=[RouterAlert(otype=5)]) + mld = ICMPv6MLQuery2() + mldv2_general_query = bytes(ether / ip / hopOpts / mld).hex() - def test_unicast_arp_request_offload(self): - eth = Ether(src=self.server_mac_address, dst=self.client_mac_address) - arp = ARP( - op=1, - psrc=self.server_ipv4_addresses[0], - pdst=self.client_ipv4_addresses[0], - hwsrc=self.server_mac_address - ) - arp_request = bytes(eth/arp).hex() + ether = Ether(src=self.client_mac_address, dst='33:33:00:00:00:16') + ip = IPv6(src=self.client_ipv6_addresses[0], dst='ff02::16', hlim=1) - eth = Ether(src=self.client_mac_address, dst=self.server_mac_address) - arp = ARP( - op=2, - psrc=self.client_ipv4_addresses[0], - pdst=self.server_ipv4_addresses[0], - hwsrc=self.client_mac_address, - hwdst=self.server_mac_address - ) - expected_arp_reply = bytes(eth/arp).hex() + mcast_addrs = apf_utils.get_exclude_all_host_ipv6_multicast_addresses( + self.clientDevice, self.client_iface_name + ) - # Add zero padding up to 60 bytes, since APFv6 ARP offload always sent out 60 bytes reply - expected_arp_reply = expected_arp_reply.ljust(ARP_OFFLOAD_REPLY_LEN * 2, "0") - - self.send_packet_and_expect_reply_received( - arp_request, "DROPPED_ARP_REQUEST_REPLIED", expected_arp_reply - ) - - def test_non_dad_ipv6_neighbor_solicitation_offload(self): - eth = Ether(src=self.server_mac_address, dst=self.client_mac_address) - ip = IPv6(src=self.server_ipv6_addresses[0], dst=self.client_ipv6_addresses[0]) - icmpv6 = ICMPv6ND_NS(tgt=self.client_ipv6_addresses[0]) - opt = ICMPv6NDOptSrcLLAddr(lladdr=self.server_mac_address) - neighbor_solicitation = bytes(eth/ip/icmpv6/opt).hex() - - eth = Ether(src=self.client_mac_address, dst=self.server_mac_address) - ip = IPv6(src=self.client_ipv6_addresses[0], dst=self.server_ipv6_addresses[0]) - icmpv6 = ICMPv6ND_NA(tgt=self.client_ipv6_addresses[0], R=1, S=1, O=1) - opt = ICMPv6NDOptDstLLAddr(lladdr=self.client_mac_address) - expected_neighbor_advertisement = bytes(eth/ip/icmpv6/opt).hex() - self.send_packet_and_expect_reply_received( - neighbor_solicitation, - "DROPPED_IPV6_NS_REPLIED_NON_DAD", - expected_neighbor_advertisement - ) - - @apf_utils.at_least_B() - def test_ipv4_icmp_echo_request_offload(self): - eth = Ether(src=self.server_mac_address, dst=self.client_mac_address) - ip = IP(src=self.server_ipv4_addresses[0], dst=self.client_ipv4_addresses[0]) - icmp = ICMP(id=1, seq=123) - echo_request = bytes(eth/ip/icmp/b"hello").hex() - - eth = Ether(src=self.client_mac_address, dst=self.server_mac_address) - ip = IP(src=self.client_ipv4_addresses[0], dst=self.server_ipv4_addresses[0]) - icmp = ICMP(type=0, id=1, seq=123) - expected_echo_reply = bytes(eth/ip/icmp/b"hello").hex() - self.send_packet_and_expect_reply_received( - echo_request, "DROPPED_IPV4_PING_REQUEST_REPLIED", expected_echo_reply - ) - - @apf_utils.at_least_B() - @apf_utils.apf_ram_at_least(3000) - def test_ipv6_icmp_echo_request_offload(self): - eth = Ether(src=self.server_mac_address, dst=self.client_mac_address) - ip = IPv6(src=self.server_ipv6_addresses[0], dst=self.client_ipv6_addresses[0]) - icmp = ICMPv6EchoRequest(id=1, seq=123) - echo_request = bytes(eth/ip/icmp/b"hello").hex() - - eth = Ether(src=self.client_mac_address, dst=self.server_mac_address) - ip = IPv6(src=self.client_ipv6_addresses[0], dst=self.server_ipv6_addresses[0]) - icmp = ICMPv6EchoReply(id=1, seq=123) - expected_echo_reply = bytes(eth/ip/icmp/b"hello").hex() - - self.send_packet_and_expect_reply_received( - echo_request, "DROPPED_IPV6_ICMP6_ECHO_REQUEST_REPLIED", expected_echo_reply - ) - - def is_running_as_root(self): - uid = adb_utils.adb_shell(self.clientDevice, "id -u") - return uid == "0" - - @apf_utils.at_least_B() - def test_igmpv3_general_query_offload(self): - if not self.is_running_as_root(): - return - # use unicast to replace multicast ether dst to prevent flaky due to DTIM skip - ether = Ether(src=self.server_mac_address, dst=self.client_mac_address) - ip = IP( - src=self.server_ipv4_addresses[0], - dst='224.0.0.1', - options=[IPOption_Router_Alert()] - ) - igmp = IGMPv3(type=0x11)/IGMPv3mq() - igmpv3_general_query = bytes(ether/ip/igmp).hex() - - mcast_addrs = ['239.0.0.1', '239.0.0.2', '239.0.0.3'] - - for addr in mcast_addrs: - adb_utils.adb_shell( - self.clientDevice, - f'ip addr add {addr}/32 dev {self.client_iface_name} autojoin' - ) - - ether = Ether(src=self.client_mac_address, dst='01:00:5e:00:00:16') - ip = IP( - src=self.client_ipv4_addresses[0], - dst='224.0.0.22', - options=[IPOption_Router_Alert()], - id=0, - flags="DF" - ) - igmpv3_hdr = IGMPv3(type=0x22) - mcast_records = [] - for addr in mcast_addrs: - mcast_records.append(IGMPv3gr(rtype=2, maddr=addr)) - - igmp = IGMPv3mr(records=mcast_records) - expected_igmpv3_report = bytes(ether/ip/igmpv3_hdr/igmp).hex() - try: - self.send_packet_and_expect_reply_received( - igmpv3_general_query, - "DROPPED_IGMP_V3_GENERAL_QUERY_REPLIED", - expected_igmpv3_report - ) - finally: - for addr in mcast_addrs: - adb_utils.adb_shell( - self.clientDevice, - f'ip addr del {addr}/32 dev {self.client_iface_name}' - ) - - @apf_utils.at_least_B() - @apf_utils.apf_ram_at_least(3000) - def test_mldv2_general_query_offload(self): - # use unicast to replace multicast ether dst to prevent flaky due to DTIM skip - ether = Ether(src=self.server_mac_address, dst=self.client_mac_address) - ip = IPv6(src=self.server_ipv6_addresses[0], dst='ff02::1', hlim=1) - hopOpts = IPv6ExtHdrHopByHop(options=[RouterAlert(otype=5)]) - mld = ICMPv6MLQuery2() - mldv2_general_query = bytes(ether/ip/hopOpts/mld).hex() - - ether = Ether(src=self.client_mac_address, dst='33:33:00:00:00:16') - ip = IPv6(src=self.client_ipv6_addresses[0], dst='ff02::16', hlim=1) - - mcast_addrs = apf_utils.get_exclude_all_host_ipv6_multicast_addresses( - self.clientDevice, self.client_iface_name - ) - - mld_records = [] - for addr in mcast_addrs: - mld_records.append(ICMPv6MLDMultAddrRec(dst=addr, rtype=2)) - mld = ICMPv6MLReport2(records=mld_records) - expected_mldv2_report = bytes(ether/ip/hopOpts/mld).hex() - self.send_packet_and_expect_reply_received( - mldv2_general_query, "DROPPED_IPV6_MLD_V2_GENERAL_QUERY_REPLIED", expected_mldv2_report - ) + mld_records = [] + for addr in mcast_addrs: + mld_records.append(ICMPv6MLDMultAddrRec(dst=addr, rtype=2)) + mld = ICMPv6MLReport2(records=mld_records) + expected_mldv2_report = bytes(ether / ip / hopOpts / mld).hex() + self.send_packet_and_expect_reply_received( + mldv2_general_query, + 'DROPPED_IPV6_MLD_V2_GENERAL_QUERY_REPLIED', + expected_mldv2_report, + ) diff --git a/tests/cts/multidevices/run_tests.py b/tests/cts/multidevices/run_tests.py index a0d0bec587..3b4afc8238 100644 --- a/tests/cts/multidevices/run_tests.py +++ b/tests/cts/multidevices/run_tests.py @@ -21,7 +21,7 @@ from connectivity_multi_devices_test import ConnectivityMultiDevicesTest from mobly import suite_runner -if __name__ == "__main__": +if __name__ == '__main__': # For MoblyBinaryHostTest, this entry point will be called twice: # 1. List tests. # -- --list_tests @@ -32,8 +32,10 @@ if __name__ == "__main__": # While the parameters before "--" is for the infrastructure, # ignore them if any. Also, do not alter parameters if there # is no "--", in case the binary is invoked manually. - if "--" in sys.argv: - index = sys.argv.index("--") + if '--' in sys.argv: + index = sys.argv.index('--') sys.argv = sys.argv[:1] + sys.argv[index + 1 :] # TODO: make the tests can be executed without manually list classes. - suite_runner.run_suite([ConnectivityMultiDevicesTest, ApfV4Test, ApfV6Test], sys.argv) + suite_runner.run_suite( + [ConnectivityMultiDevicesTest, ApfV4Test, ApfV6Test], sys.argv + )