Add packet capture and match network_stack adb command to to apf_utils
It wraps the following 3 commands for multi-devices tests - cmd network_stack capture start <interface> - cmd network_stack capture stop <interface> - cmd network_stack capture matched-packet-counts <interface> <pkt-in-hex> Add a helper function `is_packet_capture_supported` as APFv6 precondition check. Test: atest NetworkStaticLibHostPythonTests Change-Id: I526c678c6e753cee3969b04a56ab33e5e3d1600d
This commit is contained in:
parent
738bf999d5
commit
1644f95852
1 changed files with 262 additions and 14 deletions
|
|
@ -178,6 +178,29 @@ def get_hardware_address(
|
|||
"Cannot get hardware address for " + iface_name
|
||||
)
|
||||
|
||||
def is_packet_capture_supported(
|
||||
ad: android_device.AndroidDevice,
|
||||
) -> bool:
|
||||
|
||||
# Invoke the shell command with empty argument and see how NetworkStack respond.
|
||||
# If supported, an IllegalArgumentException with help page will be printed.
|
||||
functions_with_args = (
|
||||
# list all functions and args with (func, *args) tuple
|
||||
(start_capture_packets, (ad, "")),
|
||||
(stop_capture_packets, (ad, "")),
|
||||
(get_matched_packet_counts, (ad, "", ""))
|
||||
)
|
||||
|
||||
for func, args in functions_with_args:
|
||||
try:
|
||||
func(*args)
|
||||
except UnsupportedOperationException:
|
||||
return False
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
# If no UnsupportOperationException is thrown, regard it as supported
|
||||
return True
|
||||
|
||||
def is_send_raw_packet_downstream_supported(
|
||||
ad: android_device.AndroidDevice,
|
||||
|
|
@ -224,25 +247,92 @@ def send_raw_packet_downstream(
|
|||
representation of a packet starting from L2 header.
|
||||
"""
|
||||
|
||||
cmd = (
|
||||
"cmd network_stack send-raw-packet-downstream"
|
||||
f" {iface_name} {packet_in_hex}"
|
||||
)
|
||||
cmd = f"cmd network_stack send-raw-packet-downstream {iface_name} {packet_in_hex}"
|
||||
|
||||
# Expect no output or Unknown command if NetworkStack is too old. Throw otherwise.
|
||||
try:
|
||||
output = adb_utils.adb_shell(ad, cmd)
|
||||
except AdbError as e:
|
||||
output = str(e.stdout)
|
||||
if output:
|
||||
if "Unknown command" in output:
|
||||
raise UnsupportedOperationException(
|
||||
"send-raw-packet-downstream command is not supported."
|
||||
)
|
||||
adb_output = AdbOutputHandler(ad, cmd).get_output()
|
||||
if adb_output:
|
||||
raise assert_utils.UnexpectedBehaviorError(
|
||||
f"Got unexpected output: {output} for command: {cmd}."
|
||||
f"Got unexpected output: {adb_output} for command: {cmd}."
|
||||
)
|
||||
|
||||
def start_capture_packets(
|
||||
ad: android_device.AndroidDevice,
|
||||
iface_name: str
|
||||
) -> None:
|
||||
"""Starts packet capturing on a specified network interface.
|
||||
|
||||
This function initiates packet capture on the given network interface of an
|
||||
Android device using an ADB shell command. It handles potential errors
|
||||
related to unsupported commands or unexpected output.
|
||||
This command only supports downstream tethering interface.
|
||||
|
||||
Args:
|
||||
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}"
|
||||
|
||||
# Expect no output or Unknown command if NetworkStack is too old. Throw otherwise.
|
||||
adb_output = AdbOutputHandler(ad, cmd).get_output()
|
||||
if adb_output != "success":
|
||||
raise assert_utils.UnexpectedBehaviorError(
|
||||
f"Got unexpected output: {adb_output} for command: {cmd}."
|
||||
)
|
||||
|
||||
def stop_capture_packets(
|
||||
ad: android_device.AndroidDevice,
|
||||
iface_name: str
|
||||
) -> None:
|
||||
"""Stops packet capturing on a specified network interface.
|
||||
|
||||
This function terminates packet capture on the given network interface of an
|
||||
Android device using an ADB shell command. It handles potential errors
|
||||
related to unsupported commands or unexpected output.
|
||||
|
||||
Args:
|
||||
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}"
|
||||
|
||||
# Expect no output or Unknown command if NetworkStack is too old. Throw otherwise.
|
||||
adb_output = AdbOutputHandler(ad, cmd).get_output()
|
||||
if adb_output != "success":
|
||||
raise assert_utils.UnexpectedBehaviorError(
|
||||
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
|
||||
) -> int:
|
||||
"""Gets the number of captured packets matching a specific hexadecimal pattern.
|
||||
|
||||
This function retrieves the count of captured packets on the specified
|
||||
network interface that match a given hexadecimal pattern. It uses an ADB
|
||||
shell command and handles potential errors related to unsupported commands,
|
||||
unexpected output, or invalid output format.
|
||||
|
||||
Args:
|
||||
ad: The Android device object.
|
||||
iface_name: The name of the network interface (e.g., "wlan0").
|
||||
packet_in_hex: The hexadecimal string representing the packet pattern.
|
||||
|
||||
Returns:
|
||||
The number of matched packets as an integer.
|
||||
"""
|
||||
cmd = f"cmd network_stack capture matched-packet-counts {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()
|
||||
try:
|
||||
return int(adb_output)
|
||||
except ValueError as e:
|
||||
raise assert_utils.UnexpectedBehaviorError(
|
||||
f"Got unexpected exception: {e} for command: {cmd}."
|
||||
)
|
||||
|
||||
@dataclass
|
||||
class ApfCapabilities:
|
||||
|
|
@ -304,3 +394,19 @@ def assume_apf_version_support_at_least(
|
|||
f"Supported apf version {caps.apf_version_supported} < expected version"
|
||||
f" {expected_version}",
|
||||
)
|
||||
|
||||
class AdbOutputHandler:
|
||||
def __init__(self, ad, cmd):
|
||||
self._ad = ad
|
||||
self._cmd = cmd
|
||||
|
||||
def get_output(self) -> str:
|
||||
try:
|
||||
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
|
||||
Loading…
Reference in a new issue