1# Copyright 2019 - 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_cf_device_factory.""" 15 16import glob 17import os 18import unittest 19import uuid 20 21from unittest import mock 22 23from acloud.create import avd_spec 24from acloud.internal import constants 25from acloud.internal.lib import android_build_client 26from acloud.internal.lib import auth 27from acloud.internal.lib import cvd_compute_client_multi_stage 28from acloud.internal.lib import driver_test_lib 29from acloud.internal.lib import utils 30from acloud.list import list as list_instances 31from acloud.public.actions import remote_instance_cf_device_factory 32 33 34class RemoteInstanceDeviceFactoryTest(driver_test_lib.BaseDriverTest): 35 """Test RemoteInstanceDeviceFactory method.""" 36 37 def setUp(self): 38 """Set up the test.""" 39 super().setUp() 40 self.Patch(auth, "CreateCredentials", return_value=mock.MagicMock()) 41 self.Patch(android_build_client.AndroidBuildClient, "InitResourceHandle") 42 self.Patch(cvd_compute_client_multi_stage.CvdComputeClient, "InitResourceHandle") 43 self.Patch(cvd_compute_client_multi_stage.CvdComputeClient, "LaunchCvd") 44 self.Patch(list_instances, "GetInstancesFromInstanceNames", return_value=mock.MagicMock()) 45 self.Patch(list_instances, "ChooseOneRemoteInstance", return_value=mock.MagicMock()) 46 self.Patch(utils, "GetBuildEnvironmentVariable", 47 return_value="test_env_cf_arm") 48 self.Patch(glob, "glob", return_vale=["fake.img"]) 49 50 # pylint: disable=protected-access 51 @mock.patch.object(cvd_compute_client_multi_stage.CvdComputeClient, 52 "UpdateCertificate") 53 @mock.patch.object(remote_instance_cf_device_factory.RemoteInstanceDeviceFactory, 54 "_FetchBuild") 55 @mock.patch("acloud.public.actions.remote_instance_cf_device_factory." 56 "cvd_utils") 57 def testProcessArtifacts(self, mock_cvd_utils, mock_download, 58 mock_uploadca): 59 """test ProcessArtifacts.""" 60 # Test image source type is local. 61 args = mock.MagicMock() 62 args.config_file = "" 63 args.avd_type = constants.TYPE_CF 64 args.flavor = "phone" 65 args.local_image = constants.FIND_IN_BUILD_ENV 66 args.local_system_image = None 67 args.launch_args = None 68 args.autoconnect = constants.INS_KEY_WEBRTC 69 avd_spec_local_img = avd_spec.AVDSpec(args) 70 fake_image_name = "/fake/aosp_cf_x86_phone-img-eng.username.zip" 71 fake_host_package_name = "/fake/host_package.tar.gz" 72 factory_local_img = remote_instance_cf_device_factory.RemoteInstanceDeviceFactory( 73 avd_spec_local_img, 74 fake_image_name, 75 fake_host_package_name) 76 factory_local_img._ProcessArtifacts() 77 # cf default autoconnect webrtc and should upload certificates 78 mock_uploadca.assert_called_once() 79 mock_uploadca.reset_mock() 80 mock_cvd_utils.UploadArtifacts.assert_called_once_with( 81 mock.ANY, fake_image_name, fake_host_package_name) 82 mock_cvd_utils.UploadExtraImages.assert_called_once() 83 84 # given autoconnect to vnc should not upload certificates 85 args.autoconnect = constants.INS_KEY_VNC 86 avd_spec_local_img = avd_spec.AVDSpec(args) 87 factory_local_img = remote_instance_cf_device_factory.RemoteInstanceDeviceFactory( 88 avd_spec_local_img, 89 fake_image_name, 90 fake_host_package_name) 91 factory_local_img._ProcessArtifacts() 92 mock_uploadca.assert_not_called() 93 94 # Test image source type is remote. 95 args.local_image = None 96 args.build_id = "1234" 97 args.branch = "fake_branch" 98 args.build_target = "fake_target" 99 args.system_build_id = "2345" 100 args.system_branch = "sys_branch" 101 args.system_build_target = "sys_target" 102 args.kernel_build_id = "3456" 103 args.kernel_branch = "kernel_branch" 104 args.kernel_build_target = "kernel_target" 105 avd_spec_remote_img = avd_spec.AVDSpec(args) 106 self.Patch(cvd_compute_client_multi_stage.CvdComputeClient, "UpdateFetchCvd") 107 factory_remote_img = remote_instance_cf_device_factory.RemoteInstanceDeviceFactory( 108 avd_spec_remote_img) 109 factory_remote_img._ProcessArtifacts() 110 mock_download.assert_called_once() 111 112 # pylint: disable=protected-access 113 @mock.patch.dict(os.environ, {constants.ENV_BUILD_TARGET:'fake-target'}) 114 def testCreateGceInstanceNameMultiStage(self): 115 """test create gce instance.""" 116 # Mock uuid 117 args = mock.MagicMock() 118 args.config_file = "" 119 args.avd_type = constants.TYPE_CF 120 args.flavor = "phone" 121 args.local_image = constants.FIND_IN_BUILD_ENV 122 args.local_system_image = None 123 args.adb_port = None 124 args.launch_args = None 125 fake_avd_spec = avd_spec.AVDSpec(args) 126 fake_avd_spec.cfg.enable_multi_stage = True 127 fake_avd_spec._instance_name_to_reuse = None 128 129 fake_uuid = mock.MagicMock(hex="1234") 130 self.Patch(uuid, "uuid4", return_value=fake_uuid) 131 self.Patch(cvd_compute_client_multi_stage.CvdComputeClient, "CreateInstance") 132 self.Patch(cvd_compute_client_multi_stage.CvdComputeClient, 133 "GetHostImageName", return_value="fake_image") 134 fake_host_package_name = "/fake/host_package.tar.gz" 135 fake_image_name = "/fake/aosp_cf_x86_phone-img-eng.username.zip" 136 137 factory = remote_instance_cf_device_factory.RemoteInstanceDeviceFactory( 138 fake_avd_spec, 139 fake_image_name, 140 fake_host_package_name) 141 self.assertEqual(factory._CreateGceInstance(), "ins-1234-userbuild-aosp-cf-x86-phone") 142 143 # Can't get target name from zip file name. 144 fake_image_name = "/fake/aosp_cf_x86_phone.username.zip" 145 factory = remote_instance_cf_device_factory.RemoteInstanceDeviceFactory( 146 fake_avd_spec, 147 fake_image_name, 148 fake_host_package_name) 149 self.assertEqual(factory._CreateGceInstance(), "ins-1234-userbuild-fake-target") 150 151 # No image zip path, it uses local build images. 152 fake_image_name = "" 153 factory = remote_instance_cf_device_factory.RemoteInstanceDeviceFactory( 154 fake_avd_spec, 155 fake_image_name, 156 fake_host_package_name) 157 self.assertEqual(factory._CreateGceInstance(), "ins-1234-userbuild-fake-target") 158 159 def testReuseInstanceNameMultiStage(self): 160 """Test reuse instance name.""" 161 args = mock.MagicMock() 162 args.config_file = "" 163 args.avd_type = constants.TYPE_CF 164 args.flavor = "phone" 165 args.local_image = constants.FIND_IN_BUILD_ENV 166 args.local_system_image = None 167 args.adb_port = None 168 args.launch_args = None 169 fake_avd_spec = avd_spec.AVDSpec(args) 170 fake_avd_spec.cfg.enable_multi_stage = True 171 fake_avd_spec._instance_name_to_reuse = "fake-1234-userbuild-fake-target" 172 fake_uuid = mock.MagicMock(hex="1234") 173 self.Patch(uuid, "uuid4", return_value=fake_uuid) 174 self.Patch(cvd_compute_client_multi_stage.CvdComputeClient, "CreateInstance") 175 self.Patch(cvd_compute_client_multi_stage.CvdComputeClient, 176 "GetHostImageName", return_value="fake_image") 177 fake_host_package_name = "/fake/host_package.tar.gz" 178 fake_image_name = "/fake/aosp_cf_x86_phone-img-eng.username.zip" 179 factory = remote_instance_cf_device_factory.RemoteInstanceDeviceFactory( 180 fake_avd_spec, 181 fake_image_name, 182 fake_host_package_name) 183 self.assertEqual(factory._CreateGceInstance(), "fake-1234-userbuild-fake-target") 184 185 @mock.patch("acloud.public.actions.remote_instance_cf_device_factory." 186 "cvd_utils") 187 def testGetBuildInfoDict(self, mock_cvd_utils): 188 """Test GetBuildInfoDict.""" 189 fake_host_package_name = "/fake/host_package.tar.gz" 190 fake_image_name = "/fake/aosp_cf_x86_phone-img-eng.username.zip" 191 args = mock.MagicMock() 192 # Test image source type is local. 193 args.config_file = "" 194 args.avd_type = constants.TYPE_CF 195 args.flavor = "phone" 196 args.local_image = "fake_local_image" 197 args.local_system_image = None 198 args.adb_port = None 199 args.cheeps_betty_image = None 200 args.launch_args = None 201 avd_spec_local_image = avd_spec.AVDSpec(args) 202 factory = remote_instance_cf_device_factory.RemoteInstanceDeviceFactory( 203 avd_spec_local_image, 204 fake_image_name, 205 fake_host_package_name) 206 self.assertEqual(factory.GetBuildInfoDict(), None) 207 mock_cvd_utils.assert_not_called() 208 209 # Test image source type is remote. 210 args.local_image = None 211 args.build_id = "123" 212 args.branch = "fake_branch" 213 args.build_target = "fake_target" 214 avd_spec_remote_image = avd_spec.AVDSpec(args) 215 factory = remote_instance_cf_device_factory.RemoteInstanceDeviceFactory( 216 avd_spec_remote_image, 217 fake_image_name, 218 fake_host_package_name) 219 factory.GetBuildInfoDict() 220 mock_cvd_utils.GetRemoteBuildInfoDict.assert_called() 221 222 @mock.patch.object(remote_instance_cf_device_factory.RemoteInstanceDeviceFactory, 223 "_CreateGceInstance") 224 @mock.patch("acloud.public.actions.remote_instance_cf_device_factory.pull") 225 @mock.patch("acloud.public.actions.remote_instance_cf_device_factory." 226 "cvd_utils") 227 def testLocalImageCreateInstance(self, mock_cvd_utils, mock_pull, 228 mock_create_gce_instance): 229 """Test CreateInstance with local images.""" 230 self.Patch( 231 cvd_compute_client_multi_stage, 232 "CvdComputeClient", 233 return_value=mock.MagicMock()) 234 mock_create_gce_instance.return_value = "instance" 235 fake_avd_spec = mock.MagicMock() 236 fake_avd_spec.image_source = constants.IMAGE_SRC_LOCAL 237 fake_avd_spec._instance_name_to_reuse = None 238 fake_avd_spec.no_pull_log = False 239 240 mock_cvd_utils.ConvertRemoteLogs.return_value = [{"path": "/logcat"}] 241 mock_cvd_utils.UploadExtraImages.return_value = [ 242 "-boot_image", "/boot/img"] 243 244 fake_host_package_name = "/fake/host_package.tar.gz" 245 fake_image_name = "" 246 factory = remote_instance_cf_device_factory.RemoteInstanceDeviceFactory( 247 fake_avd_spec, 248 fake_image_name, 249 fake_host_package_name) 250 compute_client = factory.GetComputeClient() 251 compute_client.LaunchCvd.return_value = {"instance": "failure"} 252 factory.CreateInstance() 253 mock_create_gce_instance.assert_called_once() 254 mock_cvd_utils.UploadArtifacts.assert_called_once() 255 compute_client.LaunchCvd.assert_called_once() 256 self.assertEqual( 257 ["-boot_image", "/boot/img"], 258 compute_client.LaunchCvd.call_args[1].get("extra_args")) 259 mock_pull.GetAllLogFilePaths.assert_called_once() 260 mock_pull.PullLogs.assert_called_once() 261 self.assertEqual({"instance": "failure"}, factory.GetFailures()) 262 self.assertEqual(3, len(factory.GetLogs().get("instance"))) 263 264 @mock.patch.object(remote_instance_cf_device_factory.RemoteInstanceDeviceFactory, 265 "_CreateGceInstance") 266 @mock.patch("acloud.public.actions.remote_instance_cf_device_factory.pull") 267 @mock.patch("acloud.public.actions.remote_instance_cf_device_factory." 268 "cvd_utils") 269 def testRemoteImageCreateInstance(self, mock_cvd_utils, mock_pull, 270 mock_create_gce_instance): 271 """Test CreateInstance with remote images.""" 272 self.Patch( 273 cvd_compute_client_multi_stage, 274 "CvdComputeClient", 275 return_value=mock.MagicMock()) 276 mock_create_gce_instance.return_value = "instance" 277 fake_avd_spec = mock.MagicMock() 278 fake_avd_spec.image_source = constants.IMAGE_SRC_REMOTE 279 fake_avd_spec.host_user = None 280 fake_avd_spec.no_pull_log = True 281 282 mock_cvd_utils.ConvertRemoteLogs.return_value = [{"path": "/logcat"}] 283 mock_cvd_utils.UploadExtraImages.return_value = [] 284 285 factory = remote_instance_cf_device_factory.RemoteInstanceDeviceFactory( 286 fake_avd_spec) 287 compute_client = factory.GetComputeClient() 288 compute_client.LaunchCvd.return_value = {} 289 factory.CreateInstance() 290 291 compute_client.FetchBuild.assert_called_once() 292 mock_pull.GetAllLogFilePaths.assert_called_once() 293 mock_pull.PullLogs.assert_not_called() 294 self.assertFalse(factory.GetFailures()) 295 self.assertEqual(4, len(factory.GetLogs().get("instance"))) 296 297 def testGetOpenWrtInfoDict(self): 298 """Test GetOpenWrtInfoDict.""" 299 self.Patch(cvd_compute_client_multi_stage.CvdComputeClient, 300 "GetSshConnectCmd", return_value="fake_ssh_command") 301 args = mock.MagicMock() 302 args.config_file = "" 303 args.avd_type = constants.TYPE_CF 304 args.flavor = "phone" 305 args.local_image = "fake_local_image" 306 args.launch_args = None 307 args.openwrt = False 308 avd_spec_no_openwrt = avd_spec.AVDSpec(args) 309 factory = remote_instance_cf_device_factory.RemoteInstanceDeviceFactory( 310 avd_spec_no_openwrt) 311 self.assertEqual(None, factory.GetOpenWrtInfoDict()) 312 313 args.openwrt = True 314 avd_spec_openwrt = avd_spec.AVDSpec(args) 315 factory = remote_instance_cf_device_factory.RemoteInstanceDeviceFactory( 316 avd_spec_openwrt) 317 expect_result = {"ssh_command": "fake_ssh_command", 318 "screen_command": "screen ~/cuttlefish_runtime/console"} 319 self.assertEqual(expect_result, factory.GetOpenWrtInfoDict()) 320 321 322if __name__ == "__main__": 323 unittest.main() 324