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 logging | 8 import logging |
9 import optparse | 9 import optparse |
10 import os | 10 import os |
(...skipping 29 matching lines...) Expand all Loading... |
40 if sys.platform == 'win32': | 40 if sys.platform == 'win32': |
41 return string.replace('\n', '\r\n') | 41 return string.replace('\n', '\r\n') |
42 return string | 42 return string |
43 | 43 |
44 | 44 |
45 class DefaultsTest(unittest.TestCase): | 45 class DefaultsTest(unittest.TestCase): |
46 # Can be mocked in a test. | 46 # Can be mocked in a test. |
47 TO_SAVE = { | 47 TO_SAVE = { |
48 subprocess2: [ | 48 subprocess2: [ |
49 'Popen', 'communicate', 'call', 'check_call', 'capture', 'check_output'], | 49 'Popen', 'communicate', 'call', 'check_call', 'capture', 'check_output'], |
50 subprocess2.subprocess: ['Popen'], | 50 subprocess2.subprocess.Popen: ['__init__', 'communicate'], |
51 } | 51 } |
52 | 52 |
53 def setUp(self): | 53 def setUp(self): |
54 self.saved = {} | 54 self.saved = {} |
55 for module, names in self.TO_SAVE.iteritems(): | 55 for module, names in self.TO_SAVE.iteritems(): |
56 self.saved[module] = dict( | 56 self.saved[module] = dict( |
57 (name, getattr(module, name)) for name in names) | 57 (name, getattr(module, name)) for name in names) |
58 # TODO(maruel): Do a reopen() on sys.__stdout__ and sys.__stderr__ so they | 58 # TODO(maruel): Do a reopen() on sys.__stdout__ and sys.__stderr__ so they |
59 # can be trapped in the child process for better coverage. | 59 # can be trapped in the child process for better coverage. |
60 | 60 |
61 def tearDown(self): | 61 def tearDown(self): |
62 for module, saved in self.saved.iteritems(): | 62 for module, saved in self.saved.iteritems(): |
63 for name, value in saved.iteritems(): | 63 for name, value in saved.iteritems(): |
64 setattr(module, name, value) | 64 setattr(module, name, value) |
65 | 65 |
66 @staticmethod | 66 @staticmethod |
67 def _fake_communicate(): | 67 def _fake_communicate(): |
| 68 """Mocks subprocess2.communicate().""" |
68 results = {} | 69 results = {} |
69 def fake_communicate(args, **kwargs): | 70 def fake_communicate(args, **kwargs): |
70 assert not results | 71 assert not results |
71 results.update(kwargs) | 72 results.update(kwargs) |
72 results['args'] = args | 73 results['args'] = args |
73 return ['stdout', 'stderr'], 0 | 74 return ('stdout', 'stderr'), 0 |
74 subprocess2.communicate = fake_communicate | 75 subprocess2.communicate = fake_communicate |
75 return results | 76 return results |
76 | 77 |
77 @staticmethod | 78 @staticmethod |
78 def _fake_Popen(): | 79 def _fake_Popen(): |
| 80 """Mocks the whole subprocess2.Popen class.""" |
79 results = {} | 81 results = {} |
80 class fake_Popen(object): | 82 class fake_Popen(object): |
81 returncode = -8 | 83 returncode = -8 |
82 def __init__(self, args, **kwargs): | 84 def __init__(self, args, **kwargs): |
83 assert not results | 85 assert not results |
84 results.update(kwargs) | 86 results.update(kwargs) |
85 results['args'] = args | 87 results['args'] = args |
86 @staticmethod | 88 @staticmethod |
87 def communicate(): | 89 def communicate(): |
88 return None, None | 90 return None, None |
89 subprocess2.Popen = fake_Popen | 91 subprocess2.Popen = fake_Popen |
90 return results | 92 return results |
91 | 93 |
92 @staticmethod | 94 @staticmethod |
93 def _fake_subprocess_Popen(): | 95 def _fake_subprocess_Popen(): |
| 96 """Mocks the base class subprocess.Popen only.""" |
94 results = {} | 97 results = {} |
95 class fake_Popen(object): | 98 def __init__(self, args, **kwargs): |
96 returncode = -8 | 99 assert not results |
97 def __init__(self, args, **kwargs): | 100 results.update(kwargs) |
98 assert not results | 101 results['args'] = args |
99 results.update(kwargs) | 102 def communicate(): |
100 results['args'] = args | 103 return None, None |
101 @staticmethod | 104 subprocess2.subprocess.Popen.__init__ = __init__ |
102 def communicate(): | 105 subprocess2.subprocess.Popen.communicate = communicate |
103 return None, None | |
104 subprocess2.subprocess.Popen = fake_Popen | |
105 return results | 106 return results |
106 | 107 |
107 def test_check_call_defaults(self): | 108 def test_check_call_defaults(self): |
108 results = self._fake_communicate() | 109 results = self._fake_communicate() |
109 self.assertEquals( | 110 self.assertEquals( |
110 ['stdout', 'stderr'], subprocess2.check_call_out(['foo'], a=True)) | 111 ('stdout', 'stderr'), subprocess2.check_call_out(['foo'], a=True)) |
111 expected = { | 112 expected = { |
112 'args': ['foo'], | 113 'args': ['foo'], |
113 'a':True, | 114 'a':True, |
114 } | 115 } |
115 self.assertEquals(expected, results) | 116 self.assertEquals(expected, results) |
116 | 117 |
117 def test_capture_defaults(self): | 118 def test_capture_defaults(self): |
118 results = self._fake_communicate() | 119 results = self._fake_communicate() |
119 self.assertEquals( | 120 self.assertEquals( |
120 'stdout', subprocess2.capture(['foo'], a=True)) | 121 'stdout', subprocess2.capture(['foo'], a=True)) |
(...skipping 11 matching lines...) Expand all Loading... |
132 ((None, None), -8), subprocess2.communicate(['foo'], a=True)) | 133 ((None, None), -8), subprocess2.communicate(['foo'], a=True)) |
133 expected = { | 134 expected = { |
134 'args': ['foo'], | 135 'args': ['foo'], |
135 'a': True, | 136 'a': True, |
136 } | 137 } |
137 self.assertEquals(expected, results) | 138 self.assertEquals(expected, results) |
138 | 139 |
139 def test_Popen_defaults(self): | 140 def test_Popen_defaults(self): |
140 results = self._fake_subprocess_Popen() | 141 results = self._fake_subprocess_Popen() |
141 proc = subprocess2.Popen(['foo'], a=True) | 142 proc = subprocess2.Popen(['foo'], a=True) |
142 self.assertEquals(-8, proc.returncode) | 143 # Cleanup code in subprocess.py needs this member to be set. |
| 144 # pylint: disable=W0201 |
| 145 proc._child_created = None |
| 146 # Since subprocess.Popen.__init__() is not called, proc.returncode shouldn't |
| 147 # be present. |
| 148 self.assertFalse(hasattr(proc, 'returncode')) |
143 expected = { | 149 expected = { |
144 'args': ['foo'], | 150 'args': ['foo'], |
145 'a': True, | 151 'a': True, |
146 'shell': bool(sys.platform=='win32'), | 152 'shell': bool(sys.platform=='win32'), |
147 } | 153 } |
148 if sys.platform != 'win32': | 154 if sys.platform != 'win32': |
149 env = os.environ.copy() | 155 env = os.environ.copy() |
150 is_english = lambda name: env.get(name, 'en').startswith('en') | 156 is_english = lambda name: env.get(name, 'en').startswith('en') |
151 if not is_english('LANG'): | 157 if not is_english('LANG'): |
152 env['LANG'] = 'en_US.UTF-8' | 158 env['LANG'] = 'en_US.UTF-8' |
(...skipping 213 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
366 return options.return_value | 372 return options.return_value |
367 | 373 |
368 | 374 |
369 if __name__ == '__main__': | 375 if __name__ == '__main__': |
370 logging.basicConfig(level= | 376 logging.basicConfig(level= |
371 [logging.WARNING, logging.INFO, logging.DEBUG][ | 377 [logging.WARNING, logging.INFO, logging.DEBUG][ |
372 min(2, sys.argv.count('-v'))]) | 378 min(2, sys.argv.count('-v'))]) |
373 if len(sys.argv) > 1 and sys.argv[1] == '--child': | 379 if len(sys.argv) > 1 and sys.argv[1] == '--child': |
374 sys.exit(child_main(sys.argv[2:])) | 380 sys.exit(child_main(sys.argv[2:])) |
375 unittest.main() | 381 unittest.main() |
OLD | NEW |