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