diff --git a/staticlibs/tests/unit/host/python/apf_utils_test.py b/staticlibs/tests/unit/host/python/apf_utils_test.py index 55fbe58c7a..b0a70ae40b 100644 --- a/staticlibs/tests/unit/host/python/apf_utils_test.py +++ b/staticlibs/tests/unit/host/python/apf_utils_test.py @@ -24,13 +24,14 @@ from net_tests_utils.host.python.apf_utils import ( UnsupportedOperationException, get_apf_capabilities, get_apf_counter, - get_apf_counters_from_dumpsys, + 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, + is_apf_dump_counters_supported, start_capture_packets, stop_capture_packets, get_matched_packet_counts, @@ -50,53 +51,24 @@ 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.adb_utils.get_dumpsys_for_service") - def test_get_apf_counters_from_dumpsys_success( - self, mock_get_dumpsys: MagicMock - ) -> None: - mock_get_dumpsys.return_value = """ -IpClient.wlan0 - APF packet counters: - COUNTER_NAME1: 123 - COUNTER_NAME2: 456 -""" - counters = get_apf_counters_from_dumpsys(self.mock_ad, "wlan0") - asserts.assert_equal(counters, {"COUNTER_NAME1": 123, "COUNTER_NAME2": 456}) - - @patch("net_tests_utils.host.python.adb_utils.get_dumpsys_for_service") - def test_get_apf_counters_from_dumpsys_exceptions( - self, mock_get_dumpsys: MagicMock - ) -> None: - test_cases = [ - "", - "IpClient.wlan0\n", - "IpClient.wlan0\n APF packet counters:\n", - """ -IpClient.wlan1 - APF packet counters: - COUNTER_NAME1: 123 - COUNTER_NAME2: 456 -""", - ] - - for dumpsys_output in test_cases: - mock_get_dumpsys.return_value = dumpsys_output - with asserts.assert_raises(PatternNotFoundException): - get_apf_counters_from_dumpsys(self.mock_ad, "wlan0") - - @patch("net_tests_utils.host.python.apf_utils.get_apf_counters_from_dumpsys") + @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" mock_get_counters.return_value = { - "COUNTER_NAME1": 123, - "COUNTER_NAME2": 456, + "COUNTER_NAME1": "123", + "COUNTER_NAME2": "456", + "COUNTER_NAME3": "[ERROR: impossible]", } asserts.assert_equal( 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 + ) # Not found asserts.assert_equal( - get_apf_counter(self.mock_ad, iface, "COUNTER_NAME3"), 0 + get_apf_counter(self.mock_ad, iface, "COUNTER_NAME4"), 0 ) @patch("net_tests_utils.host.python.adb_utils.adb_shell") @@ -380,6 +352,74 @@ IpClient.wlan1 "Get matched packet counts should not be supported.", ) + @patch("net_tests_utils.host.python.adb_utils.adb_shell") + def test_get_apf_counters_from_cmd_success( + self, mock_adb_shell: MagicMock + ) -> None: + mock_adb_shell.return_value = """ + NDIANNESS: 305419896 + TOTAL_PACKETS: 157553 + FILTER_AGE_SECONDS: 356 + FILTER_AGE_16384THS: 5836719 + APF_VERSION: 6000 + APF_PROGRAM_ID: 25 + PASSED_ARP_UNICAST_REPLY: 7 + PASSED_IPV4: 14 + PASSED_IPV4_UNICAST: 153472 + PASSED_IPV6_ICMP: 1 + PASSED_IPV6_NON_ICMP: 2764 + PASSED_IPV6_UNICAST_NON_ICMP: 970 + PASSED_RA: 6 + DROPPED_RA: 2 + DROPPED_IPV4_BROADCAST_ADDR: 48 + DROPPED_IPV4_BROADCAST_NET: 63 + DROPPED_IPV4_MULTICAST: 30 + DROPPED_IPV6_NS_OTHER_HOST: 6 + DROPPED_IPV6_NS_REPLIED_NON_DAD: 9 + DROPPED_ETHERTYPE_NOT_ALLOWED: 58 + 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" + ) + + @patch("net_tests_utils.host.python.adb_utils.adb_shell") + def test_get_apf_counters_from_cmd_failure( + 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.", + ) + + @patch("net_tests_utils.host.python.adb_utils.adb_shell") + def test_get_apf_counters_from_cmd_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): + 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 diff --git a/staticlibs/testutils/host/python/apf_utils.py b/staticlibs/testutils/host/python/apf_utils.py index 4835c23c56..1f91a5e715 100644 --- a/staticlibs/testutils/host/python/apf_utils.py +++ b/staticlibs/testutils/host/python/apf_utils.py @@ -32,57 +32,28 @@ class UnsupportedOperationException(Exception): def get_apf_counter( ad: android_device.AndroidDevice, iface: str, counter_name: str ) -> int: - counters = get_apf_counters_from_dumpsys(ad, iface) - return counters.get(counter_name, 0) + counters = get_apf_counters_from_cmd(ad, iface) + try: + return int(counters.get(counter_name, 0)) + except ValueError: + return -1 - -def get_apf_counters_from_dumpsys( +def get_apf_counters_from_cmd( ad: android_device.AndroidDevice, iface_name: str ) -> dict: - dumpsys = adb_utils.get_dumpsys_for_service(ad, "network_stack") + cmd = f'cmd network_stack apf {iface_name} dump-counters' + output = AdbOutputHandler(ad, cmd).get_output() + data_dict = {} + for line in output.strip().split('\n'): + 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}.") - # Extract IpClient section of the specified interface. - # This takes inputs like: - # IpClient.wlan0 - # ... - # IpClient.wlan1 - # ... - iface_pattern = re.compile( - r"^IpClient\." + iface_name + r"\n" + r"((^\s.*\n)+)", re.MULTILINE - ) - iface_result = iface_pattern.search(dumpsys) - if iface_result is None: - raise PatternNotFoundException("Cannot find IpClient for " + iface_name) - - # Extract APF counters section from IpClient section, which looks like: - # APF packet counters: - # COUNTER_NAME: VALUE - # .... - apf_pattern = re.compile( - r"APF packet counters:.*\n.(\s+[A-Z_0-9]+: \d+\n)+", re.MULTILINE - ) - apf_result = apf_pattern.search(iface_result.group(0)) - if apf_result is None: - raise PatternNotFoundException( - "Cannot find APF counters in text: " + iface_result.group(0) - ) - - # Extract key-value pairs from APF counters section into a list of tuples, - # e.g. [('COUNTER1', '1'), ('COUNTER2', '2')]. - counter_pattern = re.compile(r"(?P[A-Z_0-9]+): (?P\d+)") - counter_result = counter_pattern.findall(apf_result.group(0)) - if counter_result is None: - raise PatternNotFoundException( - "Cannot extract APF counters in text: " + apf_result.group(0) - ) - - # Convert into a dict. - result = {} - for key, value_str in counter_result: - result[key] = int(value_str) - - ad.log.debug("Getting apf counters: " + str(result)) - return result + return data_dict def get_ipv4_addresses( ad: android_device.AndroidDevice, iface_name: str @@ -240,6 +211,22 @@ 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: + 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 + ) + except assert_utils.UnexpectedExceptionError: + return False + + # If no UnsupportOperationException is thrown, regard it as supported + return True + def is_send_raw_packet_downstream_supported( ad: android_device.AndroidDevice, ) -> bool: