1#!/usr/bin/env python 2# 3# Copyright 2018 - 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 for LocalImageLocalInstance.""" 17 18import builtins 19import os 20import subprocess 21import tempfile 22import unittest 23 24from unittest import mock 25 26from acloud import errors 27from acloud.create import local_image_local_instance 28from acloud.list import instance 29from acloud.list import list as list_instance 30from acloud.internal import constants 31from acloud.internal.lib import driver_test_lib 32from acloud.internal.lib import utils 33from acloud.public import report 34 35 36class LocalImageLocalInstanceTest(driver_test_lib.BaseDriverTest): 37 """Test LocalImageLocalInstance method.""" 38 39 LAUNCH_CVD_CMD_WITH_DISK = """sg group1 <<EOF 40sg group2 41bin/cvd start -daemon -config=phone -system_image_dir fake_image_dir -instance_dir fake_cvd_dir -undefok=report_anonymous_usage_stats,config -report_anonymous_usage_stats=y -cpus fake -x_res fake -y_res fake -dpi fake -memory_mb fake -blank_data_image_mb fake -data_policy always_create -start_vnc_server=true 42EOF""" 43 44 LAUNCH_CVD_CMD_NO_DISK = """sg group1 <<EOF 45sg group2 46bin/cvd start -daemon -config=phone -system_image_dir fake_image_dir -instance_dir fake_cvd_dir -undefok=report_anonymous_usage_stats,config -report_anonymous_usage_stats=y -cpus fake -x_res fake -y_res fake -dpi fake -memory_mb fake -start_vnc_server=true 47EOF""" 48 49 LAUNCH_CVD_CMD_NO_DISK_WITH_GPU = """sg group1 <<EOF 50sg group2 51bin/cvd start -daemon -config=phone -system_image_dir fake_image_dir -instance_dir fake_cvd_dir -undefok=report_anonymous_usage_stats,config -report_anonymous_usage_stats=y -cpus fake -x_res fake -y_res fake -dpi fake -memory_mb fake -start_vnc_server=true 52EOF""" 53 54 LAUNCH_CVD_CMD_WITH_WEBRTC = """sg group1 <<EOF 55sg group2 56bin/cvd start -daemon -config=auto -system_image_dir fake_image_dir -instance_dir fake_cvd_dir -undefok=report_anonymous_usage_stats,config -report_anonymous_usage_stats=y -start_webrtc=true 57EOF""" 58 59 LAUNCH_CVD_CMD_WITH_MIXED_IMAGES = """sg group1 <<EOF 60sg group2 61bin/cvd start -daemon -config=phone -system_image_dir fake_image_dir -instance_dir fake_cvd_dir -undefok=report_anonymous_usage_stats,config -report_anonymous_usage_stats=y -start_vnc_server=true -super_image=fake_super_image -boot_image=fake_boot_image -vendor_boot_image=fake_vendor_boot_image 62EOF""" 63 64 LAUNCH_CVD_CMD_WITH_KERNEL_IMAGES = """sg group1 <<EOF 65sg group2 66bin/cvd start -daemon -config=phone -system_image_dir fake_image_dir -instance_dir fake_cvd_dir -undefok=report_anonymous_usage_stats,config -report_anonymous_usage_stats=y -start_vnc_server=true -kernel_path=fake_kernel_image -initramfs_path=fake_initramfs_image 67EOF""" 68 69 LAUNCH_CVD_CMD_WITH_VBMETA_IMAGE = """sg group1 <<EOF 70sg group2 71bin/cvd start -daemon -config=phone -system_image_dir fake_image_dir -instance_dir fake_cvd_dir -undefok=report_anonymous_usage_stats,config -report_anonymous_usage_stats=y -start_vnc_server=true -vbmeta_image=fake_vbmeta_image 72EOF""" 73 74 LAUNCH_CVD_CMD_WITH_ARGS = """sg group1 <<EOF 75sg group2 76bin/cvd start -daemon -config=phone -system_image_dir fake_image_dir -instance_dir fake_cvd_dir -undefok=report_anonymous_usage_stats,config -report_anonymous_usage_stats=y -start_vnc_server=true -setupwizard_mode=REQUIRED 77EOF""" 78 79 LAUNCH_CVD_CMD_WITH_OPENWRT = """sg group1 <<EOF 80sg group2 81bin/launch_cvd -daemon -config=phone -system_image_dir fake_image_dir -instance_dir fake_cvd_dir -undefok=report_anonymous_usage_stats,config -report_anonymous_usage_stats=y -start_vnc_server=true -console=true 82EOF""" 83 84 LAUNCH_CVD_CMD_WITH_PET_NAME = """sg group1 <<EOF 85sg group2 86bin/cvd start -daemon -config=phone -system_image_dir fake_image_dir -instance_dir fake_cvd_dir -undefok=report_anonymous_usage_stats,config -report_anonymous_usage_stats=y -start_vnc_server=true -webrtc_device_id=pet-name 87EOF""" 88 89 LAUNCH_CVD_CMD_WITH_NO_CVD = """sg group1 <<EOF 90sg group2 91bin/launch_cvd -daemon -config=phone -system_image_dir fake_image_dir -instance_dir fake_cvd_dir -undefok=report_anonymous_usage_stats,config -report_anonymous_usage_stats=y -start_vnc_server=true 92EOF""" 93 94 LAUNCH_CVD_CMD_WITH_INS_IDS = """sg group1 <<EOF 95sg group2 96bin/cvd start -daemon -config=phone -system_image_dir fake_image_dir -instance_dir fake_cvd_dir -undefok=report_anonymous_usage_stats,config -report_anonymous_usage_stats=y -start_vnc_server=true -instance_nums=1,2 97EOF""" 98 99 _EXPECTED_DEVICES_IN_REPORT = [ 100 { 101 "instance_name": "local-instance-1", 102 "ip": "0.0.0.0:6520", 103 "adb_port": 6520, 104 "vnc_port": 6444, 105 "webrtc_port": 8443, 106 'logs': [{'path': '/log/launcher.log', 'type': 'TEXT'}], 107 "screen_command": "screen /instances/cvd/console" 108 } 109 ] 110 111 _EXPECTED_DEVICES_IN_FAILED_REPORT = [ 112 { 113 "instance_name": "local-instance-1", 114 "ip": "0.0.0.0", 115 'logs': [{'path': '/log/launcher.log', 'type': 'TEXT'}], 116 } 117 ] 118 119 def setUp(self): 120 """Initialize new LocalImageLocalInstance.""" 121 super().setUp() 122 self.local_image_local_instance = local_image_local_instance.LocalImageLocalInstance() 123 124 # pylint: disable=protected-access 125 @mock.patch("acloud.create.local_image_local_instance.utils") 126 @mock.patch.object(local_image_local_instance.LocalImageLocalInstance, 127 "GetImageArtifactsPath") 128 @mock.patch.object(local_image_local_instance.LocalImageLocalInstance, 129 "_SelectAndLockInstance") 130 @mock.patch.object(local_image_local_instance.LocalImageLocalInstance, 131 "_CheckRunningCvd") 132 @mock.patch.object(local_image_local_instance.LocalImageLocalInstance, 133 "_CreateInstance") 134 def testCreateAVD(self, mock_create, mock_check_running_cvd, 135 mock_lock_instance, mock_get_image, mock_utils): 136 """Test _CreateAVD.""" 137 mock_utils.IsSupportedPlatform.return_value = True 138 mock_get_image.return_value = local_image_local_instance.ArtifactPaths( 139 "/image/path", "/host/bin/path", "host/usr/path", 140 None, None, # misc_info 141 None, None, None, # system 142 None, None, None, None, # boot 143 None, None, None, None) # vendor 144 mock_check_running_cvd.return_value = True 145 mock_avd_spec = mock.Mock() 146 mock_avd_spec.num_avds_per_instance = 1 147 mock_avd_spec.local_instance_dir = None 148 mock_lock = mock.Mock() 149 mock_lock.Unlock.return_value = False 150 mock_lock_instance.return_value = (1, mock_lock) 151 mock_report = mock.Mock() 152 mock_create.return_value = mock_report 153 154 # Success 155 mock_report.status = report.Status.SUCCESS 156 self.local_image_local_instance._CreateAVD( 157 mock_avd_spec, no_prompts=True) 158 mock_lock_instance.assert_called_once() 159 mock_lock.SetInUse.assert_called_once_with(True) 160 mock_lock.Unlock.assert_called_once() 161 162 mock_lock_instance.reset_mock() 163 mock_lock.SetInUse.reset_mock() 164 mock_lock.Unlock.reset_mock() 165 166 # Failure with report 167 mock_report.status = report.Status.BOOT_FAIL 168 self.local_image_local_instance._CreateAVD( 169 mock_avd_spec, no_prompts=True) 170 mock_lock_instance.assert_called_once() 171 mock_lock.SetInUse.assert_not_called() 172 mock_lock.Unlock.assert_called_once() 173 174 mock_lock_instance.reset_mock() 175 mock_lock.Unlock.reset_mock() 176 177 # Failure with no report 178 mock_create.side_effect = ValueError("unit test") 179 with self.assertRaises(ValueError): 180 self.local_image_local_instance._CreateAVD( 181 mock_avd_spec, no_prompts=True) 182 mock_lock_instance.assert_called_once() 183 mock_lock.SetInUse.assert_not_called() 184 mock_lock.Unlock.assert_called_once() 185 186 def testSelectAndLockInstances(self): 187 """test _SelectAndLockInstances.""" 188 mock_avd_spec = mock.Mock(num_avds_per_instance=1) 189 mock_main_lock = mock.Mock() 190 self.Patch(local_image_local_instance.LocalImageLocalInstance, 191 "_SelectAndLockInstance", return_value=(1, mock_main_lock)) 192 ins_ids, ins_locks = self.local_image_local_instance._SelectAndLockInstances( 193 mock_avd_spec) 194 self.assertEqual([1], ins_ids) 195 self.assertEqual([mock_main_lock], ins_locks) 196 197 mock_avd_spec.num_avds_per_instance = 2 198 mock_second_lock = mock.Mock() 199 self.Patch(local_image_local_instance.LocalImageLocalInstance, 200 "_SelectOneFreeInstance", return_value=(2, mock_second_lock)) 201 ins_ids, ins_locks = self.local_image_local_instance._SelectAndLockInstances( 202 mock_avd_spec) 203 self.assertEqual([1,2], ins_ids) 204 self.assertEqual([mock_main_lock, mock_second_lock], ins_locks) 205 206 def testSelectAndLockInstance(self): 207 """test _SelectAndLockInstance.""" 208 mock_avd_spec = mock.Mock(local_instance_id=0) 209 mock_lock = mock.Mock() 210 mock_lock.Lock.return_value = True 211 mock_lock.LockIfNotInUse.side_effect = (False, True) 212 self.Patch(instance, "GetLocalInstanceLock", 213 return_value=mock_lock) 214 215 ins_id, _ = self.local_image_local_instance._SelectAndLockInstance( 216 mock_avd_spec) 217 self.assertEqual(2, ins_id) 218 mock_lock.Lock.assert_not_called() 219 self.assertEqual(2, mock_lock.LockIfNotInUse.call_count) 220 221 mock_lock.LockIfNotInUse.reset_mock() 222 223 mock_avd_spec.local_instance_id = 1 224 ins_id, _ = self.local_image_local_instance._SelectAndLockInstance( 225 mock_avd_spec) 226 self.assertEqual(1, ins_id) 227 mock_lock.Lock.assert_called_once() 228 mock_lock.LockIfNotInUse.assert_not_called() 229 230 @mock.patch.object(local_image_local_instance.LocalImageLocalInstance, 231 "_TrustCertificatesForWebRTC") 232 @mock.patch("acloud.create.local_image_local_instance.utils") 233 @mock.patch("acloud.create.local_image_local_instance.ota_tools") 234 @mock.patch("acloud.create.local_image_local_instance.create_common") 235 @mock.patch.object(local_image_local_instance.LocalImageLocalInstance, 236 "_LogCvdVersion") 237 @mock.patch.object(local_image_local_instance.LocalImageLocalInstance, 238 "_LaunchCvd") 239 @mock.patch.object(local_image_local_instance.LocalImageLocalInstance, 240 "PrepareLaunchCVDCmd") 241 @mock.patch("acloud.create.local_image_local_instance.cvd_utils") 242 @mock.patch("acloud.create.local_image_local_instance.instance") 243 def testCreateInstance(self, mock_instance, mock_cvd_utils, 244 _mock_prepare_cmd, mock_launch_cvd, 245 mock_log_cvd_version, _mock_create_common, 246 mock_ota_tools, _mock_utils, mock_trust_certs): 247 """Test the report returned by _CreateInstance.""" 248 mock_instance.GetLocalInstanceHomeDir.return_value = ( 249 "/local-instance-1") 250 mock_instance.GetLocalInstanceName.return_value = "local-instance-1" 251 mock_instance.GetLocalInstanceRuntimeDir.return_value = ( 252 "/instances/cvd") 253 mock_instance.GetLocalInstanceConfig.return_value = ( 254 "/instances/cvd/config") 255 mock_cvd_utils.FindLocalLogs.return_value = [ 256 {'path': '/log/launcher.log', 'type': 'TEXT'}] 257 artifact_paths = local_image_local_instance.ArtifactPaths( 258 "/image/path", "/host/bin/path", "/host/usr/path", 259 "/misc/info/path", "/ota/tools/dir", "/system/image/path", 260 "/system_ext/image/path", "/product/image/path", "/boot/image/path", 261 "/vendor_boot/image/path", "/kernel/image/path", 262 "/initramfs/image/path", "/vendor/image/path", 263 "/vendor_dlkm/image/path", "/odm/image/path", 264 "/odm_dlkm/image/path") 265 mock_ota_tools_object = mock.Mock() 266 mock_ota_tools.OtaTools.return_value = mock_ota_tools_object 267 mock_avd_spec = mock.Mock( 268 unlock_screen=False, connect_webrtc=True, openwrt=True, 269 use_launch_cvd=False) 270 local_ins = mock.Mock( 271 adb_port=6520, 272 vnc_port=6444 273 ) 274 local_ins.CvdStatus.return_value = True 275 self.Patch(instance, "LocalInstance", 276 return_value=local_ins) 277 self.Patch(list_instance, "GetActiveCVD", 278 return_value=local_ins) 279 self.Patch(os, "symlink") 280 281 ins_ids = [1] 282 # Success 283 result_report = self.local_image_local_instance._CreateInstance( 284 ins_ids, artifact_paths, mock_avd_spec, no_prompts=True) 285 286 self.assertEqual(result_report.data.get("devices"), 287 self._EXPECTED_DEVICES_IN_REPORT) 288 mock_ota_tools.OtaTools.assert_called_with("/ota/tools/dir") 289 mock_ota_tools_object.MixSuperImage.assert_called_with( 290 "/local-instance-1/mixed_super.img", "/misc/info/path", 291 "/image/path", 292 system_image="/system/image/path", 293 system_ext_image="/system_ext/image/path", 294 product_image="/product/image/path", 295 vendor_image="/vendor/image/path", 296 vendor_dlkm_image="/vendor_dlkm/image/path", 297 odm_image="/odm/image/path", 298 odm_dlkm_image="/odm_dlkm/image/path") 299 mock_ota_tools_object.MakeDisabledVbmetaImage.assert_called_once() 300 mock_cvd_utils.FindLocalLogs.assert_called_with( 301 "/instances/cvd", 1) 302 mock_log_cvd_version.assert_called_with("/host/bin/path") 303 304 # should call _TrustCertificatesForWebRTC 305 mock_trust_certs.assert_called_once() 306 mock_trust_certs.reset_mock() 307 308 # should not call _TrustCertificatesForWebRTC 309 mock_avd_spec.connect_webrtc = False 310 self.local_image_local_instance._CreateInstance( 311 ins_ids, artifact_paths, mock_avd_spec, no_prompts=True) 312 mock_trust_certs.assert_not_called() 313 314 # Failure 315 mock_cvd_utils.reset_mock() 316 mock_launch_cvd.side_effect = errors.LaunchCVDFail("unit test") 317 318 result_report = self.local_image_local_instance._CreateInstance( 319 ins_ids, artifact_paths, mock_avd_spec, no_prompts=True) 320 321 self.assertEqual(result_report.data.get("devices_failing_boot"), 322 self._EXPECTED_DEVICES_IN_FAILED_REPORT) 323 self.assertIn("unit test", result_report.errors[0]) 324 mock_cvd_utils.FindLocalLogs.assert_called_with( 325 "/instances/cvd", 1) 326 327 # pylint: disable=protected-access 328 @mock.patch("acloud.create.local_image_local_instance.os.path.isfile") 329 def testFindCvdHostBinaries(self, mock_isfile): 330 """Test FindCvdHostBinaries.""" 331 cvd_host_dir = "/unit/test" 332 mock_isfile.return_value = None 333 334 with self.assertRaises(errors.GetCvdLocalHostPackageError): 335 self.local_image_local_instance._FindCvdHostBinaries( 336 [cvd_host_dir]) 337 338 mock_isfile.side_effect = ( 339 lambda path: path == "/unit/test/bin/launch_cvd") 340 341 path = self.local_image_local_instance._FindCvdHostBinaries( 342 [cvd_host_dir]) 343 self.assertEqual(path, cvd_host_dir) 344 345 @staticmethod 346 def _CreateEmptyFile(path): 347 driver_test_lib.BaseDriverTest.CreateFile(path) 348 349 @mock.patch("acloud.create.local_image_local_instance.ota_tools") 350 def testGetImageArtifactsPath(self, mock_ota_tools): 351 """Test GetImageArtifactsPath without system image dir.""" 352 with tempfile.TemporaryDirectory() as temp_dir: 353 image_dir = os.path.join(temp_dir, "image") 354 cvd_dir = os.path.join(temp_dir, "cvd-host_package") 355 self._CreateEmptyFile(os.path.join(cvd_dir, "bin", "launch_cvd")) 356 self._CreateEmptyFile(os.path.join(cvd_dir, "usr/share/webrtc/certs", "server.crt")) 357 358 mock_avd_spec = mock.Mock( 359 local_image_dir=image_dir, 360 local_kernel_image=None, 361 local_system_image=None, 362 local_vendor_image=None, 363 local_vendor_boot_image=None, 364 local_tool_dirs=[cvd_dir]) 365 366 with self.assertRaisesRegex( 367 errors.GetLocalImageError, 368 r"The directory is expected to be an extracted img zip " 369 r"or ANDROID_PRODUCT_OUT\."): 370 self.local_image_local_instance.GetImageArtifactsPath( 371 mock_avd_spec) 372 373 self._CreateEmptyFile(os.path.join(image_dir, "super.img")) 374 375 paths = self.local_image_local_instance.GetImageArtifactsPath( 376 mock_avd_spec) 377 378 mock_ota_tools.FindOtaToolsDir.assert_not_called() 379 self.assertEqual( 380 paths, 381 (image_dir, cvd_dir, cvd_dir, 382 None, None, # misc_info 383 None, None, None, # system 384 None, None, None, None, # boot 385 None, None, None, None)) # vendor 386 387 # pylint: disable=too-many-locals 388 @mock.patch("acloud.create.local_image_local_instance.ota_tools") 389 def testGetImageFromBuildEnvironment(self, mock_ota_tools): 390 """Test GetImageArtifactsPath with files in build environment.""" 391 with tempfile.TemporaryDirectory() as temp_dir: 392 image_dir = os.path.join(temp_dir, "image") 393 cvd_dir = os.path.join(temp_dir, "cvd-host_package") 394 mock_ota_tools.FindOtaToolsDir.return_value = cvd_dir 395 extra_image_dir = os.path.join(temp_dir, "extra_image") 396 system_image_path = os.path.join(extra_image_dir, "system.img") 397 system_ext_image_path = os.path.join(extra_image_dir, 398 "system_ext.img") 399 product_image_path = os.path.join(extra_image_dir, "product.img") 400 misc_info_path = os.path.join(image_dir, "misc_info.txt") 401 boot_image_path = os.path.join(extra_image_dir, "boot.img") 402 vendor_boot_image_path = os.path.join(extra_image_dir, 403 "vendor_boot.img") 404 vendor_image_path = os.path.join(extra_image_dir, "vendor.img") 405 vendor_dlkm_image_path = os.path.join(extra_image_dir, "vendor_dlkm.img") 406 odm_image_path = os.path.join(extra_image_dir, "odm.img") 407 odm_dlkm_image_path = os.path.join(extra_image_dir, "odm_dlkm.img") 408 self._CreateEmptyFile(os.path.join(image_dir, "vbmeta.img")) 409 self._CreateEmptyFile(os.path.join(cvd_dir, "bin", "launch_cvd")) 410 self._CreateEmptyFile(os.path.join(cvd_dir, "usr/share/webrtc/certs", "server.crt")) 411 self._CreateEmptyFile(system_image_path) 412 self._CreateEmptyFile(system_ext_image_path) 413 self._CreateEmptyFile(product_image_path) 414 self._CreateEmptyFile(os.path.join(extra_image_dir, 415 "boot-debug.img")) 416 self._CreateEmptyFile(misc_info_path) 417 self._CreateEmptyFile(vendor_image_path) 418 self._CreateEmptyFile(vendor_dlkm_image_path) 419 self._CreateEmptyFile(odm_image_path) 420 self._CreateEmptyFile(odm_dlkm_image_path) 421 self.CreateFile(boot_image_path, b"ANDROID!test_boot_image") 422 self.CreateFile(vendor_boot_image_path) 423 424 mock_avd_spec = mock.Mock( 425 local_image_dir=image_dir, 426 local_kernel_image=extra_image_dir, 427 local_system_image=extra_image_dir, 428 local_vendor_image=extra_image_dir, 429 local_vendor_boot_image=None, 430 local_tool_dirs=[]) 431 432 with mock.patch.dict("acloud.create.local_image_local_instance." 433 "os.environ", 434 {"ANDROID_SOONG_HOST_OUT": cvd_dir, 435 "ANDROID_HOST_OUT": "/cvd"}, 436 clear=True): 437 paths = self.local_image_local_instance.GetImageArtifactsPath( 438 mock_avd_spec) 439 440 mock_ota_tools.FindOtaToolsDir.assert_called_with([cvd_dir, "/cvd"]) 441 self.assertEqual( 442 paths, 443 (image_dir, cvd_dir, cvd_dir, misc_info_path, cvd_dir, 444 system_image_path, system_ext_image_path, product_image_path, 445 boot_image_path, vendor_boot_image_path, None, None, 446 vendor_image_path, vendor_dlkm_image_path, 447 odm_image_path, odm_dlkm_image_path)) 448 449 @mock.patch("acloud.create.local_image_local_instance.ota_tools") 450 def testGetImageFromTargetFiles(self, mock_ota_tools): 451 """Test GetImageArtifactsPath with extracted target files.""" 452 ota_tools_dir = "/mock_ota_tools" 453 mock_ota_tools.FindOtaToolsDir.return_value = ota_tools_dir 454 with tempfile.TemporaryDirectory() as temp_dir: 455 image_dir = os.path.join(temp_dir, "image") 456 cvd_dir = os.path.join(temp_dir, "cvd-host_package") 457 system_image_path = os.path.join(temp_dir, "system", "test.img") 458 misc_info_path = os.path.join(image_dir, "META", "misc_info.txt") 459 kernel_image_dir = os.path.join(temp_dir, "kernel_image") 460 kernel_image_path = os.path.join(kernel_image_dir, "Image") 461 initramfs_image_path = os.path.join(kernel_image_dir, 462 "initramfs.img") 463 464 self.CreateFile(os.path.join(kernel_image_dir, "boot.img")) 465 self.CreateFile(os.path.join(image_dir, "IMAGES", "vbmeta.img")) 466 self.CreateFile(os.path.join(cvd_dir, "bin", "launch_cvd")) 467 self.CreateFile(os.path.join(cvd_dir, "usr/share/webrtc/certs", 468 "server.crt")) 469 self.CreateFile(system_image_path) 470 self.CreateFile(misc_info_path) 471 self.CreateFile(kernel_image_path) 472 self.CreateFile(initramfs_image_path) 473 474 mock_avd_spec = mock.Mock( 475 local_image_dir=image_dir, 476 local_kernel_image=kernel_image_dir, 477 local_system_image=system_image_path, 478 local_vendor_image=None, 479 local_vendor_boot_image=None, 480 local_tool_dirs=[ota_tools_dir, cvd_dir]) 481 482 with mock.patch.dict("acloud.create.local_image_local_instance." 483 "os.environ", 484 clear=True): 485 paths = self.local_image_local_instance.GetImageArtifactsPath( 486 mock_avd_spec) 487 488 mock_ota_tools.FindOtaToolsDir.assert_called_with( 489 [ota_tools_dir, cvd_dir]) 490 self.assertEqual( 491 paths, 492 (os.path.join(image_dir, "IMAGES"), cvd_dir, cvd_dir, 493 misc_info_path, ota_tools_dir, 494 system_image_path, None, None, 495 None, None, kernel_image_path, initramfs_image_path, 496 None, None, None, None)) 497 498 @mock.patch.object(utils, "CheckUserInGroups") 499 def testPrepareLaunchCVDCmd(self, mock_usergroups): 500 """test PrepareLaunchCVDCmd.""" 501 mock_usergroups.return_value = False 502 self.Patch(os.path, "isfile", return_value=True) 503 hw_property = {"cpu": "fake", "x_res": "fake", "y_res": "fake", 504 "dpi":"fake", "memory": "fake", "disk": "fake"} 505 constants.LIST_CF_USER_GROUPS = ["group1", "group2"] 506 mock_artifact_paths = mock.Mock( 507 spec=[], 508 image_dir="fake_image_dir", 509 host_bins="", 510 host_artifacts="host_artifacts", 511 misc_info=None, 512 ota_tools_dir=None, 513 system_image=None, 514 boot_image=None, 515 vendor_boot_image=None, 516 kernel_image=None, 517 initramfs_image=None, 518 vendor_image=None, 519 vendor_dlkm_image=None, 520 odm_image=None, 521 odm_dlkm_image=None) 522 523 launch_cmd = self.local_image_local_instance.PrepareLaunchCVDCmd( 524 hw_property, True, mock_artifact_paths, "fake_cvd_dir", False, 525 True, None, None, "phone") 526 self.assertEqual(launch_cmd, self.LAUNCH_CVD_CMD_WITH_DISK) 527 528 # "disk" doesn't exist in hw_property. 529 hw_property = {"cpu": "fake", "x_res": "fake", "y_res": "fake", 530 "dpi": "fake", "memory": "fake"} 531 launch_cmd = self.local_image_local_instance.PrepareLaunchCVDCmd( 532 hw_property, True, mock_artifact_paths, "fake_cvd_dir", False, 533 True, None, None, "phone") 534 self.assertEqual(launch_cmd, self.LAUNCH_CVD_CMD_NO_DISK) 535 536 # "gpu" is enabled with "default" 537 launch_cmd = self.local_image_local_instance.PrepareLaunchCVDCmd( 538 hw_property, True, mock_artifact_paths, "fake_cvd_dir", False, 539 True, None, None, "phone") 540 self.assertEqual(launch_cmd, self.LAUNCH_CVD_CMD_NO_DISK_WITH_GPU) 541 542 # Following test with hw_property is None. 543 launch_cmd = self.local_image_local_instance.PrepareLaunchCVDCmd( 544 None, True, mock_artifact_paths, "fake_cvd_dir", True, False, 545 None, None, "auto") 546 self.assertEqual(launch_cmd, self.LAUNCH_CVD_CMD_WITH_WEBRTC) 547 548 # Mix super and boot images. 549 mock_artifact_paths.boot_image = "fake_boot_image" 550 mock_artifact_paths.vendor_boot_image = "fake_vendor_boot_image" 551 launch_cmd = self.local_image_local_instance.PrepareLaunchCVDCmd( 552 None, True, mock_artifact_paths, "fake_cvd_dir", False, True, 553 "fake_super_image", None, "phone") 554 self.assertEqual(launch_cmd, self.LAUNCH_CVD_CMD_WITH_MIXED_IMAGES) 555 mock_artifact_paths.boot_image = None 556 mock_artifact_paths.vendor_boot_image = None 557 558 # Mix kernel images. 559 mock_artifact_paths.kernel_image = "fake_kernel_image" 560 mock_artifact_paths.initramfs_image = "fake_initramfs_image" 561 launch_cmd = self.local_image_local_instance.PrepareLaunchCVDCmd( 562 None, True, mock_artifact_paths, "fake_cvd_dir", False, True, 563 None, None, "phone") 564 self.assertEqual(launch_cmd, self.LAUNCH_CVD_CMD_WITH_KERNEL_IMAGES) 565 mock_artifact_paths.kernel_image = None 566 mock_artifact_paths.initramfs_image = None 567 568 # Specify vbmeta image. 569 launch_cmd = self.local_image_local_instance.PrepareLaunchCVDCmd( 570 None, True, mock_artifact_paths, "fake_cvd_dir", False, True, 571 None, None, "phone", vbmeta_image_path="fake_vbmeta_image" 572 ) 573 self.assertEqual(launch_cmd, self.LAUNCH_CVD_CMD_WITH_VBMETA_IMAGE) 574 575 # Add args into launch command with "-setupwizard_mode=REQUIRED" 576 launch_cmd = self.local_image_local_instance.PrepareLaunchCVDCmd( 577 None, True, mock_artifact_paths, "fake_cvd_dir", False, True, 578 None, "-setupwizard_mode=REQUIRED", "phone") 579 self.assertEqual(launch_cmd, self.LAUNCH_CVD_CMD_WITH_ARGS) 580 581 # Test with "openwrt" and "use_launch_cvd" are enabled. 582 launch_cmd = self.local_image_local_instance.PrepareLaunchCVDCmd( 583 None, True, mock_artifact_paths, "fake_cvd_dir", False, True, 584 None, None, "phone", openwrt=True, use_launch_cvd=True) 585 self.assertEqual(launch_cmd, self.LAUNCH_CVD_CMD_WITH_OPENWRT) 586 587 # Test with instance_ids 588 launch_cmd = self.local_image_local_instance.PrepareLaunchCVDCmd( 589 None, True, mock_artifact_paths, "fake_cvd_dir", False, True, 590 None, None, "phone", instance_ids=[1,2]) 591 self.assertEqual(launch_cmd, self.LAUNCH_CVD_CMD_WITH_INS_IDS) 592 593 # Test with "pet-name" 594 launch_cmd = self.local_image_local_instance.PrepareLaunchCVDCmd( 595 None, True, mock_artifact_paths, "fake_cvd_dir", False, True, 596 None, None, "phone", webrtc_device_id="pet-name") 597 self.assertEqual(launch_cmd, self.LAUNCH_CVD_CMD_WITH_PET_NAME) 598 599 # Test with "cvd" doesn't exist 600 self.Patch(os.path, "isfile", return_value=False) 601 launch_cmd = self.local_image_local_instance.PrepareLaunchCVDCmd( 602 None, True, mock_artifact_paths, "fake_cvd_dir", False, True, 603 None, None, "phone", openwrt=False, use_launch_cvd=False) 604 self.assertEqual(launch_cmd, self.LAUNCH_CVD_CMD_WITH_NO_CVD) 605 606 @mock.patch("acloud.create.local_image_local_instance.subprocess.run") 607 def testLogCvdVersion(self, mock_run): 608 """Test _LogCvdVersion.""" 609 with tempfile.TemporaryDirectory() as temp_dir: 610 # cvd does not exist in old versions. 611 self.local_image_local_instance._LogCvdVersion(temp_dir) 612 mock_run.assert_not_called() 613 614 # cvd command completes. 615 mock_run.return_value = mock.Mock( 616 returncode=1, stdout=None, stderr="err") 617 cvd_path = os.path.join(temp_dir, "bin", "cvd") 618 self.CreateFile(cvd_path) 619 self.local_image_local_instance._LogCvdVersion(temp_dir) 620 mock_run.assert_called_once() 621 self.assertEqual(mock_run.call_args[0][0], f"{cvd_path} version") 622 623 # cvd cannot run. 624 mock_run.reset_mock() 625 mock_run.side_effect = subprocess.SubprocessError 626 self.local_image_local_instance._LogCvdVersion(temp_dir) 627 mock_run.assert_called_once() 628 629 @mock.patch.object(utils, "GetUserAnswerYes") 630 @mock.patch.object(list_instance, "GetActiveCVD") 631 def testCheckRunningCvd(self, mock_cvd_running, mock_get_answer): 632 """test _CheckRunningCvd.""" 633 local_instance_id = 3 634 635 # Test that launch_cvd is running. 636 mock_cvd_running.return_value = True 637 mock_get_answer.return_value = False 638 answer = self.local_image_local_instance._CheckRunningCvd( 639 local_instance_id) 640 self.assertFalse(answer) 641 642 # Test that launch_cvd is not running. 643 mock_cvd_running.return_value = False 644 answer = self.local_image_local_instance._CheckRunningCvd( 645 local_instance_id) 646 self.assertTrue(answer) 647 648 # pylint: disable=protected-access 649 @mock.patch("acloud.create.local_image_local_instance.subprocess.Popen") 650 @mock.patch.dict("os.environ", clear=True) 651 def testLaunchCVD(self, mock_popen): 652 """test _LaunchCvd should call subprocess.Popen with the env.""" 653 self.Patch(builtins, "open", mock.mock_open()) 654 local_instance_id = 3 655 launch_cvd_cmd = "launch_cvd" 656 host_bins_path = "host_bins_path" 657 host_artifacts_path = "host_artifacts_path" 658 cvd_home_dir = "fake_home" 659 timeout = 100 660 mock_proc = mock.Mock(returncode=0) 661 mock_popen.return_value = mock_proc 662 mock_proc.communicate.return_value = ("stdout", "stderr") 663 664 self.local_image_local_instance._LaunchCvd(launch_cvd_cmd, 665 local_instance_id, 666 host_bins_path, 667 host_artifacts_path, 668 cvd_home_dir, 669 timeout) 670 671 mock_popen.assert_called_once() 672 mock_proc.communicate.assert_called_once_with(timeout=timeout) 673 674 @mock.patch("acloud.create.local_image_local_instance.subprocess.Popen") 675 def testLaunchCVDFailure(self, mock_popen): 676 """test _LaunchCvd with subprocess errors.""" 677 self.Patch(builtins, "open", mock.mock_open()) 678 mock_proc = mock.Mock(returncode=9) 679 mock_popen.return_value = mock_proc 680 with self.assertRaises(errors.LaunchCVDFail) as launch_cvd_failure: 681 self.local_image_local_instance._LaunchCvd("launch_cvd", 682 3, 683 "host_bins_path", 684 "host_artifacts_path", 685 "cvd_home_dir", 686 100) 687 self.assertIn("returned 9", str(launch_cvd_failure.exception)) 688 689 @mock.patch("acloud.create.local_image_local_instance.list_instance") 690 @mock.patch("acloud.create.local_image_local_instance.subprocess.Popen") 691 def testLaunchCVDTimeout(self, mock_popen, mock_list_instance): 692 """test _LaunchCvd with subprocess timeout.""" 693 self.Patch(builtins, "open", mock.mock_open()) 694 mock_proc = mock.Mock(returncode=255) 695 mock_popen.return_value = mock_proc 696 mock_proc.communicate.side_effect = [ 697 subprocess.TimeoutExpired(cmd="launch_cvd", timeout=100), 698 ("stdout", "stderr") 699 ] 700 mock_instance = mock.Mock() 701 mock_list_instance.GetActiveCVD.return_value = mock_instance 702 mock_instance.Delete.side_effect = subprocess.CalledProcessError( 703 cmd="stop_cvd", returncode=255) 704 with self.assertRaises(errors.LaunchCVDFail) as launch_cvd_failure: 705 self.local_image_local_instance._LaunchCvd("launch_cvd", 706 3, 707 "host_bins_path", 708 "host_artifacts_path", 709 "cvd_home_dir", 710 100) 711 self.assertIn("100 secs", str(launch_cvd_failure.exception)) 712 mock_list_instance.GetActiveCVD.assert_called_with(3) 713 mock_instance.Delete.assert_called() 714 mock_proc.terminate.assert_called() 715 716 def testGetWebrtcSigServerPort(self): 717 """test GetWebrtcSigServerPort.""" 718 instance_id = 3 719 expected_port = 8445 720 self.assertEqual( 721 self.local_image_local_instance.GetWebrtcSigServerPort(instance_id), 722 expected_port) 723 724 def testGetConfigFromAndroidInfo(self): 725 """Test GetConfigFromAndroidInfo""" 726 self.Patch(os.path, "exists", return_value=True) 727 mock_open = mock.mock_open(read_data="config=phone") 728 expected = "phone" 729 with mock.patch("builtins.open", mock_open): 730 self.assertEqual( 731 self.local_image_local_instance._GetConfigFromAndroidInfo("file"), 732 expected) 733 734 735if __name__ == "__main__": 736 unittest.main() 737