packages/antlion/test_utils/net/net_test_utils.py

#!/usr/bin/env python3
#
# Copyright 2022 The Fuchsia Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import logging
import os

from antlion.controllers import adb
from antlion.controllers.adb_lib.error import AdbError
from antlion.test_utils.net import connectivity_const as cconst
from antlion.utils import start_standing_subprocess, stop_standing_subprocess

VPN_CONST = cconst.VpnProfile
VPN_TYPE = cconst.VpnProfileType
VPN_PARAMS = cconst.VpnReqParams
TCPDUMP_PATH = "/data/local/tmp/"
USB_CHARGE_MODE = "svc usb setFunctions"
USB_TETHERING_MODE = "svc usb setFunctions rndis"
ENABLE_HARDWARE_OFFLOAD = "settings put global tether_offload_disabled 0"
DISABLE_HARDWARE_OFFLOAD = "settings put global tether_offload_disabled 1"
DEVICE_IP_ADDRESS = "ip address"
LOCALHOST = "192.168.1.1"

# Time to wait for radio to up and running after reboot
WAIT_TIME_AFTER_REBOOT = 10

GCE_SSH = "gcloud compute ssh "
GCE_SCP = "gcloud compute scp "


def verify_ping_to_vpn_ip(ad, vpn_ping_addr):
    """Verify if IP behind VPN server is pingable.

    Ping should pass, if VPN is connected.
    Ping should fail, if VPN is disconnected.

    Args:
        ad: android device object
        vpn_ping_addr: target ping addr
    """
    ping_result = None
    pkt_loss = "100% packet loss"
    logging.info(f"Pinging: {vpn_ping_addr}")
    try:
        ping_result = ad.adb.shell(f"ping -c 3 -W 2 {vpn_ping_addr}")
    except AdbError:
        pass
    return ping_result and pkt_loss not in ping_result


def start_tcpdump(ad, test_name, interface="any"):
    """Start tcpdump on all interfaces.

    Args:
        ad: android device object.
        test_name: tcpdump file name will have this
    """
    ad.log.info("Starting tcpdump on all interfaces")
    ad.adb.shell("killall -9 tcpdump", ignore_status=True)
    ad.adb.shell(f"mkdir {TCPDUMP_PATH}", ignore_status=True)
    ad.adb.shell(f"rm -rf {TCPDUMP_PATH}/*", ignore_status=True)

    file_name = f"{TCPDUMP_PATH}/tcpdump_{ad.serial}_{test_name}.pcap"
    ad.log.info("tcpdump file is %s", file_name)
    cmd = f"adb -s {ad.serial} shell tcpdump -i {interface} -s0 -w {file_name}"
    try:
        return start_standing_subprocess(cmd, 5)
    except Exception:
        ad.log.exception(f"Could not start standing process {repr(cmd)}")

    return None


def stop_tcpdump(
    ad, proc, test_name, pull_dump=True, adb_pull_timeout=adb.DEFAULT_ADB_PULL_TIMEOUT
):
    """Stops tcpdump on any iface.

       Pulls the tcpdump file in the tcpdump dir if necessary.

    Args:
        ad: android device object.
        proc: need to know which pid to stop
        test_name: test name to save the tcpdump file
        pull_dump: pull tcpdump file or not
        adb_pull_timeout: timeout for adb_pull

    Returns:
      log_path of the tcpdump file
    """
    ad.log.info("Stopping and pulling tcpdump if any")
    if proc is None:
        return None
    try:
        stop_standing_subprocess(proc)
    except Exception as e:
        ad.log.warning(e)
    if pull_dump:
        log_path = os.path.join(ad.device_log_path, f"TCPDUMP_{ad.serial}")
        os.makedirs(log_path, exist_ok=True)
        ad.adb.pull(f"{TCPDUMP_PATH}/. {log_path}", timeout=adb_pull_timeout)
        ad.adb.shell(f"rm -rf {TCPDUMP_PATH}/*", ignore_status=True)
        file_name = f"tcpdump_{ad.serial}_{test_name}.pcap"
        return f"{log_path}/{file_name}"
    return None