Merge changes I65fecdcd,I0acc9fcb into main

* changes:
  Add pre-test check for APF offload status
  Add util to get APF offload config

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

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
+        )
+    )

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]:

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,

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)