OLD | NEW |
| (Empty) |
1 #!/usr/bin/env python | |
2 # Copyright (c) 2012 The Chromium Authors. All rights reserved. | |
3 # Use of this source code is governed by a BSD-style license that can be | |
4 # found in the LICENSE file. | |
5 | |
6 """Unit tests for commit_queue.py.""" | |
7 | |
8 import logging | |
9 import os | |
10 import StringIO | |
11 import sys | |
12 import time | |
13 import traceback | |
14 import unittest | |
15 | |
16 ROOT_DIR = os.path.dirname(os.path.abspath(__file__)) | |
17 sys.path.insert(0, os.path.join(ROOT_DIR, '..')) | |
18 | |
19 import commit_queue | |
20 import context | |
21 import creds | |
22 | |
23 from testing_support import auto_stub | |
24 | |
25 # From /tests | |
26 import mocks | |
27 | |
28 | |
29 class Stop(Exception): | |
30 pass | |
31 | |
32 | |
33 class PendingManagerMock(auto_stub.SimpleMock): | |
34 def __init__(self, unit_test): | |
35 super(PendingManagerMock, self).__init__(unit_test) | |
36 self.context = context.Context( | |
37 mocks.RietveldMock(unit_test), mocks.SvnCheckoutMock(unit_test), None) | |
38 self.count = 0 | |
39 | |
40 def load(self, *args, **kwargs): | |
41 self._register_call(*args, **kwargs) | |
42 | |
43 def save(self, *args, **kwargs): | |
44 self._register_call(*args, **kwargs) | |
45 | |
46 def close(self, *args, **kwargs): | |
47 self._register_call(*args, **kwargs) | |
48 | |
49 def look_for_new_pending_commit(self, *args, **kwargs): | |
50 self._register_call(*args, **kwargs) | |
51 self.count += 1 | |
52 if self.count > 3: | |
53 raise Stop() | |
54 | |
55 def process_new_pending_commit(self, *args, **kwargs): | |
56 self._register_call(*args, **kwargs) | |
57 | |
58 def update_status(self, *args, **kwargs): | |
59 self._register_call(*args, **kwargs) | |
60 | |
61 def scan_results(self, *args, **kwargs): | |
62 self._register_call(*args, **kwargs) | |
63 | |
64 | |
65 class CredentialsMock(object): | |
66 @staticmethod | |
67 def get(user): | |
68 return '1%s1' % user | |
69 | |
70 | |
71 class CommitQueueTest(auto_stub.TestCase): | |
72 def setUp(self): | |
73 super(CommitQueueTest, self).setUp() | |
74 self.mock(sys, 'argv', ['commit_queue.py']) | |
75 self.mock(sys, 'stdout', StringIO.StringIO()) | |
76 self.mock(sys, 'stderr', StringIO.StringIO()) | |
77 self.mock(commit_queue.projects, 'load_project', None) | |
78 self.mock(commit_queue, 'SetupLogging', lambda _: None) | |
79 # Setup logging attached to the mocked sys.stderr, printing | |
80 # only the exception name to make tests less fragile. | |
81 handler = logging.StreamHandler() | |
82 formatter = logging.Formatter() | |
83 formatter.formatException = lambda _: traceback.format_exc(0) | |
84 handler.setFormatter(formatter) | |
85 logging.getLogger().handlers = [handler] | |
86 self._time = 1 | |
87 self.mock(time, 'time', self._get_time) | |
88 self.mock(creds, 'Credentials', self._get_cred) | |
89 | |
90 def tearDown(self): | |
91 try: | |
92 if not self.has_failed(): | |
93 self._check('stdout', '') | |
94 self._check('stderr', '') | |
95 finally: | |
96 super(CommitQueueTest, self).tearDown() | |
97 | |
98 def _check(self, pipe, expected): | |
99 self.assertEqual(expected, self._pop(pipe)) | |
100 | |
101 def _get_time(self): | |
102 self._time += 10 | |
103 return self._time | |
104 | |
105 @staticmethod | |
106 def _pop(pipe): | |
107 data = getattr(sys, pipe).getvalue() | |
108 setattr(sys, pipe, StringIO.StringIO()) | |
109 return data | |
110 | |
111 def _get_cred(self, pwd): | |
112 rootdir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) | |
113 workdir = os.path.join(rootdir, 'workdir') | |
114 self.assertEqual(os.path.join(workdir, '.gaia_pwd'), pwd) | |
115 return CredentialsMock() | |
116 | |
117 def testHelp(self): | |
118 sys.argv.append('--help') | |
119 try: | |
120 commit_queue.main() | |
121 self.fail() | |
122 except SystemExit as e: | |
123 self.assertEqual(0, e.code) | |
124 output = self._pop('stdout') | |
125 # Cannot compare for the exact string since the formatting depends on the | |
126 # screen size. | |
127 self.assertIn('Minimum delay between each polling loop', output) | |
128 self.assertIn('Run for real instead of dry-run mode which', output) | |
129 self.assertLess(600, len(output), output) | |
130 | |
131 def testChromium(self): | |
132 sys.argv.extend(('--project', 'chromium')) | |
133 calls = [] | |
134 def load_project(*args): | |
135 calls.append(args) | |
136 return PendingManagerMock(self) | |
137 | |
138 self.mock(commit_queue.projects, 'load_project', load_project) | |
139 try: | |
140 commit_queue.main() | |
141 self.fail() | |
142 except Stop: | |
143 pass | |
144 self.assertEqual(1, len(calls)) | |
145 self.assertEqual('chromium', calls[0][0]) | |
146 self.assertEqual('commit-bot@chromium.org', calls[0][1]) | |
147 self.assertEqual( | |
148 os.path.join(os.path.dirname(ROOT_DIR), 'workdir'), calls[0][2]) | |
149 self.assertEqual(None, calls[0][4]) | |
150 self._check( | |
151 'stdout', | |
152 'Using read-only Rietveld\n' | |
153 'Using read-only checkout\n' | |
154 'Using read-only chromium-status interface\n') | |
155 self._check( | |
156 'stderr', | |
157 'CQ loop terminating\n' | |
158 'Traceback (most recent call last):\n' | |
159 'Stop\n\n' | |
160 'Saving db...\ndb save successful.\n') | |
161 | |
162 def testDryRun(self): | |
163 sys.argv.extend(('--project', 'chromium')) | |
164 pc = PendingManagerMock(self) | |
165 self.mock( | |
166 commit_queue.projects, | |
167 'load_project', | |
168 lambda *args: pc) | |
169 try: | |
170 commit_queue.main() | |
171 self.fail() | |
172 except Stop: | |
173 pass | |
174 self.assertEqual( | |
175 'ReadOnlyCheckout', pc.context.checkout.__class__.__name__) | |
176 # Ugh. | |
177 self.assertEqual( | |
178 'RietveldMock', pc.context.rietveld.__class__.__name__) | |
179 self._check( | |
180 'stdout', | |
181 'Using read-only Rietveld\n' | |
182 'Using read-only checkout\n' | |
183 'Using read-only chromium-status interface\n') | |
184 self._check( | |
185 'stderr', | |
186 'CQ loop terminating\n' | |
187 'Traceback (most recent call last):\n' | |
188 'Stop\n\n' | |
189 'Saving db...\ndb save successful.\n') | |
190 | |
191 | |
192 if __name__ == '__main__': | |
193 unittest.main() | |
OLD | NEW |