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 |