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