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_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/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]: 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)