utils.py

#
# Copyright (C) 2017 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.
#
# pylint: disable=not-callable

import contextlib
import datetime
import logging
import os
from pathlib import Path
import shlex
import shutil
import stat
import subprocess
from typing import Dict, List

import constants


ORIG_ENV = dict(os.environ)

def logger():
    """Returns the module level logger."""
    return logging.getLogger(__name__)


def subprocess_run(cmd, *args, **kwargs):
    """subprocess.run with logging."""
    logger().debug('subprocess.run:%s %s',
                  datetime.datetime.now().strftime("%H:%M:%S"),
                  list2cmdline(cmd))
    if kwargs.pop('dry_run', None):
        return None
    return subprocess.run(cmd, *args, **kwargs, text=True)


def unchecked_call(cmd, *args, **kwargs):
    """subprocess.call with logging."""
    return subprocess_run(cmd, *args, **kwargs).returncode


def check_call(cmd, *args, **kwargs):
    """subprocess.check_call with logging."""
    return subprocess_run(cmd, *args, **kwargs, check=True)


def check_output(cmd, *args, **kwargs):
    """subprocess.check_output with logging."""
    return subprocess_run(cmd, *args, **kwargs, check=True, stdout=subprocess.PIPE).stdout


def is_available_mac_ver(ver: str) -> bool:
    """Returns whether a version string is equal to or under MAC_MIN_VERSION."""
    _parse_version = lambda ver: list(int(v) for v in ver.split('.'))
    return _parse_version(ver) <= _parse_version(constants.MAC_MIN_VERSION)


def list2cmdline(args: List[str]) -> str:
    """Joins arguments into a Bourne-shell cmdline.

    Like shlex.join from Python 3.8, but is flexible about the argument type.
    Each argument can be a str, a bytes, or a path-like object. (subprocess.call
    is similarly flexible.)

    Similar to the undocumented subprocess.list2cmdline, but does Bourne-style
    escaping rather than MSVCRT escaping.
    """
    return ' '.join([shlex.quote(os.fsdecode(arg)) for arg in args])


def create_script(script_path: Path, cmd: List[str], env: Dict[str, str]) -> None:
    with script_path.open('w') as outf:
        outf.write('#!/bin/sh\n')
        for k, v in env.items():
            if v != ORIG_ENV.get(k):
                outf.write(f'export {k}="{v}"\n')
        outf.write(list2cmdline(cmd) + ' $@\n')
    script_path.chmod(0o755)


def check_gcertstatus() -> None:
    """Ensure gcert valid for > 1 hour."""
    try:
        check_call([
            'gcertstatus', '-quiet', '-check_ssh=false', '-check_remaining=1h'
        ])
    except subprocess.CalledProcessError:
        print('Run prodaccess before executing this script.')
        raise


@contextlib.contextmanager
def chdir_context(directory):
    prev_dir = os.getcwd()
    try:
        os.chdir(directory)
        yield
    finally:
        os.chdir(prev_dir)