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
18import subprocess
19import mock
20
21import unittest
22from acloud.internal.lib import android_compute_client
23from acloud.internal.lib import auth
24from acloud.internal.lib import driver_test_lib
25from acloud.public.acloud_kernel import kernel_swapper
26
27
28class 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
96if __name__ == '__main__':
97    unittest.main()
98