• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright 2024 - The Android Open Source Project
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7#     http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14"""Tests for remote_instance_trusty_device_factory."""
15
16import glob
17import logging
18import os
19import tempfile
20import unittest
21import uuid
22
23from unittest import mock
24
25from acloud.create import avd_spec
26from acloud.internal import constants
27from acloud.internal.lib import android_build_client
28from acloud.internal.lib import auth
29from acloud.internal.lib import cvd_compute_client_multi_stage
30from acloud.internal.lib import driver_test_lib
31from acloud.list import list as list_instances
32from acloud.public.actions import remote_instance_trusty_device_factory
33
34logger = logging.getLogger(__name__)
35
36_EXPECTED_CONFIG_JSON = """{"linux": "kernel", "linux_arch": "arm64", \
37"initrd": "ramdisk.img", "atf": "atf/qemu/debug", \
38"qemu": "bin/trusty_qemu_system_aarch64", \
39"extra_qemu_flags": ["-machine", "gic-version=3"], "android_image_dir": ".", \
40"rpmbd": "bin/rpmb_dev", "arch": "arm64", "adb": "bin/adb"}"""
41
42
43class RemoteInstanceDeviceFactoryTest(driver_test_lib.BaseDriverTest):
44    """Test RemoteInstanceDeviceFactory."""
45
46    def setUp(self):
47        super().setUp()
48        self.Patch(auth, "CreateCredentials", return_value=mock.MagicMock())
49        self.Patch(android_build_client.AndroidBuildClient, "InitResourceHandle")
50        self.Patch(android_build_client.AndroidBuildClient, "DownloadArtifact")
51        self.Patch(cvd_compute_client_multi_stage.CvdComputeClient, "InitResourceHandle")
52        self.Patch(list_instances, "GetInstancesFromInstanceNames", return_value=mock.MagicMock())
53        self.Patch(list_instances, "ChooseOneRemoteInstance", return_value=mock.MagicMock())
54        self.Patch(glob, "glob", return_value=["fake.img"])
55
56    # pylint: disable=protected-access
57    @mock.patch("acloud.public.actions.remote_instance_trusty_device_factory."
58                "cvd_utils")
59    def testLocalImage(self, mock_cvd_utils):
60        """test ProcessArtifacts with local image."""
61        fake_emulator_package = "/fake/trusty_build/trusty_image_package.tar.gz"
62        fake_image_name = "/fake/qemu_trusty_arm64-img-eng.username.zip"
63        fake_host_package_name = "/fake/trusty_host_package.tar.gz"
64        fake_tmp_path = "/fake/tmp_file"
65
66        args = mock.MagicMock()
67        args.config_file = ""
68        args.avd_type = constants.TYPE_TRUSTY
69        args.flavor = "phone"
70        args.local_image = constants.FIND_IN_BUILD_ENV
71        args.launch_args = None
72        args.autoconnect = constants.INS_KEY_WEBRTC
73        args.local_trusty_image = fake_emulator_package
74        args.trusty_host_package = fake_host_package_name
75        args.reuse_gce = None
76        mock_cvd_utils.GCE_BASE_DIR = "gce_base_dir"
77
78        # Test local images
79        avd_spec_local_img = avd_spec.AVDSpec(args)
80
81        self.Patch(os.path, "exists", return_value=True)
82        factory_local_img = remote_instance_trusty_device_factory.RemoteInstanceDeviceFactory(
83            avd_spec_local_img,
84            fake_image_name)
85        mock_ssh = mock.Mock()
86        factory_local_img._ssh = mock_ssh
87
88        temp_config = ""
89        def WriteTempConfig(s):
90            nonlocal temp_config
91            temp_config += s
92        temp_config_mock = mock.MagicMock()
93        temp_config_mock.__enter__().name = fake_tmp_path
94        temp_config_mock.__enter__().write.side_effect = WriteTempConfig
95        self.Patch(tempfile, "NamedTemporaryFile", return_value=temp_config_mock)
96
97        factory_local_img._ProcessArtifacts()
98
99        mock_cvd_utils.UploadArtifacts.assert_called_once_with(
100            mock.ANY, mock_cvd_utils.GCE_BASE_DIR, fake_image_name,
101            fake_host_package_name)
102        mock_ssh.Run.assert_called_once_with(
103            f"tar -xzf - -C {mock_cvd_utils.GCE_BASE_DIR} "
104            f"< {fake_emulator_package}")
105        self.assertEqual(temp_config, _EXPECTED_CONFIG_JSON)
106        mock_ssh.ScpPushFile.assert_called_with(
107            fake_tmp_path, f"{mock_cvd_utils.GCE_BASE_DIR}/config.json")
108
109    # pylint: disable=protected-access
110    @mock.patch("acloud.public.actions.remote_instance_trusty_device_factory."
111                "cvd_utils")
112    def testRemoteImage(self, mock_cvd_utils):
113        """test ProcessArtifacts with remote image source."""
114        fake_tmp_path = "/fake/tmp_file"
115
116        args = mock.MagicMock()
117        args.config_file = ""
118        args.avd_type = constants.TYPE_TRUSTY
119        args.flavor = "phone"
120        args.local_image = None
121        args.launch_args = None
122        args.autoconnect = constants.INS_KEY_WEBRTC
123        args.local_trusty_image = None
124        args.reuse_gce = None
125        args.build_id = "default_build_id"
126        args.branch = "default_branch"
127        args.build_target = "default_target"
128        args.kernel_build_id = "kernel_build_id"
129        args.kernel_build_target = "kernel_target"
130        args.host_package_build_id = None
131        args.host_package_branch = None
132        args.host_package_build_target = None
133        mock_cvd_utils.GCE_BASE_DIR = "gce_base_dir"
134
135        avd_spec_remote_img = avd_spec.AVDSpec(args)
136        factory_remote_img = remote_instance_trusty_device_factory.RemoteInstanceDeviceFactory(
137            avd_spec_remote_img)
138        mock_ssh = mock.Mock()
139        factory_remote_img._ssh = mock_ssh
140
141        temp_file_mock = mock.MagicMock()
142        temp_file_mock.__enter__().name = fake_tmp_path
143        self.Patch(tempfile, "NamedTemporaryFile", return_value=temp_file_mock)
144
145        factory_remote_img._ProcessArtifacts()
146
147        # Download trusty image package
148        factory_remote_img.GetComputeClient().build_api.DownloadArtifact.called_once()
149
150        mock_ssh.Run.assert_has_calls(
151            [
152                mock.call(
153                    "cvd fetch -credential_source=gce "
154                    "-default_build=default_build_id/default_target "
155                    "-kernel_build=kernel_build_id/kernel_target "
156                    "-host_package_build=default_build_id/default_target{trusty-host_package.tar.gz}",
157                    timeout=300,
158                ),
159                mock.call(
160                    f"cd {mock_cvd_utils.GCE_BASE_DIR}/bin && "
161                    "./replace_ramdisk_modules "
162                    f"--android-ramdisk={mock_cvd_utils.GCE_BASE_DIR}/ramdisk.img "
163                    f"--kernel-ramdisk={mock_cvd_utils.GCE_BASE_DIR}/initramfs.img "
164                    f"--output-ramdisk={mock_cvd_utils.GCE_BASE_DIR}/ramdisk.img",
165                    timeout=300,
166                ),
167                mock.call(
168                    f"tar -xzf - -C {mock_cvd_utils.GCE_BASE_DIR} "
169                    f"< {fake_tmp_path}"
170                ),
171            ]
172        )
173
174    @mock.patch.object(remote_instance_trusty_device_factory.RemoteInstanceDeviceFactory,
175                       "CreateGceInstance")
176    @mock.patch("acloud.public.actions.remote_instance_trusty_device_factory."
177                "cvd_utils")
178    def testLocalImageCreateInstance(self, mock_cvd_utils, mock_create_gce_instance):
179        """Test CreateInstance with local images."""
180        self.Patch(
181            cvd_compute_client_multi_stage,
182            "CvdComputeClient",
183            return_value=mock.MagicMock())
184        mock_cvd_utils.GCE_BASE_DIR = "gce_base_dir"
185        mock_create_gce_instance.return_value = "instance"
186        fake_avd_spec = mock.MagicMock()
187        fake_avd_spec.image_source = constants.IMAGE_SRC_LOCAL
188        fake_avd_spec._instance_name_to_reuse = None
189        fake_avd_spec.no_pull_log = False
190        fake_avd_spec.base_instance_num = None
191        fake_avd_spec.num_avds_per_instance = None
192
193        mock_cvd_utils.HOST_KERNEL_LOG = {"path": "/host_kernel.log"}
194
195        fake_image_name = ""
196        factory = remote_instance_trusty_device_factory.RemoteInstanceDeviceFactory(
197            fake_avd_spec,
198            fake_image_name)
199        mock_ssh = mock.Mock()
200        factory._ssh = mock_ssh
201        factory.CreateInstance()
202        mock_create_gce_instance.assert_called_once()
203        mock_cvd_utils.UploadArtifacts.assert_called_once()
204        # First call is unpacking host package
205        # then unpacking image archive
206        # and finally run
207        self.assertEqual(mock_ssh.Run.call_count, 3)
208        self.assertIn(
209            "gce_base_dir/run.py --verbose --config=config.json",
210            mock_ssh.Run.call_args[0][0],
211        )
212
213        self.assertEqual(3, len(factory.GetLogs().get("instance")))
214
215
216if __name__ == "__main__":
217    unittest.main()
218