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
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
|
# Copyright 2014 Google Inc.
#
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Module to host the VerboseSubprocess, ChangeDir, and ReSearch classes.
"""
import os
import re
import subprocess
def print_subprocess_args(prefix, *args, **kwargs):
"""Print out args in a human-readable manner."""
def quote_and_escape(string):
"""Quote and escape a string if necessary."""
if ' ' in string or '\n' in string:
string = '"%s"' % string.replace('"', '\\"')
return string
if 'cwd' in kwargs:
print '%scd %s' % (prefix, kwargs['cwd'])
print prefix + ' '.join(quote_and_escape(arg) for arg in args[0])
if 'cwd' in kwargs:
print '%scd -' % prefix
class VerboseSubprocess(object):
"""Call subprocess methods, but print out command before executing.
Attributes:
verbose: (boolean) should we print out the command or not. If
not, this is the same as calling the subprocess method
quiet: (boolean) suppress stdout on check_call and call.
prefix: (string) When verbose, what to print before each command.
"""
def __init__(self, verbose):
self.verbose = verbose
self.quiet = not verbose
self.prefix = '~~$ '
def check_call(self, *args, **kwargs):
"""Wrapper for subprocess.check_call().
Args:
*args: to be passed to subprocess.check_call()
**kwargs: to be passed to subprocess.check_call()
Returns:
Whatever subprocess.check_call() returns.
Raises:
OSError or subprocess.CalledProcessError: raised by check_call.
"""
if self.verbose:
print_subprocess_args(self.prefix, *args, **kwargs)
if self.quiet:
with open(os.devnull, 'w') as devnull:
return subprocess.check_call(*args, stdout=devnull, **kwargs)
else:
return subprocess.check_call(*args, **kwargs)
def call(self, *args, **kwargs):
"""Wrapper for subprocess.check().
Args:
*args: to be passed to subprocess.check_call()
**kwargs: to be passed to subprocess.check_call()
Returns:
Whatever subprocess.call() returns.
Raises:
OSError or subprocess.CalledProcessError: raised by call.
"""
if self.verbose:
print_subprocess_args(self.prefix, *args, **kwargs)
if self.quiet:
with open(os.devnull, 'w') as devnull:
return subprocess.call(*args, stdout=devnull, **kwargs)
else:
return subprocess.call(*args, **kwargs)
def check_output(self, *args, **kwargs):
"""Wrapper for subprocess.check_output().
Args:
*args: to be passed to subprocess.check_output()
**kwargs: to be passed to subprocess.check_output()
Returns:
Whatever subprocess.check_output() returns.
Raises:
OSError or subprocess.CalledProcessError: raised by check_output.
"""
if self.verbose:
print_subprocess_args(self.prefix, *args, **kwargs)
return subprocess.check_output(*args, **kwargs)
def strip_output(self, *args, **kwargs):
"""Wrap subprocess.check_output and str.strip().
Pass the given arguments into subprocess.check_output() and return
the results, after stripping any excess whitespace.
Args:
*args: to be passed to subprocess.check_output()
**kwargs: to be passed to subprocess.check_output()
Returns:
The output of the process as a string without leading or
trailing whitespace.
Raises:
OSError or subprocess.CalledProcessError: raised by check_output.
"""
if self.verbose:
print_subprocess_args(self.prefix, *args, **kwargs)
return str(subprocess.check_output(*args, **kwargs)).strip()
def popen(self, *args, **kwargs):
"""Wrapper for subprocess.Popen().
Args:
*args: to be passed to subprocess.Popen()
**kwargs: to be passed to subprocess.Popen()
Returns:
The output of subprocess.Popen()
Raises:
OSError or subprocess.CalledProcessError: raised by Popen.
"""
if self.verbose:
print_subprocess_args(self.prefix, *args, **kwargs)
return subprocess.Popen(*args, **kwargs)
class ChangeDir(object):
"""Use with a with-statement to temporarily change directories."""
# pylint: disable=I0011,R0903
def __init__(self, directory, verbose=False):
self._directory = directory
self._verbose = verbose
def __enter__(self):
if self._directory != os.curdir:
if self._verbose:
print '~~$ cd %s' % self._directory
cwd = os.getcwd()
os.chdir(self._directory)
self._directory = cwd
def __exit__(self, etype, value, traceback):
if self._directory != os.curdir:
if self._verbose:
print '~~$ cd %s' % self._directory
os.chdir(self._directory)
class ReSearch(object):
"""A collection of static methods for regexing things."""
@staticmethod
def search_within_stream(input_stream, pattern, default=None):
"""Search for regular expression in a file-like object.
Opens a file for reading and searches line by line for a match to
the regex and returns the parenthesized group named return for the
first match. Does not search across newlines.
For example:
pattern = '^root(:[^:]*){4}:(?P<return>[^:]*)'
with open('/etc/passwd', 'r') as stream:
return search_within_file(stream, pattern)
should return root's home directory (/root on my system).
Args:
input_stream: file-like object to be read
pattern: (string) to be passed to re.compile
default: what to return if no match
Returns:
A string or whatever default is
"""
pattern_object = re.compile(pattern)
for line in input_stream:
match = pattern_object.search(line)
if match:
return match.group('return')
return default
@staticmethod
def search_within_string(input_string, pattern, default=None):
"""Search for regular expression in a string.
Args:
input_string: (string) to be searched
pattern: (string) to be passed to re.compile
default: what to return if no match
Returns:
A string or whatever default is
"""
match = re.search(pattern, input_string)
return match.group('return') if match else default
@staticmethod
def search_within_output(verbose, pattern, default, *args, **kwargs):
"""Search for regular expression in a process output.
Does not search across newlines.
Args:
verbose: (boolean) shoule we call print_subprocess_args?
pattern: (string) to be passed to re.compile
default: what to return if no match
*args: to be passed to subprocess.Popen()
**kwargs: to be passed to subprocess.Popen()
Returns:
A string or whatever default is
"""
if verbose:
print_subprocess_args('~~$ ', *args, **kwargs)
proc = subprocess.Popen(*args, stdout=subprocess.PIPE, **kwargs)
return ReSearch.search_within_stream(proc.stdout, pattern, default)
|
