• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/python
2# Copyright (c) 2014 The Chromium OS 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"""Unittests for deploy_server_local.py."""
7
8from __future__ import print_function
9
10import mock
11import subprocess
12import unittest
13
14import deploy_server_local as dsl
15
16
17class TestDeployServerLocal(unittest.TestCase):
18    """Test deploy_server_local with commands mocked out."""
19
20    orig_timer = dsl.SERVICE_STABILITY_TIMER
21
22    PROD_STATUS = ('\x1b[1mproject autotest/                             '
23                   '  \x1b[m\x1b[1mbranch prod\x1b[m\n')
24
25    PROD_VERSIONS = '''\x1b[1mproject autotest/\x1b[m
26/usr/local/autotest
27ebb2182
28
29\x1b[1mproject autotest/site_utils/autotest_private/\x1b[m
30/usr/local/autotest/site_utils/autotest_private
3178b9626
32
33\x1b[1mproject autotest/site_utils/autotest_tools/\x1b[m
34/usr/local/autotest/site_utils/autotest_tools
35a1598f7
36'''
37
38
39    def setUp(self):
40        dsl.SERVICE_STABILITY_TIMER = 0.01
41
42    def tearDown(self):
43        dsl.SERVICE_STABILITY_TIMER = self.orig_timer
44
45    def test_strip_terminal_codes(self):
46        """Test deploy_server_local.strip_terminal_codes."""
47        # Leave format free lines alone.
48        result = dsl.strip_terminal_codes('')
49        self.assertEqual(result, '')
50
51        result = dsl.strip_terminal_codes('This is normal text.')
52        self.assertEqual(result, 'This is normal text.')
53
54        result = dsl.strip_terminal_codes('Line1\nLine2\n')
55        self.assertEqual(result, 'Line1\nLine2\n')
56
57        result = dsl.strip_terminal_codes('Line1\nLine2\n')
58        self.assertEqual(result, 'Line1\nLine2\n')
59
60        # Test cleaning lines with formatting.
61        result = dsl.strip_terminal_codes('\x1b[1m')
62        self.assertEqual(result, '')
63
64        result = dsl.strip_terminal_codes('\x1b[m')
65        self.assertEqual(result, '')
66
67        result = dsl.strip_terminal_codes('\x1b[1mm')
68        self.assertEqual(result, 'm')
69
70        result = dsl.strip_terminal_codes(self.PROD_STATUS)
71        self.assertEqual(result,
72                'project autotest/                               branch prod\n')
73
74        result = dsl.strip_terminal_codes(self.PROD_VERSIONS)
75        self.assertEqual(result, '''project autotest/
76/usr/local/autotest
77ebb2182
78
79project autotest/site_utils/autotest_private/
80/usr/local/autotest/site_utils/autotest_private
8178b9626
82
83project autotest/site_utils/autotest_tools/
84/usr/local/autotest/site_utils/autotest_tools
85a1598f7
86''')
87
88    @mock.patch('subprocess.check_output', autospec=True)
89    def test_verify_repo_clean(self, run_cmd):
90        """Test deploy_server_local.verify_repo_clean.
91
92        @param run_cmd: Mock of subprocess call used.
93        """
94        # If repo returns what we expect, exit cleanly.
95        run_cmd.return_value = 'nothing to commit (working directory clean)\n'
96        dsl.verify_repo_clean()
97
98        # If repo contains any branches (even clean ones), raise.
99        run_cmd.return_value = self.PROD_STATUS
100        with self.assertRaises(dsl.DirtyTreeException):
101            dsl.verify_repo_clean()
102
103        # If repo doesn't return what we expect, raise.
104        run_cmd.return_value = "That's a very dirty repo you've got."
105        with self.assertRaises(dsl.DirtyTreeException):
106            dsl.verify_repo_clean()
107
108    @mock.patch('subprocess.check_output', autospec=True)
109    def test_repo_versions(self, run_cmd):
110        """Test deploy_server_local.repo_versions.
111
112        @param run_cmd: Mock of subprocess call used.
113        """
114        expected = {
115            'autotest':
116            ('/usr/local/autotest', 'ebb2182'),
117            'autotest/site_utils/autotest_private':
118            ('/usr/local/autotest/site_utils/autotest_private', '78b9626'),
119            'autotest/site_utils/autotest_tools':
120            ('/usr/local/autotest/site_utils/autotest_tools', 'a1598f7'),
121        }
122
123        run_cmd.return_value = self.PROD_VERSIONS
124        result = dsl.repo_versions()
125        self.assertEquals(result, expected)
126
127        run_cmd.assert_called_with(
128                ['repo', 'forall', '-p', '-c',
129                 'pwd && git log -1 --format=%h'])
130
131    @mock.patch('subprocess.check_output', autospec=True)
132    def test_repo_sync_not_for_push_servers(self, run_cmd):
133        """Test deploy_server_local.repo_sync.
134
135        @param run_cmd: Mock of subprocess call used.
136        """
137        dsl.repo_sync()
138        expect_cmds = [mock.call(['git', 'checkout', 'cros/prod'], stderr=-2)]
139        run_cmd.assert_has_calls(expect_cmds)
140
141    @mock.patch('subprocess.check_output', autospec=True)
142    def test_repo_sync_for_push_servers(self, run_cmd):
143        """Test deploy_server_local.repo_sync.
144
145        @param run_cmd: Mock of subprocess call used.
146        """
147        dsl.repo_sync(update_push_servers=True)
148        expect_cmds = [mock.call(['git', 'checkout', 'cros/master'], stderr=-2)]
149        run_cmd.assert_has_calls(expect_cmds)
150
151    def test_discover_commands(self):
152        """Test deploy_server_local.discover_update_commands and
153        discover_restart_services."""
154        # It should always be a list, and should always be callable in
155        # any local environment, though the result will vary.
156        result = dsl.discover_update_commands()
157        self.assertIsInstance(result, list)
158
159    @mock.patch('subprocess.check_output', autospec=True)
160    def test_update_command(self, run_cmd):
161        """Test deploy_server_local.update_command.
162
163        @param run_cmd: Mock of subprocess call used.
164        """
165        # Call with a bad command name.
166        with self.assertRaises(dsl.UnknownCommandException):
167            dsl.update_command('Unknown Command')
168        self.assertFalse(run_cmd.called)
169
170        # Call with a valid command name.
171        dsl.update_command('apache')
172        run_cmd.assert_called_with('sudo service apache2 reload', shell=True,
173                                   stderr=subprocess.STDOUT)
174
175        # Call with a valid command name that uses AUTOTEST_REPO expansion.
176        dsl.update_command('build_externals')
177        expanded_cmd = dsl.common.autotest_dir+'/utils/build_externals.py'
178        run_cmd.assert_called_with(expanded_cmd, shell=True,
179                                   stderr=subprocess.STDOUT)
180
181        # Test a failed command.
182        failure = subprocess.CalledProcessError(10, expanded_cmd, 'output')
183
184        run_cmd.side_effect = failure
185        with self.assertRaises(subprocess.CalledProcessError) as unstable:
186            dsl.update_command('build_externals')
187
188    @mock.patch('subprocess.check_call', autospec=True)
189    def test_restart_service(self, run_cmd):
190        """Test deploy_server_local.restart_service.
191
192        @param run_cmd: Mock of subprocess call used.
193        """
194        # Standard call.
195        dsl.restart_service('foobar')
196        run_cmd.assert_called_with(['sudo', 'service', 'foobar', 'restart'],
197                                   stderr=-2)
198
199    @mock.patch('subprocess.check_output', autospec=True)
200    def test_restart_status(self, run_cmd):
201        """Test deploy_server_local.service_status.
202
203        @param run_cmd: Mock of subprocess call used.
204        """
205        # Standard call.
206        dsl.service_status('foobar')
207        run_cmd.assert_called_with(['sudo', 'service', 'foobar', 'status'])
208
209    @mock.patch.object(dsl, 'restart_service', autospec=True)
210    def _test_restart_services(self, service_results, _restart):
211        """Helper for testing restart_services.
212
213        @param service_results: {'service_name': ['status_1', 'status_2']}
214        """
215        # each call to service_status should return the next status value for
216        # that service.
217        with mock.patch.object(dsl, 'service_status', autospec=True,
218                               side_effect=lambda n: service_results[n].pop(0)):
219            dsl.restart_services(service_results.keys())
220
221    def test_restart_services(self):
222        """Test deploy_server_local.restart_services."""
223        single_stable = {'foo': ['status_ok', 'status_ok']}
224        double_stable = {'foo': ['status_a', 'status_a'],
225                         'bar': ['status_b', 'status_b']}
226
227        # Verify we can handle stable services.
228        self._test_restart_services(single_stable)
229        self._test_restart_services(double_stable)
230
231        single_unstable = {'foo': ['status_ok', 'status_not_ok']}
232        triple_unstable = {'foo': ['status_a', 'status_a'],
233                           'bar': ['status_b', 'status_b_not_ok'],
234                           'joe': ['status_c', 'status_c_not_ok']}
235
236        # Verify we can handle unstable services and report the right failures.
237        with self.assertRaises(dsl.UnstableServices) as unstable:
238            self._test_restart_services(single_unstable)
239        self.assertEqual(unstable.exception.args[0], ['foo'])
240
241        with self.assertRaises(dsl.UnstableServices) as unstable:
242            self._test_restart_services(triple_unstable)
243        self.assertEqual(unstable.exception.args[0], ['bar', 'joe'])
244
245    @mock.patch('subprocess.check_output', autospec=True)
246    def test_report_changes_no_update(self, run_cmd):
247        """Test deploy_server_local.report_changes.
248
249        @param run_cmd: Mock of subprocess call used.
250        """
251
252        before = {
253            'autotest': ('/usr/local/autotest', 'auto_before'),
254            'autotest_private': ('/dir/autotest_private', '78b9626'),
255            'other': ('/fake/unchanged', 'constant_hash'),
256        }
257
258        run_cmd.return_value = 'hash1 Fix change.\nhash2 Bad change.\n'
259
260        result = dsl.report_changes(before, None)
261
262        self.assertEqual(result, """autotest: auto_before
263autotest_private: 78b9626
264other: constant_hash
265""")
266
267        self.assertFalse(run_cmd.called)
268
269    @mock.patch('subprocess.check_output', autospec=True)
270    def test_report_changes_diff(self, run_cmd):
271        """Test deploy_server_local.report_changes.
272
273        @param run_cmd: Mock of subprocess call used.
274        """
275
276        before = {
277            'autotest': ('/usr/local/autotest', 'auto_before'),
278            'autotest_private': ('/dir/autotest_private', '78b9626'),
279            'other': ('/fake/unchanged', 'constant_hash'),
280        }
281
282        after = {
283            'autotest': ('/usr/local/autotest', 'auto_after'),
284            'autotest_tools': ('/dir/autotest_tools', 'a1598f7'),
285            'other': ('/fake/unchanged', 'constant_hash'),
286        }
287
288        run_cmd.return_value = 'hash1 Fix change.\nhash2 Bad change.\n'
289
290        result = dsl.report_changes(before, after)
291
292        self.assertEqual(result, """autotest:
293hash1 Fix change.
294hash2 Bad change.
295
296autotest_private:
297Removed.
298
299autotest_tools:
300Added.
301
302other:
303No Change.
304""")
305
306        run_cmd.assert_called_with(
307                ['git', 'log', 'auto_before..auto_after', '--oneline'],
308                cwd='/usr/local/autotest', stderr=subprocess.STDOUT)
309
310    def test_parse_arguments(self):
311        """Test deploy_server_local.parse_arguments."""
312        # No arguments.
313        results = dsl.parse_arguments([])
314        self.assertDictContainsSubset(
315                {'verify': True, 'update': True, 'actions': True,
316                 'report': True, 'dryrun': False, 'update_push_servers': False},
317                vars(results))
318
319        # Update test_push servers.
320        results = dsl.parse_arguments(['--update_push_servers'])
321        self.assertDictContainsSubset(
322                {'verify': True, 'update': True, 'actions': True,
323                 'report': True, 'dryrun': False, 'update_push_servers': True},
324                vars(results))
325
326        # Dryrun.
327        results = dsl.parse_arguments(['--dryrun'])
328        self.assertDictContainsSubset(
329                {'verify': False, 'update': False, 'actions': True,
330                 'report': True, 'dryrun': True, 'update_push_servers': False},
331                vars(results))
332
333        # Restart only.
334        results = dsl.parse_arguments(['--actions-only'])
335        self.assertDictContainsSubset(
336                {'verify': False, 'update': False, 'actions': True,
337                 'report': False, 'dryrun': False,
338                 'update_push_servers': False},
339                vars(results))
340
341        # All skip arguments.
342        results = dsl.parse_arguments(['--skip-verify', '--skip-update',
343                                       '--skip-actions', '--skip-report'])
344        self.assertDictContainsSubset(
345                {'verify': False, 'update': False, 'actions': False,
346                 'report': False, 'dryrun': False,
347                 'update_push_servers': False},
348                vars(results))
349
350        # All arguments.
351        results = dsl.parse_arguments(['--skip-verify', '--skip-update',
352                                       '--skip-actions', '--skip-report',
353                                       '--actions-only', '--dryrun',
354                                       '--update_push_servers'])
355        self.assertDictContainsSubset(
356                {'verify': False, 'update': False, 'actions': False,
357                 'report': False, 'dryrun': True, 'update_push_servers': True},
358                vars(results))
359
360
361if __name__ == '__main__':
362    unittest.main()
363