bernsteinbear.com
this repo has no description
1import logging
2import os
3import subprocess
4import sys
5import shlex
6import textwrap
7
8SCRIPT_PATH = os.path.dirname(os.path.realpath(__file__))
9
10
11def run(
12 cmd,
13 verbose=True,
14 cwd=None,
15 check=True,
16 capture_output=False,
17 encoding="utf-8",
18 log_level=logging.DEBUG,
19 # Only pipe stdout; we want stderr (logging, errors) to be unbuffered.
20 stdout=subprocess.PIPE,
21 # Specify an integer number of seconds
22 timeout=-1,
23 **kwargs,
24):
25 if verbose:
26 info = "$ "
27 if cwd is not None:
28 info += f"cd {cwd}; "
29 info += " ".join(shlex.quote(c) for c in cmd)
30 if capture_output:
31 info += " >& ..."
32 lines = textwrap.wrap(
33 info,
34 break_on_hyphens=False,
35 break_long_words=False,
36 replace_whitespace=False,
37 subsequent_indent=" ",
38 )
39 logging.log(log_level, " \\\n".join(lines))
40 if timeout != -1:
41 cmd = ["timeout", "--signal=KILL", f"{timeout}s", *cmd]
42 try:
43 return subprocess.run(
44 cmd,
45 cwd=cwd,
46 check=check,
47 capture_output=capture_output,
48 encoding=encoding,
49 stdout=stdout,
50 **kwargs,
51 )
52 except subprocess.CalledProcessError as e:
53 if e.returncode == -9:
54 # Error code from `timeout` command signaling it had to be killed
55 raise TimeoutError("Command timed out", cmd)
56 logging.critical("%s", e.stdout)
57 raise
58
59
60def get_repo_root(dirname):
61 proc = run(
62 ["git", "rev-parse", "--show-toplevel"],
63 cwd=dirname,
64 )
65 return proc.stdout.strip()
66
67
68def normalize_revision(repo_root, revspec):
69 proc = run(
70 ["git", "rev-parse", revspec],
71 cwd=repo_root,
72 )
73 return proc.stdout.strip()