testing/web-platform/tests/tools/wpt/utils.py
author pyup.io bot <github-bot@pyup.io>
Wed, 06 Mar 2019 10:34:49 +0000
changeset 522222 263580f65aea68089aff482a07c42cc8989511cb
parent 502090 ee1d4613f8624421711351352b6c656550513b03
child 522313 66b1aba3fb4372ec47986e728b1da9b32dd3d368
permissions -rw-r--r--
Bug 1528830 [wpt PR 15450] - Update flake8 to 3.7.6, a=testonly Automatic update from web-platform-tests Update flake8 to 3.7.6 (#15450) -- wpt-commits: 34e6ee21a64e1b2a9b774f102b1b183ea9858512 wpt-pr: 15450

import logging
import os
import subprocess
import tarfile
import zipfile
from io import BytesIO

try:
    from typing import Any
except ImportError:
    pass

logger = logging.getLogger(__name__)


class Kwargs(dict):
    def set_if_none(self,
                    name,            # type: str
                    value,           # type: Any
                    err_fn=None,     # type: (Kwargs, str) -> Any
                    desc=None,       # type: str
                    extra_cond=None  # type: (Kwargs) -> bool
                    ):
        if desc is None:
            desc = name

        if name not in self or self[name] is None:
            if extra_cond is not None and not extra_cond(self):
                return
            if callable(value):
                value = value()
            if not value:
                if err_fn is not None:
                    return err_fn(self, "Failed to find %s" % desc)
                else:
                    return
            self[name] = value
            logger.info("Set %s to %s" % (desc, value))


def call(*args):
    """Log terminal command, invoke it as a subprocess.

    Returns a bytestring of the subprocess output if no error.
    """
    logger.debug(" ".join(args))
    try:
        return subprocess.check_output(args)
    except subprocess.CalledProcessError as e:
        logger.critical("%s exited with return code %i" %
                        (e.cmd, e.returncode))
        logger.critical(e.output)
        raise


def seekable(fileobj):
    """Attempt to use file.seek on given file, with fallbacks."""
    try:
        fileobj.seek(fileobj.tell())
    except Exception:
        return BytesIO(fileobj.read())
    else:
        return fileobj


def untar(fileobj, dest="."):
    """Extract tar archive."""
    logger.debug("untar")
    fileobj = seekable(fileobj)
    with tarfile.open(fileobj=fileobj) as tar_data:
        tar_data.extractall(path=dest)


def unzip(fileobj, dest=None, limit=None):
    """Extract zip archive."""
    logger.debug("unzip")
    fileobj = seekable(fileobj)
    with zipfile.ZipFile(fileobj) as zip_data:
        for info in zip_data.infolist():
            if limit is not None and info.filename not in limit:
                continue
            zip_data.extract(info, path=dest)
            perm = info.external_attr >> 16 & 0x1FF
            os.chmod(os.path.join(dest, info.filename), perm)


def get(url):
    """Issue GET request to a given URL and return the response."""
    import requests

    logger.debug("GET %s" % url)
    resp = requests.get(url, stream=True)
    resp.raise_for_status()
    return resp