acts/framework/acts/utils.py

#!/usr/bin/env python3.4
#
#   Copyright 2016 - The Android Open Source Project
#
#   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 base64
import concurrent.futures
import datetime
import json
import functools
import logging
import os
import random
import re
import signal
import string
import subprocess
import time
import traceback
import zipfile

from acts.controllers import adb
from acts import tracelogger

# File name length is limited to 255 chars on some OS, so we need to make sure
# the file names we output fits within the limit.
from acts.libs.proc import job

MAX_FILENAME_LEN = 255


class ActsUtilsError(Exception):
    """Generic error raised for exceptions in ACTS utils."""


class NexusModelNames:
    # TODO(angli): This will be fixed later by angli.
    ONE = 'sprout'
    N5 = 'hammerhead'
    N5v2 = 'bullhead'
    N6 = 'shamu'
    N6v2 = 'angler'
    N6v3 = 'marlin'
    N5v3 = 'sailfish'


class DozeModeStatus:
    ACTIVE = "ACTIVE"
    IDLE = "IDLE"


class CapablityPerDevice:
    energy_info_models = [
        "shamu", "volantis", "volantisg", "angler", "bullhead", "ryu",
        "marlin", "sailfish"
    ]
    tdls_models = [
        "shamu", "hammerhead", "angler", "bullhead", "marlin", "sailfish"
    ]


ascii_letters_and_digits = string.ascii_letters + string.digits
valid_filename_chars = "-_." + ascii_letters_and_digits

models = ("sprout", "occam", "hammerhead", "bullhead", "razor", "razorg",
          "shamu", "angler", "volantis", "volantisg", "mantaray", "fugu",
          "ryu", "marlin", "sailfish")

manufacture_name_to_model = {
    "flo": "razor",
    "flo_lte": "razorg",
    "flounder": "volantis",
    "flounder_lte": "volantisg",
    "dragon": "ryu"
}

GMT_to_olson = {
    "GMT-9": "America/Anchorage",
    "GMT-8": "US/Pacific",
    "GMT-7": "US/Mountain",
    "GMT-6": "US/Central",
    "GMT-5": "US/Eastern",
    "GMT-4": "America/Barbados",
    "GMT-3": "America/Buenos_Aires",
    "GMT-2": "Atlantic/South_Georgia",
    "GMT-1": "Atlantic/Azores",
    "GMT+0": "Africa/Casablanca",
    "GMT+1": "Europe/Amsterdam",
    "GMT+2": "Europe/Athens",
    "GMT+3": "Europe/Moscow",
    "GMT+4": "Asia/Baku",
    "GMT+5": "Asia/Oral",
    "GMT+6": "Asia/Almaty",
    "GMT+7": "Asia/Bangkok",
    "GMT+8": "Asia/Hong_Kong",
    "GMT+9": "Asia/Tokyo",
    "GMT+10": "Pacific/Guam",
    "GMT+11": "Pacific/Noumea",
    "GMT+12": "Pacific/Fiji",
    "GMT+13": "Pacific/Tongatapu",
    "GMT-11": "Pacific/Midway",
    "GMT-10": "Pacific/Honolulu"
}


def abs_path(path):
    """Resolve the '.' and '~' in a path to get the absolute path.

    Args:
        path: The path to expand.

    Returns:
        The absolute path of the input path.
    """
    return os.path.abspath(os.path.expanduser(path))


def create_dir(path):
    """Creates a directory if it does not exist already.

    Args:
        path: The path of the directory to create.
    """
    full_path = abs_path(path)
    if not os.path.exists(full_path):
        os.makedirs(full_path)


def get_current_epoch_time():
    """Current epoch time in milliseconds.

    Returns:
        An integer representing the current epoch time in milliseconds.
    """
    return int(round(time.time() * 1000))


def get_current_human_time():
    """Returns the current time in human readable format.

    Returns:
        The current time stamp in Month-Day-Year Hour:Min:Sec format.
    """
    return time.strftime("%m-%d-%Y %H:%M:%S ")


def epoch_to_human_time(epoch_time):
    """Converts an epoch timestamp to human readable time.

    This essentially converts an output of get_current_epoch_time to an output
    of get_current_human_time

    Args:
        epoch_time: An integer representing an epoch timestamp in milliseconds.

    Returns:
        A time string representing the input time.
        None if input param is invalid.
    """
    if isinstance(epoch_time, int):
        try:
            d = datetime.datetime.fromtimestamp(epoch_time / 1000)
            return d.strftime("%m-%d-%Y %H:%M:%S ")
        except ValueError:
            return None


