blob: 13978a47421116fd9c1cd45936e584431150618c [file] [log] [blame]
commit-bot@chromium.org0ed90022014-01-30 22:12:30 +00001# Copyright 2014 Google Inc.
2#
3# Use of this source code is governed by a BSD-style license that can be
4# found in the LICENSE file.
5
6
7"""Module to host the VerboseSubprocess, ChangeDir, and ReSearch classes.
8"""
9
10import os
11import re
12import subprocess
13
14
15def print_subprocess_args(prefix, *args, **kwargs):
16 """Print out args in a human-readable manner."""
17 def quote_and_escape(string):
18 """Quote and escape a string if necessary."""
19 if ' ' in string or '\n' in string:
20 string = '"%s"' % string.replace('"', '\\"')
21 return string
22 if 'cwd' in kwargs:
23 print '%scd %s' % (prefix, kwargs['cwd'])
24 print prefix + ' '.join(quote_and_escape(arg) for arg in args[0])
25 if 'cwd' in kwargs:
26 print '%scd -' % prefix
27
28
29class VerboseSubprocess(object):
30 """Call subprocess methods, but print out command before executing.
31
32 Attributes:
33 verbose: (boolean) should we print out the command or not. If
34 not, this is the same as calling the subprocess method
35 quiet: (boolean) suppress stdout on check_call and call.
36 prefix: (string) When verbose, what to print before each command.
37 """
38
39 def __init__(self, verbose):
40 self.verbose = verbose
41 self.quiet = not verbose
42 self.prefix = '~~$ '
43
44 def check_call(self, *args, **kwargs):
45 """Wrapper for subprocess.check_call().
46
47 Args:
48 *args: to be passed to subprocess.check_call()
49 **kwargs: to be passed to subprocess.check_call()
50 Returns:
51 Whatever subprocess.check_call() returns.
52 Raises:
53 OSError or subprocess.CalledProcessError: raised by check_call.
54 """
55 if self.verbose:
56 print_subprocess_args(self.prefix, *args, **kwargs)
57 if self.quiet:
58 with open(os.devnull, 'w') as devnull:
59 return subprocess.check_call(*args, stdout=devnull, **kwargs)
60 else:
61 return subprocess.check_call(*args, **kwargs)
62
63 def call(self, *args, **kwargs):
64 """Wrapper for subprocess.check().
65
66 Args:
67 *args: to be passed to subprocess.check_call()
68 **kwargs: to be passed to subprocess.check_call()
69 Returns:
70 Whatever subprocess.call() returns.
71 Raises:
72 OSError or subprocess.CalledProcessError: raised by call.
73 """
74 if self.verbose:
75 print_subprocess_args(self.prefix, *args, **kwargs)
76 if self.quiet:
77 with open(os.devnull, 'w') as devnull:
78 return subprocess.call(*args, stdout=devnull, **kwargs)
79 else:
80 return subprocess.call(*args, **kwargs)
81
82 def check_output(self, *args, **kwargs):
83 """Wrapper for subprocess.check_output().
84
85 Args:
86 *args: to be passed to subprocess.check_output()
87 **kwargs: to be passed to subprocess.check_output()
88 Returns:
89 Whatever subprocess.check_output() returns.
90 Raises:
91 OSError or subprocess.CalledProcessError: raised by check_output.
92 """
93 if self.verbose:
94 print_subprocess_args(self.prefix, *args, **kwargs)
95 return subprocess.check_output(*args, **kwargs)
96
97 def strip_output(self, *args, **kwargs):
98 """Wrap subprocess.check_output and str.strip().
99
100 Pass the given arguments into subprocess.check_output() and return
101 the results, after stripping any excess whitespace.
102
103 Args:
104 *args: to be passed to subprocess.check_output()
105 **kwargs: to be passed to subprocess.check_output()
106
107 Returns:
108 The output of the process as a string without leading or
109 trailing whitespace.
110 Raises:
111 OSError or subprocess.CalledProcessError: raised by check_output.
112 """
113 if self.verbose:
114 print_subprocess_args(self.prefix, *args, **kwargs)
115 return str(subprocess.check_output(*args, **kwargs)).strip()
116
117 def popen(self, *args, **kwargs):
118 """Wrapper for subprocess.Popen().
119
120 Args:
121 *args: to be passed to subprocess.Popen()
122 **kwargs: to be passed to subprocess.Popen()
123 Returns:
124 The output of subprocess.Popen()
125 Raises:
126 OSError or subprocess.CalledProcessError: raised by Popen.
127 """
128 if self.verbose:
129 print_subprocess_args(self.prefix, *args, **kwargs)
130 return subprocess.Popen(*args, **kwargs)
131
132
133class ChangeDir(object):
134 """Use with a with-statement to temporarily change directories."""
135 # pylint: disable=I0011,R0903
136
137 def __init__(self, directory, verbose=False):
138 self._directory = directory
139 self._verbose = verbose
140
141 def __enter__(self):
142 if self._directory != os.curdir:
143 if self._verbose:
144 print '~~$ cd %s' % self._directory
145 cwd = os.getcwd()
146 os.chdir(self._directory)
147 self._directory = cwd
148
149 def __exit__(self, etype, value, traceback):
150 if self._directory != os.curdir:
151 if self._verbose:
152 print '~~$ cd %s' % self._directory
153 os.chdir(self._directory)
154
155
156class ReSearch(object):
157 """A collection of static methods for regexing things."""
158
159 @staticmethod
160 def search_within_stream(input_stream, pattern, default=None):
161 """Search for regular expression in a file-like object.
162
163 Opens a file for reading and searches line by line for a match to
164 the regex and returns the parenthesized group named return for the
165 first match. Does not search across newlines.
166
167 For example:
168 pattern = '^root(:[^:]*){4}:(?P<return>[^:]*)'
169 with open('/etc/passwd', 'r') as stream:
170 return search_within_file(stream, pattern)
171 should return root's home directory (/root on my system).
172
173 Args:
174 input_stream: file-like object to be read
175 pattern: (string) to be passed to re.compile
176 default: what to return if no match
177
178 Returns:
179 A string or whatever default is
180 """
181 pattern_object = re.compile(pattern)
182 for line in input_stream:
183 match = pattern_object.search(line)
184 if match:
185 return match.group('return')
186 return default
187
188 @staticmethod
189 def search_within_string(input_string, pattern, default=None):
190 """Search for regular expression in a string.
191
192 Args:
193 input_string: (string) to be searched
194 pattern: (string) to be passed to re.compile
195 default: what to return if no match
196
197 Returns:
198 A string or whatever default is
199 """
200 match = re.search(pattern, input_string)
201 return match.group('return') if match else default
202
203 @staticmethod
204 def search_within_output(verbose, pattern, default, *args, **kwargs):
205 """Search for regular expression in a process output.
206
207 Does not search across newlines.
208
209 Args:
210 verbose: (boolean) shoule we call print_subprocess_args?
211 pattern: (string) to be passed to re.compile
212 default: what to return if no match
213 *args: to be passed to subprocess.Popen()
214 **kwargs: to be passed to subprocess.Popen()
215
216 Returns:
217 A string or whatever default is
218 """
219 if verbose:
220 print_subprocess_args('~~$ ', *args, **kwargs)
221 proc = subprocess.Popen(*args, stdout=subprocess.PIPE, **kwargs)
222 return ReSearch.search_within_stream(proc.stdout, pattern, default)
223
224