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
19
20import unittest
21import mock
22
23from acloud.internal.lib import android_compute_client
24from acloud.internal.lib import auth
25from acloud.internal.lib import driver_test_lib
26from acloud.public.acloud_kernel import kernel_swapper
27
28
29#pylint: disable=too-many-instance-attributes
30class KernelSwapperTest(driver_test_lib.BaseDriverTest):
31    """Test kernel_swapper."""
32
33    def setUp(self):
34        """Set up the test."""
35        super(KernelSwapperTest, self).setUp()
36        self.cfg = mock.MagicMock()
37        self.credentials = mock.MagicMock()
38        self.Patch(auth, 'CreateCredentials', return_value=self.credentials)
39        self.compute_client = mock.MagicMock()
40        self.Patch(
41            android_compute_client,
42            'AndroidComputeClient',
43            return_value=self.compute_client)
44        self.subprocess_call = self.Patch(subprocess, 'check_call')
45
46        self.fake_ip = '123.456.789.000'
47        self.fake_instance = 'fake-instance'
48        self.compute_client.GetInstanceIP.return_value = self.fake_ip
49
50        self.kswapper = kernel_swapper.KernelSwapper(self.cfg,
51                                                     self.fake_instance)
52        self.ssh_cmd_prefix = 'ssh %s root@%s' % (' '.join(
53            kernel_swapper.SSH_FLAGS), self.fake_ip)
54        self.scp_cmd_prefix = 'scp %s' % ' '.join(kernel_swapper.SSH_FLAGS)
55
56    def testPushFile(self):
57        """Test RebootTarget."""
58        fake_src_path = 'fake-src'
59        fake_dest_path = 'fake-dest'
60        scp_cmd = ' '.join([
61            self.scp_cmd_prefix,
62            '%s root@%s:%s' % (fake_src_path, self.fake_ip, fake_dest_path)
63        ])
64
65        self.kswapper.PushFile(fake_src_path, fake_dest_path)
66        self.subprocess_call.assert_called_once_with(scp_cmd, shell=True)
67
68    def testRebootTarget(self):
69        """Test RebootTarget."""
70        self.kswapper.RebootTarget()
71        reboot_cmd = ' '.join(
72            [self.ssh_cmd_prefix,
73             '"%s"' % kernel_swapper.REBOOT_CMD])
74
75        self.subprocess_call.assert_called_once_with(reboot_cmd, shell=True)
76        self.compute_client.WaitForBoot.assert_called_once_with(
77            self.fake_instance)
78
79    def testSwapKernel(self):
80        """Test SwapKernel."""
81        fake_local_kernel_image = 'fake-kernel'
82        mount_cmd = ' '.join(
83            [self.ssh_cmd_prefix,
84             '"%s"' % kernel_swapper.MOUNT_CMD])
85        scp_cmd = ' '.join([
86            self.scp_cmd_prefix,
87            '%s root@%s:%s' % (fake_local_kernel_image, self.fake_ip, '/boot')
88        ])
89        reboot_cmd = ' '.join(
90            [self.ssh_cmd_prefix,
91             '"%s"' % kernel_swapper.REBOOT_CMD])
92
93        self.kswapper.SwapKernel(fake_local_kernel_image)
94        self.subprocess_call.assert_has_calls([
95            mock.call(mount_cmd, shell=True),
96            mock.call(scp_cmd, shell=True),
97            mock.call(reboot_cmd, shell=True)
98        ])
99
100
101if __name__ == '__main__':
102    unittest.main()
103