1 #!/usr/bin/env python 2 # 3 # Copyright 2016 - The Android Open Source Project 4 # 5 # Licensed under the Apache License, Version 2.0 (the "License"); 6 # you may not use this file except in compliance with the License. 7 # You may obtain a copy of the License at 8 # 9 # http://www.apache.org/licenses/LICENSE-2.0 10 # 11 # Unless required by applicable law or agreed to in writing, software 12 # distributed under the License is distributed on an "AS IS" BASIS, 13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14 # See the License for the specific language governing permissions and 15 # limitations under the License. 16 """Tests acloud.public.acloud_kernel.kernel_swapper.""" 17 18 import subprocess 19 import mock 20 21 import unittest 22 from acloud.internal.lib import android_compute_client 23 from acloud.internal.lib import auth 24 from acloud.internal.lib import driver_test_lib 25 from acloud.public.acloud_kernel import kernel_swapper 26 27 28 class KernelSwapperTest(driver_test_lib.BaseDriverTest): 29 """Test kernel_swapper.""" 30 31 def setUp(self): 32 """Set up the test.""" 33 super(KernelSwapperTest, self).setUp() 34 self.cfg = mock.MagicMock() 35 self.credentials = mock.MagicMock() 36 self.Patch(auth, 'CreateCredentials', return_value=self.credentials) 37 self.compute_client = mock.MagicMock() 38 self.Patch( 39 android_compute_client, 40 'AndroidComputeClient', 41 return_value=self.compute_client) 42 self.subprocess_call = self.Patch(subprocess, 'check_call') 43 44 self.fake_ip = '123.456.789.000' 45 self.fake_instance = 'fake-instance' 46 self.compute_client.GetInstanceIP.return_value = self.fake_ip 47 48 self.kswapper = kernel_swapper.KernelSwapper(self.cfg, 49 self.fake_instance) 50 self.ssh_cmd_prefix = 'ssh %s root@%s' % ( 51 ' '.join(kernel_swapper.SSH_FLAGS), self.fake_ip) 52 self.scp_cmd_prefix = 'scp %s' % ' '.join(kernel_swapper.SSH_FLAGS) 53 54 def testPushFile(self): 55 """Test RebootTarget.""" 56 fake_src_path = 'fake-src' 57 fake_dest_path = 'fake-dest' 58 scp_cmd = ' '.join([self.scp_cmd_prefix, '%s root@%s:%s' % 59 (fake_src_path, self.fake_ip, fake_dest_path)]) 60 61 self.kswapper.PushFile(fake_src_path, fake_dest_path) 62 self.subprocess_call.assert_called_once_with(scp_cmd, shell=True) 63 64 def testRebootTarget(self): 65 """Test RebootTarget.""" 66 self.kswapper.RebootTarget() 67 reboot_cmd = ' '.join([ 68 self.ssh_cmd_prefix, '"%s"' % kernel_swapper.REBOOT_CMD 69 ]) 70 71 self.subprocess_call.assert_called_once_with(reboot_cmd, shell=True) 72 self.compute_client.WaitForBoot.assert_called_once_with( 73 self.fake_instance) 74 75 def testSwapKernel(self): 76 """Test SwapKernel.""" 77 fake_local_kernel_image = 'fake-kernel' 78 mount_cmd = ' '.join([ 79 self.ssh_cmd_prefix, '"%s"' % kernel_swapper.MOUNT_CMD 80 ]) 81 scp_cmd = ' '.join([self.scp_cmd_prefix, '%s root@%s:%s' % 82 (fake_local_kernel_image, self.fake_ip, '/boot')]) 83 reboot_cmd = ' '.join([ 84 self.ssh_cmd_prefix, '"%s"' % kernel_swapper.REBOOT_CMD 85 ]) 86 87 self.kswapper.SwapKernel(fake_local_kernel_image) 88 self.subprocess_call.assert_has_calls([ 89 mock.call( 90 mount_cmd, shell=True), mock.call( 91 scp_cmd, shell=True), mock.call( 92 reboot_cmd, shell=True) 93 ]) 94 95 96 if __name__ == '__main__': 97 unittest.main() 98