def get_timezone_olson_id():
    """Return the Olson ID of the local (non-DST) timezone.

    Returns:
        A string representing one of the Olson IDs of the local (non-DST)
        timezone.
    """
    tzoffset = int(time.timezone / 3600)
    gmt = None
    if tzoffset <= 0:
        gmt = "GMT+{}".format(-tzoffset)
    else:
        gmt = "GMT-{}".format(tzoffset)
    return GMT_to_olson[gmt]


def find_files(paths, file_predicate):
    """Locate files whose names and extensions match the given predicate in
    the specified directories.

    Args:
        paths: A list of directory paths where to find the files.
        file_predicate: A function that returns True if the file name and
          extension are desired.

    Returns:
        A list of files that match the predicate.
    """
    file_list = []
    if not isinstance(paths, list):
        paths = [paths]
    for path in paths:
        p = abs_path(path)
        for dirPath, subdirList, fileList in os.walk(p):
            for fname in fileList:
                name, ext = os.path.splitext(fname)
                if file_predicate(name, ext):
                    file_list.append((dirPath, name, ext))
    return file_list


def load_config(file_full_path):
    """Loads a JSON config file.

    Returns:
        A JSON object.
    """
    with open(file_full_path, 'r') as f:
        try:
            conf = json.load(f)
        except Exception as e:
            logging.error("Exception error to load %s: %s", f, e)
            raise
        return conf


def load_file_to_base64_str(f_path):
    """Loads the content of a file into a base64 string.

    Args:
        f_path: full path to the file including the file name.

    Returns:
        A base64 string representing the content of the file in utf-8 encoding.
    """
    path = abs_path(f_path)
    with open(path, 'rb') as f:
        f_bytes = f.read()
        base64_str = base64.b64encode(f_bytes).decode("utf-8")
        return base64_str


def dump_string_to_file(content, file_path, mode='w'):
    """ Dump content of a string to

    Args:
        content: content to be dumped to file
        file_path: full path to the file including the file name.
        mode: file open mode, 'w' (truncating file) by default
    :return:
    """
    full_path = abs_path(file_path)
    with open(full_path, mode) as f:
        f.write(content)


def find_field(item_list, cond, comparator, target_field):
    """Finds the value of a field in a dict object that satisfies certain
    conditions.

    Args:
        item_list: A list of dict objects.
        cond: A param that defines the condition.
        comparator: A function that checks if an dict satisfies the condition.
        target_field: Name of the field whose value to be returned if an item
            satisfies the condition.

    Returns:
        Target value or None if no item satisfies the condition.
    """
    for item in item_list:
        if comparator(item, cond) and target_field in item:
            return item[target_field]
    return None


def rand_ascii_str(length):
    """Generates a random string of specified length, composed of ascii letters
    and digits.

    Args:
        length: The number of characters in the string.

    Returns:
        The random string generated.
    """
    letters = [random.choice(ascii_letters_and_digits) for i in range(length)]
    return ''.join(letters)


# Thead/Process related functions.
def concurrent_exec(func, param_list):
    """Executes a function with different parameters pseudo-concurrently.

    This is basically a map function. Each element (should be an iterable) in
    the param_list is unpacked and passed into the function. Due to Python's
    GIL, there's no true concurrency. This is suited for IO-bound tasks.

    Args:
        func: The function that parforms a task.
        param_list: A list of iterables, each being a set of params to be
            passed into the function.

    Returns:
        A list of return values from each function execution. If an execution
        caused an exception, the exception object will be the corresponding
        result.
    """
    with concurrent.futures.ThreadPoolExecutor(max_workers=30) as executor:
        # Start the load operations and mark each future with its params
        future_to_params = {executor.submit(func, *p): p for p in param_list}
        return_vals = []
        for future in concurrent.futures.as_completed(future_to_params):
            params = future_to_params[future]
            try:
                return_vals.append(future.result())
            except Exception as exc:
                print("{} generated an exception: {}".format(
                    params, traceback.format_exc()))
                return_vals.append(exc)
        return return_vals


