| 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 |
| 11 import time | 11 import time |
| 12 import unittest | 12 import unittest |
| 13 | 13 |
| 14 ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) | 14 ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) |
| 15 sys.path.insert(0, ROOT_DIR) | 15 sys.path.insert(0, ROOT_DIR) |
| 16 | 16 |
| 17 import subprocess2 | 17 import subprocess2 |
| 18 | 18 |
| 19 # Method could be a function | 19 # Method could be a function |
| 20 # pylint: disable=R0201 | 20 # pylint: disable=R0201 |
| 21 | 21 |
| 22 class Subprocess2Test(unittest.TestCase): | 22 class Subprocess2Test(unittest.TestCase): |
| 23 # Can be mocked in a test. | 23 # Can be mocked in a test. |
| 24 TO_SAVE = { | 24 TO_SAVE = { |
| 25 subprocess2: ['Popen', 'call', 'check_call', 'capture', 'check_output'], | 25 subprocess2: [ |
| 26 subprocess2.subprocess: ['Popen'], | 26 'Popen', 'communicate', 'call', 'check_call', 'capture', 'check_output'], |
| 27 subprocess2.subprocess: ['Popen'], |
| 27 } | 28 } |
| 28 | 29 |
| 29 def setUp(self): | 30 def setUp(self): |
| 30 self.exe_path = __file__ | 31 self.exe_path = __file__ |
| 31 self.exe = [sys.executable, self.exe_path, '--child'] | 32 self.exe = [sys.executable, self.exe_path, '--child'] |
| 32 self.saved = {} | 33 self.saved = {} |
| 33 for module, names in self.TO_SAVE.iteritems(): | 34 for module, names in self.TO_SAVE.iteritems(): |
| 34 self.saved[module] = dict( | 35 self.saved[module] = dict( |
| 35 (name, getattr(module, name)) for name in names) | 36 (name, getattr(module, name)) for name in names) |
| 36 | 37 |
| 37 def tearDown(self): | 38 def tearDown(self): |
| 38 for module, saved in self.saved.iteritems(): | 39 for module, saved in self.saved.iteritems(): |
| 39 for name, value in saved.iteritems(): | 40 for name, value in saved.iteritems(): |
| 40 setattr(module, name, value) | 41 setattr(module, name, value) |
| 41 | 42 |
| 42 @staticmethod | 43 @staticmethod |
| 43 def _fake_call(): | 44 def _fake_communicate(): |
| 44 results = {} | 45 results = {} |
| 45 def fake_call(args, **kwargs): | 46 def fake_communicate(args, **kwargs): |
| 46 assert not results | 47 assert not results |
| 47 results.update(kwargs) | 48 results.update(kwargs) |
| 48 results['args'] = args | 49 results['args'] = args |
| 49 return ['stdout', 'stderr'], 0 | 50 return ['stdout', 'stderr'], 0 |
| 50 subprocess2.call = fake_call | 51 subprocess2.communicate = fake_communicate |
| 51 return results | 52 return results |
| 52 | 53 |
| 53 @staticmethod | 54 @staticmethod |
| 54 def _fake_Popen(): | 55 def _fake_Popen(): |
| 55 results = {} | 56 results = {} |
| 56 class fake_Popen(object): | 57 class fake_Popen(object): |
| 57 returncode = -8 | 58 returncode = -8 |
| 58 def __init__(self, args, **kwargs): | 59 def __init__(self, args, **kwargs): |
| 59 assert not results | 60 assert not results |
| 60 results.update(kwargs) | 61 results.update(kwargs) |
| (...skipping 11 matching lines...) Expand all Loading... |
| 72 def __init__(self, args, **kwargs): | 73 def __init__(self, args, **kwargs): |
| 73 assert not results | 74 assert not results |
| 74 results.update(kwargs) | 75 results.update(kwargs) |
| 75 results['args'] = args | 76 results['args'] = args |
| 76 def communicate(self): | 77 def communicate(self): |
| 77 return None, None | 78 return None, None |
| 78 subprocess2.subprocess.Popen = fake_Popen | 79 subprocess2.subprocess.Popen = fake_Popen |
| 79 return results | 80 return results |
| 80 | 81 |
| 81 def test_check_call_defaults(self): | 82 def test_check_call_defaults(self): |
| 82 results = self._fake_call() | 83 results = self._fake_communicate() |
| 83 self.assertEquals( | 84 self.assertEquals( |
| 84 ['stdout', 'stderr'], subprocess2.check_call(['foo'], a=True)) | 85 ['stdout', 'stderr'], subprocess2.check_call(['foo'], a=True)) |
| 85 expected = { | 86 expected = { |
| 86 'args': ['foo'], | 87 'args': ['foo'], |
| 87 'a':True, | 88 'a':True, |
| 88 } | 89 } |
| 89 self.assertEquals(expected, results) | 90 self.assertEquals(expected, results) |
| 90 | 91 |
| 91 def test_call_defaults(self): | 92 def test_communicate_defaults(self): |
| 92 results = self._fake_Popen() | 93 results = self._fake_Popen() |
| 93 self.assertEquals(((None, None), -8), subprocess2.call(['foo'], a=True)) | 94 self.assertEquals( |
| 95 ((None, None), -8), subprocess2.communicate(['foo'], a=True)) |
| 94 expected = { | 96 expected = { |
| 95 'args': ['foo'], | 97 'args': ['foo'], |
| 96 'a': True, | 98 'a': True, |
| 97 } | 99 } |
| 98 self.assertEquals(expected, results) | 100 self.assertEquals(expected, results) |
| 99 | 101 |
| 100 def test_Popen_defaults(self): | 102 def test_Popen_defaults(self): |
| 101 results = self._fake_subprocess_Popen() | 103 results = self._fake_subprocess_Popen() |
| 102 proc = subprocess2.Popen(['foo'], a=True) | 104 proc = subprocess2.Popen(['foo'], a=True) |
| 103 self.assertEquals(-8, proc.returncode) | 105 self.assertEquals(-8, proc.returncode) |
| 104 expected = { | 106 expected = { |
| 105 'args': ['foo'], | 107 'args': ['foo'], |
| 106 'a': True, | 108 'a': True, |
| 107 'shell': bool(sys.platform=='win32'), | 109 'shell': bool(sys.platform=='win32'), |
| 108 } | 110 } |
| 109 if sys.platform != 'win32': | 111 if sys.platform != 'win32': |
| 110 env = os.environ.copy() | 112 env = os.environ.copy() |
| 111 is_english = lambda name: env.get(name, 'en').startswith('en') | 113 is_english = lambda name: env.get(name, 'en').startswith('en') |
| 112 if not is_english('LANG'): | 114 if not is_english('LANG'): |
| 113 env['LANG'] = 'en_US.UTF-8' | 115 env['LANG'] = 'en_US.UTF-8' |
| 114 expected['env'] = env | 116 expected['env'] = env |
| 115 if not is_english('LANGUAGE'): | 117 if not is_english('LANGUAGE'): |
| 116 env['LANGUAGE'] = 'en_US.UTF-8' | 118 env['LANGUAGE'] = 'en_US.UTF-8' |
| 117 expected['env'] = env | 119 expected['env'] = env |
| 118 self.assertEquals(expected, results) | 120 self.assertEquals(expected, results) |
| 119 | 121 |
| 120 def test_check_output_defaults(self): | 122 def test_check_output_defaults(self): |
| 121 results = self._fake_call() | 123 results = self._fake_communicate() |
| 122 # It's discarding 'stderr' because it assumes stderr=subprocess2.STDOUT but | 124 # It's discarding 'stderr' because it assumes stderr=subprocess2.STDOUT but |
| 123 # fake_call() doesn't 'implement' that. | 125 # fake_communicate() doesn't 'implement' that. |
| 124 self.assertEquals('stdout', subprocess2.check_output(['foo'], a=True)) | 126 self.assertEquals('stdout', subprocess2.check_output(['foo'], a=True)) |
| 125 expected = { | 127 expected = { |
| 126 'args': ['foo'], | 128 'args': ['foo'], |
| 127 'a':True, | 129 'a':True, |
| 128 'stdin': subprocess2.VOID, | 130 'stdin': subprocess2.VOID, |
| 129 'stdout': subprocess2.PIPE, | 131 'stdout': subprocess2.PIPE, |
| 130 'stderr': subprocess2.STDOUT, | 132 'stderr': subprocess2.STDOUT, |
| 131 } | 133 } |
| 132 self.assertEquals(expected, results) | 134 self.assertEquals(expected, results) |
| 133 | 135 |
| 134 def test_timeout(self): | 136 def test_timeout(self): |
| 135 # It'd be better to not discard stdout. | 137 # It'd be better to not discard stdout. |
| 136 out, returncode = subprocess2.call( | 138 out, returncode = subprocess2.communicate( |
| 137 self.exe + ['--sleep', '--stdout'], | 139 self.exe + ['--sleep', '--stdout'], |
| 138 timeout=0.01, | 140 timeout=0.01, |
| 139 stdout=subprocess2.PIPE) | 141 stdout=subprocess2.PIPE) |
| 140 self.assertEquals(subprocess2.TIMED_OUT, returncode) | 142 self.assertEquals(subprocess2.TIMED_OUT, returncode) |
| 141 self.assertEquals(['', None], out) | 143 self.assertEquals(['', None], out) |
| 142 | 144 |
| 143 def test_void(self): | 145 def test_void(self): |
| 144 out = subprocess2.check_output( | 146 out = subprocess2.check_output( |
| 145 self.exe + ['--stdout', '--stderr'], | 147 self.exe + ['--stdout', '--stderr'], |
| 146 stdout=subprocess2.VOID) | 148 stdout=subprocess2.VOID) |
| (...skipping 50 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 197 do('CCC') | 199 do('CCC') |
| 198 if options.sleep: | 200 if options.sleep: |
| 199 time.sleep(10) | 201 time.sleep(10) |
| 200 return options.return_value | 202 return options.return_value |
| 201 | 203 |
| 202 | 204 |
| 203 if __name__ == '__main__': | 205 if __name__ == '__main__': |
| 204 if len(sys.argv) > 1 and sys.argv[1] == '--child': | 206 if len(sys.argv) > 1 and sys.argv[1] == '--child': |
| 205 sys.exit(child_main(sys.argv[2:])) | 207 sys.exit(child_main(sys.argv[2:])) |
| 206 unittest.main() | 208 unittest.main() |
| OLD | NEW |