1import subprocess 2from functools import partial 3from typing import Callable 4 5from mozlog import get_default_logger 6 7from wptserve.utils import isomorphic_decode 8 9logger = None 10 11def vcs(bin_name: str) -> Callable[..., None]: 12 def inner(command, *args, **kwargs): 13 global logger 14 15 if logger is None: 16 logger = get_default_logger("vcs") 17 18 repo = kwargs.pop("repo", None) 19 log_error = kwargs.pop("log_error", True) 20 stdout = kwargs.pop("stdout", None) 21 stdin = kwargs.pop("stdin", None) 22 if kwargs: 23 raise TypeError(kwargs) 24 25 args = list(args) 26 27 proc_kwargs = {} 28 if repo is not None: 29 # Make sure `cwd` is str type to work in different sub-versions of Python 3. 30 # Before 3.8, bytes were not accepted on Windows for `cwd`. 31 proc_kwargs["cwd"] = isomorphic_decode(repo) 32 if stdout is not None: 33 proc_kwargs["stdout"] = stdout 34 if stdin is not None: 35 proc_kwargs["stdin"] = stdin 36 37 command_line = [bin_name, command] + args 38 logger.debug(" ".join(command_line)) 39 try: 40 func = subprocess.check_output if not stdout else subprocess.check_call 41 return func(command_line, stderr=subprocess.STDOUT, **proc_kwargs) 42 except OSError as e: 43 if log_error: 44 logger.error(e) 45 raise 46 except subprocess.CalledProcessError as e: 47 if log_error: 48 logger.error(e.output) 49 raise 50 return inner 51 52git = vcs("git") 53hg = vcs("hg") 54 55 56def bind_to_repo(vcs_func, repo, log_error=True): 57 return partial(vcs_func, repo=repo, log_error=log_error) 58 59 60def is_git_root(path, log_error=True): 61 try: 62 rv = git("rev-parse", "--show-cdup", repo=path, log_error=log_error) 63 except subprocess.CalledProcessError: 64 return False 65 return rv == b"\n" 66