-
Notifications
You must be signed in to change notification settings - Fork 6
Expand file tree
/
Copy path__init__.py
More file actions
86 lines (66 loc) · 2.66 KB
/
__init__.py
File metadata and controls
86 lines (66 loc) · 2.66 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
from __future__ import absolute_import
import subprocess
import sys
import os
OTS_SANITIZE = os.path.join(os.path.dirname(__file__), "ots-sanitize")
__all__ = ["sanitize", "OTSError", "CalledProcessError"]
try:
from ._version import version as __version__
except ImportError:
__version__ = "0.0.0+unknown"
class OTSError(Exception):
pass
# subprocess.CalledProcessError on python < 3.5 doesn't have 'stderr' argument.
# Regardless, it's a good idea to wrap subprocess' exceptions with our own.
class CalledProcessError(OTSError, subprocess.CalledProcessError):
def __init__(self, returncode, cmd, output=None, stderr=None):
subprocess.CalledProcessError.__init__(self, returncode, cmd, output=output)
self.stderr = stderr
@property
def stdout(self):
"""Alias for output attribute, to match stderr"""
return self.output
try:
from subprocess import CompletedProcess as _CompletedProcess
class CompletedProcess(_CompletedProcess):
pass
except ImportError: # only added from python 3.5
from collections import namedtuple
class CompletedProcess(
namedtuple("CompletedProcess", "args returncode stdout stderr")
):
def check_returncode(self):
if self.returncode:
raise CalledProcessError(
self.returncode, self.args, self.stdout, self.stderr
)
def _run(args, capture_output=False, check=False, **kwargs):
if capture_output:
if ("stdout" in kwargs) or ("stderr" in kwargs):
raise ValueError(
"stdout and stderr arguments may not be used with capture_output."
)
kwargs["stdout"] = subprocess.PIPE
kwargs["stderr"] = subprocess.PIPE
process = subprocess.Popen(args, **kwargs)
try:
stdout, stderr = process.communicate()
except:
process.kill()
process.wait()
raise
retcode = process.poll()
if check and retcode:
raise CalledProcessError(retcode, args, output=stdout, stderr=stderr)
return CompletedProcess(args, retcode, stdout, stderr)
def sanitize(*args, **kwargs):
"""Run the embedded ots-sanitize executable with the list of positional
arguments (strings).
Return an ots.CompletedProcess object with the following attributes:
args, returncode, stdout, stderr.
If check=True, and the subprocess exits with a non-zero exit code, an
ots.CalledProcessError exception will be raised.
If capture_output=True, stdout and stderr will be captured.
All extra keyword arguments are forwarded to subprocess.Popen constructor.
"""
return _run([OTS_SANITIZE] + list(args), **kwargs)