From 7c190ecb7860736f26a92c3ba12befd1b0425a39 Mon Sep 17 00:00:00 2001 From: Jimi Chen Date: Tue, 8 Jul 2025 02:51:29 +0000 Subject: [PATCH 1/2] Add util to get APF offload config Adds a new helper function `get_apf_config_from_cmd` to retrieve the list of enabled APF offloads for a given network interface. Flag: EXEMPT mainline Bug: 419427388 Test: atest NetworkStaticLibHostPythonTests Change-Id: I0acc9fcbf2e8bc6f4d8f309c2ec64c003c325c42 --- .../tests/unit/host/python/apf_utils_test.py | 37 +++++++++++++++++++ staticlibs/testutils/host/python/apf_utils.py | 21 +++++++++++ 2 files changed, 58 insertions(+) diff --git a/staticlibs/tests/unit/host/python/apf_utils_test.py b/staticlibs/tests/unit/host/python/apf_utils_test.py index 6bf7a962a5..ffe6f117e7 100644 --- a/staticlibs/tests/unit/host/python/apf_utils_test.py +++ b/staticlibs/tests/unit/host/python/apf_utils_test.py @@ -23,6 +23,7 @@ from net_tests_utils.host.python.apf_utils import ( PatternNotFoundException, UnsupportedOperationException, get_apf_capabilities, + get_apf_config_from_cmd, get_apf_counter, get_apf_counters_from_cmd, get_exclude_all_host_ipv6_multicast_addresses, @@ -386,6 +387,42 @@ class TestApfUtils(base_test.BaseTestClass, parameterized.TestCase): 'Dump APF packet counters should not be supported.', ) + @patch('net_tests_utils.host.python.adb_utils.adb_shell') + def test_get_apf_config_from_cmd_success( + self, mock_adb_shell: MagicMock + ) -> None: + """Tests successful parsing of APF config.""" + mock_adb_shell.return_value = ( + '{ mcast: DROP, doze: FALSE, offloads: [ ARP, ND, MLD, ] total RAs: 2' + ' filtered RAs: 2 mDNSs: 0 }' + ) + config = get_apf_config_from_cmd(self.mock_ad, TEST_IFACE_NAME) + asserts.assert_equal(config, ['ARP', 'ND', 'MLD']) + mock_adb_shell.assert_called_once_with( + self.mock_ad, f'cmd network_stack apf {TEST_IFACE_NAME} config' + ) + + @patch('net_tests_utils.host.python.adb_utils.adb_shell') + def test_get_apf_config_from_cmd_failure_returns_empty( + self, mock_adb_shell: MagicMock + ) -> None: + """Tests that unexpected output results in an empty list.""" + # This function is designed to return an empty list on parsing failure. + mock_adb_shell.return_value = 'some unexpected output' + config = get_apf_config_from_cmd(self.mock_ad, TEST_IFACE_NAME) + asserts.assert_equal(config, []) + + @patch('net_tests_utils.host.python.adb_utils.adb_shell') + def test_get_apf_config_from_cmd_unsupported( + self, mock_adb_shell: MagicMock + ) -> None: + """Tests the case where the command is not supported by the device.""" + mock_adb_shell.side_effect = AdbError( + cmd='', stdout='Unknown command', stderr='', ret_code=1 + ) + with asserts.assert_raises(UnsupportedOperationException): + get_apf_config_from_cmd(self.mock_ad, TEST_IFACE_NAME) + @parameterized.parameters( ('2,2048,1', ApfCapabilities(2, 2048, 1)), # Valid input ('3,1024,0', ApfCapabilities(3, 1024, 0)), # Valid input diff --git a/staticlibs/testutils/host/python/apf_utils.py b/staticlibs/testutils/host/python/apf_utils.py index 3a1c8f5e3e..23cbb1e48d 100644 --- a/staticlibs/testutils/host/python/apf_utils.py +++ b/staticlibs/testutils/host/python/apf_utils.py @@ -59,6 +59,27 @@ def get_apf_counters_from_cmd( return data_dict +def get_apf_config_from_cmd( + ad: android_device.AndroidDevice, iface_name: str +) -> list[str]: + cmd = f'cmd network_stack apf {iface_name} config' + output = AdbOutputHandler(ad, cmd).get_output() + if not output or '\n' in output.strip(): + return [] + + match = re.search(r'offloads:\s*\[(.*?)\]', output) + if not match: + return [] + + # The captured group contains the list items (e.g., " ARP, ND, MLD, ... ") + list_content = match.group(1) + + # Split the content string by commas or spaces, filter empty items, + # and return the result as a list. + items = [item for item in re.split(r'[,\s]+', list_content) if item] + return items + + def get_ipv4_addresses( ad: android_device.AndroidDevice, iface_name: str ) -> list[str]: From af1e460e26f8e73348f75f5ddead8bb29c732125 Mon Sep 17 00:00:00 2001 From: Jimi Chen Date: Tue, 8 Jul 2025 04:09:49 +0000 Subject: [PATCH 2/2] Add pre-test check for APF offload status This commit introduces a verification step that runs before APF offload tests. It uses the `cmd network_stack apf config` shell command to dump the list of enabled offloads. By checking this list first, we can ensure that the specific offload under test is actually enabled, avoiding misleading failures and improving test reliability. Flag: EXEMPT mainline Bug: 419427388 Test: ./cts-tradefed run commandAndExit cts -m CtsConnectivityMultiDevicesTestCases -t ApfV6Test --shard-count 2 Test: atest CtsConnectivityMultiDevicesTestCases:ApfV6Test Change-Id: I65fecdcd4f2f9e9586da2684bf68cec215d0f359 --- .../testutils/host/python/apf_test_base.py | 14 +++++--- tests/cts/multidevices/apfv4_test.py | 3 ++ tests/cts/multidevices/apfv6_test.py | 36 +++++++++++++++---- 3 files changed, 43 insertions(+), 10 deletions(-) diff --git a/staticlibs/testutils/host/python/apf_test_base.py b/staticlibs/testutils/host/python/apf_test_base.py index b0fe28fdf3..1ccf0ab423 100644 --- a/staticlibs/testutils/host/python/apf_test_base.py +++ b/staticlibs/testutils/host/python/apf_test_base.py @@ -12,11 +12,12 @@ # 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 +APF_ACTIVATION_WAIT_TIME_SEC = 5 + class ApfTestBase(multi_devices_test_base.MultiDevicesTestBase): @@ -76,9 +77,6 @@ class ApfTestBase(multi_devices_test_base.MultiDevicesTestBase): # Enable doze mode to activate APF. adb_utils.set_doze_mode(self.clientDevice, True) - # Longer wait time is required for APF to become active in CTS test suite. - time.sleep(5) - def teardown_class(self): adb_utils.set_doze_mode(self.clientDevice, False) tether_utils.cleanup_tethering_for_upstream_type( @@ -142,3 +140,11 @@ class ApfTestBase(multi_devices_test_base.MultiDevicesTestBase): finally: apf_utils.stop_capture_packets(self.serverDevice, self.server_iface_name) + + def expect_apf_offload_enabled(self, offload: str): + assert_utils.expect_with_retry( + lambda: offload + in apf_utils.get_apf_config_from_cmd( + self.clientDevice, self.client_iface_name + ) + ) diff --git a/tests/cts/multidevices/apfv4_test.py b/tests/cts/multidevices/apfv4_test.py index 6071d40769..dc31703ff1 100644 --- a/tests/cts/multidevices/apfv4_test.py +++ b/tests/cts/multidevices/apfv4_test.py @@ -44,6 +44,9 @@ class ApfV4Test(apf_test_base.ApfTestBase, parameterized.TestCase): self.clientDevice, self.client_iface_name, 4 ) + # Longer wait time is required for APF to become active in CTS test suite. + time.sleep(apf_test_base.APF_ACTIVATION_WAIT_TIME_SEC) + # APF L2 packet filtering on V+ Android allows only specific # types: IPv4, ARP, IPv6, EAPOL, WAPI. # Tests can use any disallowed packet type. Currently, diff --git a/tests/cts/multidevices/apfv6_test.py b/tests/cts/multidevices/apfv6_test.py index 4512c2d3fe..dc68ee8e13 100644 --- a/tests/cts/multidevices/apfv6_test.py +++ b/tests/cts/multidevices/apfv6_test.py @@ -12,6 +12,7 @@ # 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_test_base, apf_utils, assert_utils from scapy.contrib.igmpv3 import IGMPv3, IGMPv3gr, IGMPv3mq, IGMPv3mr @@ -55,6 +56,10 @@ class ApfV6Test(apf_test_base.ApfTestBase): super().teardown_class() def test_unicast_arp_request_offload(self): + # Longer wait time is required for APF to become active in CTS test suite. + time.sleep(apf_test_base.APF_ACTIVATION_WAIT_TIME_SEC) + self.expect_apf_offload_enabled('ARP') + eth = Ether(src=self.server_mac_address, dst=self.client_mac_address) arp = ARP( op=1, @@ -84,6 +89,10 @@ class ApfV6Test(apf_test_base.ApfTestBase): ) def test_non_dad_ipv6_neighbor_solicitation_offload(self): + # Longer wait time is required for APF to become active in CTS test suite. + time.sleep(apf_test_base.APF_ACTIVATION_WAIT_TIME_SEC) + self.expect_apf_offload_enabled('ND') + 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] @@ -107,6 +116,10 @@ class ApfV6Test(apf_test_base.ApfTestBase): @apf_utils.at_least_B() def test_ipv4_icmp_echo_request_offload(self): + # Longer wait time is required for APF to become active in CTS test suite. + time.sleep(apf_test_base.APF_ACTIVATION_WAIT_TIME_SEC) + self.expect_apf_offload_enabled('Ping4') + 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] @@ -127,6 +140,10 @@ class ApfV6Test(apf_test_base.ApfTestBase): @apf_utils.at_least_B() @apf_utils.apf_ram_at_least(3000) def test_ipv6_icmp_echo_request_offload(self): + # Longer wait time is required for APF to become active in CTS test suite. + time.sleep(apf_test_base.APF_ACTIVATION_WAIT_TIME_SEC) + self.expect_apf_offload_enabled('Ping6') + 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] @@ -149,6 +166,15 @@ class ApfV6Test(apf_test_base.ApfTestBase): @apf_utils.at_least_B() def test_igmpv3_general_query_offload(self): + mcast_addrs = ['239.0.0.1', '239.0.0.2', '239.0.0.3'] + + self.client.createMulticastSocket(self.client_iface_name) + for addr in mcast_addrs: + self.client.joinMulticastGroup(addr) + + # Longer wait time is required for APF to become active in CTS test suite. + time.sleep(apf_test_base.APF_ACTIVATION_WAIT_TIME_SEC) + self.expect_apf_offload_enabled('IGMP') # 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( @@ -159,12 +185,6 @@ class ApfV6Test(apf_test_base.ApfTestBase): 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'] - - self.client.createMulticastSocket(self.client_iface_name) - for addr in mcast_addrs: - self.client.joinMulticastGroup(addr) - ether = Ether(src=self.client_mac_address, dst='01:00:5e:00:00:16') ip = IP( src=self.client_ipv4_addresses[0], @@ -194,6 +214,10 @@ class ApfV6Test(apf_test_base.ApfTestBase): @apf_utils.at_least_B() @apf_utils.apf_ram_at_least(3000) def test_mldv2_general_query_offload(self): + # Longer wait time is required for APF to become active in CTS test suite. + time.sleep(apf_test_base.APF_ACTIVATION_WAIT_TIME_SEC) + self.expect_apf_offload_enabled('MLD') + # 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)