OLD | NEW |
1 #!/usr/bin/env python | 1 #!/usr/bin/env python |
2 # Copyright (c) 2011 The Chromium Authors. All rights reserved. | 2 # Copyright (c) 2011 The Chromium Authors. All rights reserved. |
3 # Use of this source code is governed by a BSD-style license that can be | 3 # Use of this source code is governed by a BSD-style license that can be |
4 # found in the LICENSE file. | 4 # found in the LICENSE file. |
5 | 5 |
6 """Unit tests for subprocess2.py.""" | 6 """Unit tests for subprocess2.py.""" |
7 | 7 |
8 import optparse | 8 import optparse |
9 import os | 9 import os |
10 import sys | 10 import sys |
(...skipping 16 matching lines...) Expand all Loading... |
27 subprocess2.subprocess: ['Popen'], | 27 subprocess2.subprocess: ['Popen'], |
28 } | 28 } |
29 | 29 |
30 def setUp(self): | 30 def setUp(self): |
31 self.exe_path = __file__ | 31 self.exe_path = __file__ |
32 self.exe = [sys.executable, self.exe_path, '--child'] | 32 self.exe = [sys.executable, self.exe_path, '--child'] |
33 self.saved = {} | 33 self.saved = {} |
34 for module, names in self.TO_SAVE.iteritems(): | 34 for module, names in self.TO_SAVE.iteritems(): |
35 self.saved[module] = dict( | 35 self.saved[module] = dict( |
36 (name, getattr(module, name)) for name in names) | 36 (name, getattr(module, name)) for name in names) |
| 37 # TODO(maruel): Do a reopen() on sys.__stdout__ and sys.__stderr__ so they |
| 38 # can be trapped in the child process for better coverage. |
37 | 39 |
38 def tearDown(self): | 40 def tearDown(self): |
39 for module, saved in self.saved.iteritems(): | 41 for module, saved in self.saved.iteritems(): |
40 for name, value in saved.iteritems(): | 42 for name, value in saved.iteritems(): |
41 setattr(module, name, value) | 43 setattr(module, name, value) |
42 | 44 |
43 @staticmethod | 45 @staticmethod |
44 def _fake_communicate(): | 46 def _fake_communicate(): |
45 results = {} | 47 results = {} |
46 def fake_communicate(args, **kwargs): | 48 def fake_communicate(args, **kwargs): |
(...skipping 35 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
82 def test_check_call_defaults(self): | 84 def test_check_call_defaults(self): |
83 results = self._fake_communicate() | 85 results = self._fake_communicate() |
84 self.assertEquals( | 86 self.assertEquals( |
85 ['stdout', 'stderr'], subprocess2.check_call_out(['foo'], a=True)) | 87 ['stdout', 'stderr'], subprocess2.check_call_out(['foo'], a=True)) |
86 expected = { | 88 expected = { |
87 'args': ['foo'], | 89 'args': ['foo'], |
88 'a':True, | 90 'a':True, |
89 } | 91 } |
90 self.assertEquals(expected, results) | 92 self.assertEquals(expected, results) |
91 | 93 |
| 94 def test_capture_defaults(self): |
| 95 results = self._fake_communicate() |
| 96 self.assertEquals( |
| 97 'stdout', subprocess2.capture(['foo'], a=True)) |
| 98 expected = { |
| 99 'args': ['foo'], |
| 100 'a':True, |
| 101 'stdin': subprocess2.VOID, |
| 102 'stdout': subprocess2.PIPE, |
| 103 } |
| 104 self.assertEquals(expected, results) |
| 105 |
92 def test_communicate_defaults(self): | 106 def test_communicate_defaults(self): |
93 results = self._fake_Popen() | 107 results = self._fake_Popen() |
94 self.assertEquals( | 108 self.assertEquals( |
95 ((None, None), -8), subprocess2.communicate(['foo'], a=True)) | 109 ((None, None), -8), subprocess2.communicate(['foo'], a=True)) |
96 expected = { | 110 expected = { |
97 'args': ['foo'], | 111 'args': ['foo'], |
98 'a': True, | 112 'a': True, |
99 } | 113 } |
100 self.assertEquals(expected, results) | 114 self.assertEquals(expected, results) |
101 | 115 |
(...skipping 20 matching lines...) Expand all Loading... |
122 def test_check_output_defaults(self): | 136 def test_check_output_defaults(self): |
123 results = self._fake_communicate() | 137 results = self._fake_communicate() |
124 # It's discarding 'stderr' because it assumes stderr=subprocess2.STDOUT but | 138 # It's discarding 'stderr' because it assumes stderr=subprocess2.STDOUT but |
125 # fake_communicate() doesn't 'implement' that. | 139 # fake_communicate() doesn't 'implement' that. |
126 self.assertEquals('stdout', subprocess2.check_output(['foo'], a=True)) | 140 self.assertEquals('stdout', subprocess2.check_output(['foo'], a=True)) |
127 expected = { | 141 expected = { |
128 'args': ['foo'], | 142 'args': ['foo'], |
129 'a':True, | 143 'a':True, |
130 'stdin': subprocess2.VOID, | 144 'stdin': subprocess2.VOID, |
131 'stdout': subprocess2.PIPE, | 145 'stdout': subprocess2.PIPE, |
132 'stderr': subprocess2.STDOUT, | |
133 } | 146 } |
134 self.assertEquals(expected, results) | 147 self.assertEquals(expected, results) |
135 | 148 |
136 def test_timeout(self): | 149 def test_timeout(self): |
137 # It'd be better to not discard stdout. | 150 # It'd be better to not discard stdout. |
138 out, returncode = subprocess2.communicate( | 151 out, returncode = subprocess2.communicate( |
139 self.exe + ['--sleep', '--stdout'], | 152 self.exe + ['--sleep', '--stdout'], |
140 timeout=0.01, | 153 timeout=0.01, |
141 stdout=subprocess2.PIPE) | 154 stdout=subprocess2.PIPE) |
142 self.assertEquals(subprocess2.TIMED_OUT, returncode) | 155 self.assertEquals(subprocess2.TIMED_OUT, returncode) |
143 self.assertEquals(['', None], out) | 156 self.assertEquals(['', None], out) |
144 | 157 |
145 def test_void(self): | 158 def test_check_output_no_stdout(self): |
146 out = subprocess2.check_output( | 159 try: |
| 160 subprocess2.check_output(self.exe, stdout=subprocess2.PIPE) |
| 161 self.fail() |
| 162 except TypeError: |
| 163 pass |
| 164 |
| 165 def test_stdout_void(self): |
| 166 (out, err), code = subprocess2.communicate( |
147 self.exe + ['--stdout', '--stderr'], | 167 self.exe + ['--stdout', '--stderr'], |
148 stdout=subprocess2.VOID) | 168 stdout=subprocess2.VOID, |
| 169 stderr=subprocess2.PIPE) |
149 self.assertEquals(None, out) | 170 self.assertEquals(None, out) |
150 out = subprocess2.check_output( | 171 self.assertEquals('a\nbb\nccc\n', err) |
| 172 self.assertEquals(0, code) |
| 173 |
| 174 def test_stderr_void(self): |
| 175 (out, err), code = subprocess2.communicate( |
151 self.exe + ['--stdout', '--stderr'], | 176 self.exe + ['--stdout', '--stderr'], |
152 universal_newlines=True, | 177 universal_newlines=True, |
| 178 stdout=subprocess2.PIPE, |
153 stderr=subprocess2.VOID) | 179 stderr=subprocess2.VOID) |
154 self.assertEquals('A\nBB\nCCC\n', out) | 180 self.assertEquals('A\nBB\nCCC\n', out) |
| 181 self.assertEquals(None, err) |
| 182 self.assertEquals(0, code) |
155 | 183 |
156 def test_check_output_throw(self): | 184 def test_check_output_throw_stdout(self): |
| 185 try: |
| 186 subprocess2.check_output( |
| 187 self.exe + ['--fail', '--stdout'], universal_newlines=True) |
| 188 self.fail() |
| 189 except subprocess2.CalledProcessError, e: |
| 190 self.assertEquals('A\nBB\nCCC\n', e.stdout) |
| 191 self.assertEquals(None, e.stderr) |
| 192 self.assertEquals(64, e.returncode) |
| 193 |
| 194 def test_check_output_throw_no_stderr(self): |
157 try: | 195 try: |
158 subprocess2.check_output( | 196 subprocess2.check_output( |
159 self.exe + ['--fail', '--stderr'], universal_newlines=True) | 197 self.exe + ['--fail', '--stderr'], universal_newlines=True) |
160 self.fail() | 198 self.fail() |
161 except subprocess2.CalledProcessError, e: | 199 except subprocess2.CalledProcessError, e: |
| 200 self.assertEquals('', e.stdout) |
| 201 self.assertEquals(None, e.stderr) |
| 202 self.assertEquals(64, e.returncode) |
| 203 |
| 204 def test_check_output_throw_stderr(self): |
| 205 try: |
| 206 subprocess2.check_output( |
| 207 self.exe + ['--fail', '--stderr'], stderr=subprocess2.PIPE, |
| 208 universal_newlines=True) |
| 209 self.fail() |
| 210 except subprocess2.CalledProcessError, e: |
| 211 self.assertEquals('', e.stdout) |
| 212 self.assertEquals('a\nbb\nccc\n', e.stderr) |
| 213 self.assertEquals(64, e.returncode) |
| 214 |
| 215 def test_check_output_throw_stderr_stdout(self): |
| 216 try: |
| 217 subprocess2.check_output( |
| 218 self.exe + ['--fail', '--stderr'], stderr=subprocess2.STDOUT, |
| 219 universal_newlines=True) |
| 220 self.fail() |
| 221 except subprocess2.CalledProcessError, e: |
162 self.assertEquals('a\nbb\nccc\n', e.stdout) | 222 self.assertEquals('a\nbb\nccc\n', e.stdout) |
163 self.assertEquals(None, e.stderr) | 223 self.assertEquals(None, e.stderr) |
164 self.assertEquals(64, e.returncode) | 224 self.assertEquals(64, e.returncode) |
165 | 225 |
166 def test_check_call_throw(self): | 226 def test_check_call_throw(self): |
167 try: | 227 try: |
168 subprocess2.check_call(self.exe + ['--fail', '--stderr']) | 228 subprocess2.check_call(self.exe + ['--fail', '--stderr']) |
169 self.fail() | 229 self.fail() |
170 except subprocess2.CalledProcessError, e: | 230 except subprocess2.CalledProcessError, e: |
171 self.assertEquals(None, e.stdout) | 231 self.assertEquals(None, e.stdout) |
(...skipping 27 matching lines...) Expand all Loading... |
199 do('CCC') | 259 do('CCC') |
200 if options.sleep: | 260 if options.sleep: |
201 time.sleep(10) | 261 time.sleep(10) |
202 return options.return_value | 262 return options.return_value |
203 | 263 |
204 | 264 |
205 if __name__ == '__main__': | 265 if __name__ == '__main__': |
206 if len(sys.argv) > 1 and sys.argv[1] == '--child': | 266 if len(sys.argv) > 1 and sys.argv[1] == '--child': |
207 sys.exit(child_main(sys.argv[2:])) | 267 sys.exit(child_main(sys.argv[2:])) |
208 unittest.main() | 268 unittest.main() |
OLD | NEW |