| OLD | NEW |
| 1 #!/usr/bin/python | 1 #!/usr/bin/python |
| 2 # Copyright (c) 2010 The Chromium Authors. All rights reserved. | 2 # Copyright (c) 2010 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 import StringIO | 6 import StringIO |
| 7 | 7 |
| 8 # Fixes include path. | 8 # Fixes include path. |
| 9 from super_mox import SuperMoxTestBase | 9 from super_mox import SuperMoxTestBase |
| 10 | 10 |
| (...skipping 60 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 71 self.fail() | 71 self.fail() |
| 72 except gclient_utils.CheckCallError, e: | 72 except gclient_utils.CheckCallError, e: |
| 73 self.assertEqual(e.command, command) | 73 self.assertEqual(e.command, command) |
| 74 self.assertEqual(e.cwd, None) | 74 self.assertEqual(e.cwd, None) |
| 75 self.assertEqual(e.retcode, 42) | 75 self.assertEqual(e.retcode, 42) |
| 76 self.assertEqual(e.stdout, 'bleh') | 76 self.assertEqual(e.stdout, 'bleh') |
| 77 self.assertEqual(e.stderr, 'foo') | 77 self.assertEqual(e.stderr, 'foo') |
| 78 | 78 |
| 79 | 79 |
| 80 class SubprocessCallAndFilterTestCase(GclientUtilBase): | 80 class SubprocessCallAndFilterTestCase(GclientUtilBase): |
| 81 def testSubprocessCallAndFilter(self): | 81 class ProcessIdMock(object): |
| 82 command = ['boo', 'foo', 'bar'] | 82 def __init__(self, test_string): |
| 83 self.stdout = StringIO.StringIO(test_string) |
| 84 def wait(self): |
| 85 pass |
| 86 |
| 87 def _inner(self, command, test_string): |
| 83 in_directory = 'bleh' | 88 in_directory = 'bleh' |
| 84 fail_status = None | |
| 85 pattern = 'a(.*)b' | |
| 86 test_string = 'ahah\naccb\nallo\naddb\n' | |
| 87 env = gclient_utils.os.environ.copy() | 89 env = gclient_utils.os.environ.copy() |
| 88 env['LANGUAGE'] = 'en' | 90 env['LANGUAGE'] = 'en' |
| 89 class Mock(object): | |
| 90 stdout = StringIO.StringIO(test_string) | |
| 91 def wait(self): | |
| 92 pass | |
| 93 kid = Mock() | |
| 94 print("\n________ running 'boo foo bar' in 'bleh'") | 91 print("\n________ running 'boo foo bar' in 'bleh'") |
| 95 for i in test_string: | 92 for i in test_string: |
| 96 gclient_utils.sys.stdout.write(i) | 93 gclient_utils.sys.stdout.write(i) |
| 97 gclient_utils.subprocess.Popen( | 94 gclient_utils.subprocess.Popen( |
| 98 command, bufsize=0, cwd=in_directory, | 95 command, |
| 96 cwd=in_directory, |
| 99 shell=(gclient_utils.sys.platform == 'win32'), | 97 shell=(gclient_utils.sys.platform == 'win32'), |
| 100 env=env, | 98 env=env, |
| 101 stdout=gclient_utils.subprocess.PIPE, | 99 stdout=gclient_utils.subprocess.PIPE, |
| 102 stderr=gclient_utils.subprocess.STDOUT).AndReturn(kid) | 100 stderr=gclient_utils.subprocess.STDOUT, |
| 101 bufsize=0).AndReturn(self.ProcessIdMock(test_string)) |
| 102 |
| 103 self.mox.ReplayAll() | 103 self.mox.ReplayAll() |
| 104 compiled_pattern = gclient_utils.re.compile(pattern) | 104 compiled_pattern = gclient_utils.re.compile(r'a(.*)b') |
| 105 line_list = [] | 105 line_list = [] |
| 106 capture_list = [] | 106 capture_list = [] |
| 107 def FilterLines(line): | 107 def FilterLines(line): |
| 108 line_list.append(line) | 108 line_list.append(line) |
| 109 match = compiled_pattern.search(line) | 109 match = compiled_pattern.search(line) |
| 110 if match: | 110 if match: |
| 111 capture_list.append(match.group(1)) | 111 capture_list.append(match.group(1)) |
| 112 gclient_utils.SubprocessCallAndFilter( | 112 gclient_utils.SubprocessCallAndFilter( |
| 113 command, in_directory, | 113 command, in_directory, True, True, None, FilterLines) |
| 114 True, True, | |
| 115 fail_status, FilterLines) | |
| 116 self.assertEquals(line_list, ['ahah', 'accb', 'allo', 'addb']) | 114 self.assertEquals(line_list, ['ahah', 'accb', 'allo', 'addb']) |
| 117 self.assertEquals(capture_list, ['cc', 'dd']) | 115 self.assertEquals(capture_list, ['cc', 'dd']) |
| 118 | 116 |
| 117 def testSubprocessCallAndFilter(self): |
| 118 command = ['boo', 'foo', 'bar'] |
| 119 test_string = 'ahah\naccb\nallo\naddb\n' |
| 120 self._inner(command, test_string) |
| 121 |
| 122 def testNoLF(self): |
| 123 # Exactly as testSubprocessCallAndFilter without trailing \n |
| 124 command = ['boo', 'foo', 'bar'] |
| 125 test_string = 'ahah\naccb\nallo\naddb' |
| 126 self._inner(command, test_string) |
| 127 |
| 119 | 128 |
| 120 class SplitUrlRevisionTestCase(GclientUtilBase): | 129 class SplitUrlRevisionTestCase(GclientUtilBase): |
| 121 def testSSHUrl(self): | 130 def testSSHUrl(self): |
| 122 url = "ssh://test@example.com/test.git" | 131 url = "ssh://test@example.com/test.git" |
| 123 rev = "ac345e52dc" | 132 rev = "ac345e52dc" |
| 124 out_url, out_rev = gclient_utils.SplitUrlRevision(url) | 133 out_url, out_rev = gclient_utils.SplitUrlRevision(url) |
| 125 self.assertEquals(out_rev, None) | 134 self.assertEquals(out_rev, None) |
| 126 self.assertEquals(out_url, url) | 135 self.assertEquals(out_url, url) |
| 127 out_url, out_rev = gclient_utils.SplitUrlRevision("%s@%s" % (url, rev)) | 136 out_url, out_rev = gclient_utils.SplitUrlRevision("%s@%s" % (url, rev)) |
| 128 self.assertEquals(out_rev, rev) | 137 self.assertEquals(out_rev, rev) |
| (...skipping 26 matching lines...) Expand all Loading... |
| 155 out_url, out_rev = gclient_utils.SplitUrlRevision("%s@%s" % (url, rev)) | 164 out_url, out_rev = gclient_utils.SplitUrlRevision("%s@%s" % (url, rev)) |
| 156 self.assertEquals(out_rev, rev) | 165 self.assertEquals(out_rev, rev) |
| 157 self.assertEquals(out_url, url) | 166 self.assertEquals(out_url, url) |
| 158 | 167 |
| 159 | 168 |
| 160 if __name__ == '__main__': | 169 if __name__ == '__main__': |
| 161 import unittest | 170 import unittest |
| 162 unittest.main() | 171 unittest.main() |
| 163 | 172 |
| 164 # vim: ts=2:sw=2:tw=80:et: | 173 # vim: ts=2:sw=2:tw=80:et: |
| OLD | NEW |