| 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 |