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 deploy_server_local as dsl 15 16 17 class 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 27 ebb2182 28 29 \x1b[1mproject autotest/site_utils/autotest_private/\x1b[m 30 /usr/local/autotest/site_utils/autotest_private 31 78b9626 32 33 \x1b[1mproject autotest/site_utils/autotest_tools/\x1b[m 34 /usr/local/autotest/site_utils/autotest_tools 35 a1598f7 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 77 ebb2182 78 79 project autotest/site_utils/autotest_private/ 80 /usr/local/autotest/site_utils/autotest_private 81 78b9626 82 83 project autotest/site_utils/autotest_tools/ 84 /usr/local/autotest/site_utils/autotest_tools 85 a1598f7 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 263 autotest_private: 78b9626 264 other: 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: 293 hash1 Fix change. 294 hash2 Bad change. 295 296 autotest_private: 297 Removed. 298 299 autotest_tools: 300 Added. 301 302 other: 303 No 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 361 if __name__ == '__main__': 362 unittest.main() 363