def exe_cmd(*cmds):
    """Executes commands in a new shell.

    Args:
        cmds: A sequence of commands and arguments.

    Returns:
        The output of the command run.

    Raises:
        OSError is raised if an error occurred during the command execution.
    """
    cmd = ' '.join(cmds)
    proc = subprocess.Popen(
        cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
    (out, err) = proc.communicate()
    if not err:
        return out
    raise OSError(err)


def require_sl4a(android_devices):
    """Makes sure sl4a connection is established on the given AndroidDevice
    objects.

    Args:
        android_devices: A list of AndroidDevice objects.

    Raises:
        AssertionError is raised if any given android device does not have SL4A
        connection established.
    """
    for ad in android_devices:
        msg = "SL4A connection not established properly on %s." % ad.serial
        assert ad.droid, msg


def _assert_subprocess_running(proc):
    """Checks if a subprocess has terminated on its own.

    Args:
        proc: A subprocess returned by subprocess.Popen.

    Raises:
        ActsUtilsError is raised if the subprocess has stopped.
    """
    ret = proc.poll()
    if ret is not None:
        out, err = proc.communicate()
        raise ActsUtilsError("Process %d has terminated. ret: %d, stderr: %s,"
                             " stdout: %s" % (proc.pid, ret, err, out))


def start_standing_subprocess(cmd, check_health_delay=0):
    """Starts a long-running subprocess.

    This is not a blocking call and the subprocess started by it should be
    explicitly terminated with stop_standing_subprocess.

    For short-running commands, you should use exe_cmd, which blocks.

    You can specify a health check after the subprocess is started to make sure
    it did not stop prematurely.

    Args:
        cmd: string, the command to start the subprocess with.
        check_health_delay: float, the number of seconds to wait after the
                            subprocess starts to check its health. Default is 0,
                            which means no check.

    Returns:
        The subprocess that got started.
    """
    proc = subprocess.Popen(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        shell=True,
        preexec_fn=os.setpgrp)
    logging.debug("Start standing subprocess with cmd: %s", cmd)
    if check_health_delay > 0:
        time.sleep(check_health_delay)
        _assert_subprocess_running(proc)
    return proc


def stop_standing_subprocess(proc, kill_signal=signal.SIGTERM):
    """Stops a subprocess started by start_standing_subprocess.

    Before killing the process, we check if the process is running, if it has
    terminated, ActsUtilsError is raised.

    Catches and ignores the PermissionError which only happens on Macs.

    Args:
        proc: Subprocess to terminate.
    """
    pid = proc.pid
    logging.debug("Stop standing subprocess %d", pid)
    _assert_subprocess_running(proc)
    try:
        os.killpg(pid, kill_signal)
    except PermissionError:
        pass


def wait_for_standing_subprocess(proc, timeout=None):
    """Waits for a subprocess started by start_standing_subprocess to finish
    or times out.

    Propagates the exception raised by the subprocess.wait(.) function.
    The subprocess.TimeoutExpired exception is raised if the process timed-out
    rather then terminating.

    If no exception is raised: the subprocess terminated on its own. No need
    to call stop_standing_subprocess() to kill it.

    If an exception is raised: the subprocess is still alive - it did not
    terminate. Either call stop_standing_subprocess() to kill it, or call
    wait_for_standing_subprocess() to keep waiting for it to terminate on its
    own.

    Args:
        p: Subprocess to wait for.
        timeout: An integer number of seconds to wait before timing out.
    """
    proc.wait(timeout)


def sync_device_time(ad):
    """Sync the time of an android device with the current system time.

    Both epoch time and the timezone will be synced.

    Args:
        ad: The android device to sync time on.
    """
    droid = ad.droid
    droid.setTimeZone(get_timezone_olson_id())
    droid.setTime(get_current_epoch_time())


# Timeout decorator block
class TimeoutError(Exception):
    """Exception for timeout decorator related errors.
    """
    pass


def _timeout_handler(signum, frame):
    """Handler function used by signal to terminate a timed out function.
    """
    raise TimeoutError()


def timeout(sec):
    """A decorator used to add time out check to a function.

    This only works in main thread due to its dependency on signal module.
    Do NOT use it if the decorated funtion does not run in the Main thread.

    Args:
        sec: Number of seconds to wait before the function times out.
            No timeout if set to 0

    Returns:
        What the decorated function returns.

    Raises:
        TimeoutError is raised when time out happens.
    """

    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            if sec:
                signal.signal(signal.SIGALRM, _timeout_handler)
                signal.alarm(sec)
            try:
                return func(*args, **kwargs)
            except TimeoutError:
                raise TimeoutError(("Function {} timed out after {} "
                                    "seconds.").format(func.__name__, sec))
            finally:
                signal.alarm(0)

        return wrapper

    return decorator


def trim_model_name(model):
    """Trim any prefix and postfix and return the android designation of the
    model name.

    e.g. "m_shamu" will be trimmed to "shamu".

    Args:
        model: model name to be trimmed.

    Returns
        Trimmed model name if one of the known model names is found.
        None otherwise.
    """
    # Directly look up first.
    if model in models:
        return model
    if model in manufacture_name_to_model:
        return manufacture_name_to_model[model]
    # If not found, try trimming off prefix/postfix and look up again.
    tokens = re.split("_|-", model)
    for t in tokens:
        if t in models:
            return t
        if t in manufacture_name_to_model:
            return manufacture_name_to_model[t]
    return None


def force_airplane_mode(ad, new_state, timeout_value=60):
    """Force the device to set airplane mode on or off by adb shell command.

    Args:
        ad: android device object.
        new_state: Turn on airplane mode if True.
            Turn off airplane mode if False.
        timeout_value: max wait time for 'adb wait-for-device'

    Returns:
        True if success.
        False if timeout.
    """
    # Using timeout decorator.
    # Wait for device with timeout. If after <timeout_value> seconds, adb
    # is still waiting for device, throw TimeoutError exception.
    @timeout(timeout_value)
    def wait_for_device_with_timeout(ad):
        ad.adb.wait_for_device()

    try:
        wait_for_device_with_timeout(ad)
        ad.adb.shell("settings put global airplane_mode_on {}".format(
            1 if new_state else 0))
        ad.adb.shell("am broadcast -a android.intent.action.AIRPLANE_MODE")
    except TimeoutError:
        # adb wait for device timeout
        return False
    return True


def enable_doze(ad):
    """Force the device into doze mode.

    Args:
        ad: android device object.

    Returns:
        True if device is in doze mode.
        False otherwise.
    """
    ad.adb.shell("dumpsys battery unplug")
    ad.adb.shell("dumpsys deviceidle enable")
    ad.adb.shell("dumpsys deviceidle force-idle")
    ad.droid.goToSleepNow()
    time.sleep(5)
    adb_shell_result = ad.adb.shell("dumpsys deviceidle get deep")
    if not adb_shell_result.startswith(DozeModeStatus.IDLE):
        info = ("dumpsys deviceidle get deep: {}".format(adb_shell_result))
        print(info)
        return False
    return True


def disable_doze(ad):
    """Force the device not in doze mode.

    Args:
        ad: android device object.

    Returns:
        True if device is not in doze mode.
        False otherwise.
    """
    ad.adb.shell("dumpsys deviceidle disable")
    ad.adb.shell("dumpsys battery reset")
    adb_shell_result = ad.adb.shell("dumpsys deviceidle get deep")
    if not adb_shell_result.startswith(DozeModeStatus.ACTIVE):
        info = ("dumpsys deviceidle get deep: {}".format(adb_shell_result))
        print(info)
        return False
    return True


def enable_doze_light(ad):
    """Force the device into doze light mode.

    Args:
        ad: android device object.

    Returns:
        True if device is in doze light mode.
        False otherwise.
    """
    ad.adb.shell("dumpsys battery unplug")
    ad.droid.goToSleepNow()
    time.sleep(5)
    ad.adb.shell("cmd deviceidle enable light")
    ad.adb.shell("cmd deviceidle step light")
    adb_shell_result = ad.adb.shell("dumpsys deviceidle get light")
    if not adb_shell_result.startswith(DozeModeStatus.IDLE):
        info = ("dumpsys deviceidle get light: {}".format(adb_shell_result))
        print(info)
        return False
    return True


def disable_doze_light(ad):
    """Force the device not in doze light mode.

    Args:
        ad: android device object.

    Returns:
        True if device is not in doze light mode.
        False otherwise.
    """
    ad.adb.shell("dumpsys battery reset")
    ad.adb.shell("cmd deviceidle disable light")
    adb_shell_result = ad.adb.shell("dumpsys deviceidle get light")
    if not adb_shell_result.startswith(DozeModeStatus.ACTIVE):
        info = ("dumpsys deviceidle get light: {}".format(adb_shell_result))
        print(info)
        return False
    return True


def set_ambient_display(ad, new_state):
    """Set "Ambient Display" in Settings->Display

    Args:
        ad: android device object.
        new_state: new state for "Ambient Display". True or False.
    """
    ad.adb.shell(
        "settings put secure doze_enabled {}".format(1 if new_state else 0))


def set_adaptive_brightness(ad, new_state):
    """Set "Adaptive Brightness" in Settings->Display

    Args:
        ad: android device object.
        new_state: new state for "Adaptive Brightness". True or False.
    """
    ad.adb.shell("settings put system screen_brightness_mode {}".format(
        1 if new_state else 0))


def set_auto_rotate(ad, new_state):
    """Set "Auto-rotate" in QuickSetting

    Args:
        ad: android device object.
        new_state: new state for "Auto-rotate". True or False.
    """
    ad.adb.shell("settings put system accelerometer_rotation {}".format(
        1 if new_state else 0))


def set_location_service(ad, new_state):
    """Set Location service on/off in Settings->Location

    Args:
        ad: android device object.
        new_state: new state for "Location service".
            If new_state is False, turn off location service.
            If new_state if True, set location service to "High accuracy".
    """
    ad.adb.shell("content insert --uri "
                 " content://com.google.settings/partner --bind "
                 "name:s:network_location_opt_in --bind value:s:1")
    ad.adb.shell("content insert --uri "
                 " content://com.google.settings/partner --bind "
                 "name:s:use_location_for_services --bind value:s:1")
    if new_state:
        ad.adb.shell("settings put secure location_providers_allowed +gps")
        ad.adb.shell("settings put secure location_providers_allowed +network")
    else:
        ad.adb.shell("settings put secure location_providers_allowed -gps")
        ad.adb.shell("settings put secure location_providers_allowed -network")


def set_mobile_data_always_on(ad, new_state):
    """Set Mobile_Data_Always_On feature bit

    Args:
        ad: android device object.
        new_state: new state for "mobile_data_always_on"
            if new_state is False, set mobile_data_always_on disabled.
            if new_state if True, set mobile_data_always_on enabled.
    """
    ad.adb.shell("settings put global mobile_data_always_on {}".format(
        1 if new_state else 0))


def bypass_setup_wizard(ad, bypass_wait_time=3):
    """Bypass the setup wizard on an input Android device

    Args:
        ad: android device object.
        bypass_wait_time: Do not set this variable. Only modified for framework
        tests.

    Returns:
        True if Andorid device successfully bypassed the setup wizard.
        False if failed.
    """
    try:
        ad.adb.shell("am start -n \"com.google.android.setupwizard/"
                     ".SetupWizardExitActivity\"")
        logging.debug("No error during default bypass call.")
    except adb.AdbError as adb_error:
        if adb_error.stdout == "ADB_CMD_OUTPUT:0":
            if adb_error.stderr and \
                    not adb_error.stderr.startswith("Error type 3\n"):
                logging.error(
                    "ADB_CMD_OUTPUT:0, but error is %s " % adb_error.stderr)
                raise adb_error
            logging.debug("Bypass wizard call received harmless error 3: "
                          "No setup to bypass.")
        elif adb_error.stdout == "ADB_CMD_OUTPUT:255":
            # Run it again as root.
            ad.adb.root_adb()
            logging.debug("Need root access to bypass setup wizard.")
            try:
                ad.adb.shell("am start -n \"com.google.android.setupwizard/"
                             ".SetupWizardExitActivity\"")
                logging.debug("No error during rooted bypass call.")
            except adb.AdbError as adb_error:
                if adb_error.stdout == "ADB_CMD_OUTPUT:0":
                    if adb_error.stderr and \
                            not adb_error.stderr.startswith("Error type 3\n"):
                        logging.error("Rooted ADB_CMD_OUTPUT:0, but error is "
                                      "%s " % adb_error.stderr)
                        raise adb_error
                    logging.debug(
                        "Rooted bypass wizard call received harmless "
                        "error 3: No setup to bypass.")

    # magical sleep to wait for the gservices override broadcast to complete
    time.sleep(bypass_wait_time)

    provisioned_state = int(
        ad.adb.shell("settings get global device_provisioned"))
    if provisioned_state != 1:
        logging.error("Failed to bypass setup wizard.")
        return False
    logging.debug("Setup wizard successfully bypassed.")
    return True


def parse_ping_ouput(ad, count, out, loss_tolerance=20):
    """Ping Parsing util.

    Args:
        ad: Android Device Object.
        count: Number of ICMP packets sent
        out: shell output text of ping operation
        loss_tolerance: Threshold after which flag test as false
    Returns:
        False: if packet loss is more than loss_tolerance%
        True: if all good
    """
    out = out.split('\n')[-3:]
    stats = out[1].split(',')
    # For failure case, line of interest becomes the last line
    if len(stats) != 4:
        stats = out[2].split(',')
    packet_loss = float(stats[2].split('%')[0])
    packet_xmit = int(stats[0].split()[0])
    packet_rcvd = int(stats[1].split()[0])
    min_packet_xmit_rcvd = (100 - loss_tolerance) * 0.01

    if (packet_loss >= loss_tolerance
            or packet_xmit < count * min_packet_xmit_rcvd
            or packet_rcvd < count * min_packet_xmit_rcvd):
        ad.log.error(
            "More than %d %% packet loss seen, Expected Packet_count %d \
            Packet loss %.2f%% Packets_xmitted %d Packets_rcvd %d",
            loss_tolerance, count, packet_loss, packet_xmit, packet_rcvd)
        return False
    ad.log.info("Pkt_count %d Pkt_loss %.2f%% Pkt_xmit %d Pkt_rcvd %d", count,
                packet_loss, packet_xmit, packet_rcvd)
    return True


def adb_shell_ping(ad,
                   count=120,
                   dest_ip="www.google.com",
                   timeout=200,
                   loss_tolerance=20):
    """Ping utility using adb shell.

    Args:
        ad: Android Device Object.
        count: Number of ICMP packets to send
        dest_ip: hostname or IP address
                 default www.google.com
        timeout: timeout for icmp pings to complete.
    """
    ping_cmd = "ping -W 1"
    if count:
        ping_cmd += " -c %d" % count
    if dest_ip:
        ping_cmd += " %s | tee /data/ping.txt" % dest_ip
    try:
        ad.log.info("Starting ping test to %s using adb command %s", dest_ip,
                    ping_cmd)
        out = ad.adb.shell(ping_cmd, timeout=timeout)
        if not parse_ping_ouput(ad, count, out, loss_tolerance):
            return False
        return True
    except Exception as e:
        ad.log.warning("Ping Test to %s failed with exception %s", dest_ip, e)
        return False
    finally:
        ad.adb.shell("rm /data/ping.txt", timeout=10, ignore_status=True)


def unzip_maintain_permissions(zip_path, extract_location):
    """Unzip a .zip file while maintaining permissions.

    Args:
        zip_path: The path to the zipped file.
        extract_location: the directory to extract to.
    """
    with zipfile.ZipFile(zip_path, 'r') as zip_file:
        for info in zip_file.infolist():
            _extract_file(zip_file, info, extract_location)


def _extract_file(zip_file, zip_info, extract_location):
    """Extracts a single entry from a ZipFile while maintaining permissions.

    Args:
        zip_file: A zipfile.ZipFile.
        zip_info: A ZipInfo object from zip_file.
        extract_location: The directory to extract to.
    """
    out_path = zip_file.extract(zip_info.filename, path=extract_location)
    perm = zip_info.external_attr >> 16
    os.chmod(out_path, perm)


def get_directory_size(path):
    """Computes the total size of the files in a directory, including subdirectories.

    Args:
        path: The path of the directory.
    Returns:
        The size of the provided directory.
    """
    total = 0
    for dirpath, dirnames, filenames in os.walk(path):
        for filename in filenames:
            total += os.path.getsize(os.path.join(dirpath, filename))
    return total


def get_process_uptime(process):
    """Returns the runtime in [[dd-]hh:]mm:ss, or '' if not running."""
    pid = job.run('pidof %s' % process, ignore_status=True).stdout
    runtime = ''
    if pid:
        runtime = job.run('ps -o etime= -p "%s"' % pid).stdout
    return runtime


def get_device_process_uptime(adb, process):
    """Returns the uptime of a device process."""
    pid = adb.shell('pidof %s' % process, ignore_status=True)
    runtime = ''
    if pid:
        runtime = adb.shell('ps -o etime= -p "%s"' % pid)
    return runtime