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