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 instance class.""" 17 18import collections 19import datetime 20import os 21import subprocess 22import unittest 23 24from unittest import mock 25 26# pylint: disable=import-error 27import dateutil.parser 28import dateutil.tz 29 30from acloud.internal import constants 31from acloud.internal.lib import cvd_runtime_config 32from acloud.internal.lib import driver_test_lib 33from acloud.internal.lib import gcompute_client 34from acloud.internal.lib import utils 35from acloud.internal.lib.adb_tools import AdbTools 36from acloud.list import instance 37 38 39ForwardedPorts = collections.namedtuple("ForwardedPorts", 40 [constants.VNC_PORT, 41 constants.ADB_PORT, 42 constants.FASTBOOT_PORT]) 43 44class InstanceTest(driver_test_lib.BaseDriverTest): 45 """Test instance.""" 46 PS_SSH_TUNNEL = ("/fake_ps_1 --fake arg \n" 47 "/fake_ps_2 --fake arg \n" 48 "/usr/bin/ssh -i ~/.ssh/acloud_rsa " 49 "-o UserKnownHostsFile=/dev/null " 50 "-o StrictHostKeyChecking=no -L 54321:127.0.0.1:6520 -L 6789:127.0.0.1:7520" 51 "-L 12345:127.0.0.1:6444 -N -f -l user 1.1.1.1").encode() 52 GCE_INSTANCE = { 53 constants.INS_KEY_NAME: "fake_ins_name", 54 constants.INS_KEY_CREATETIME: "fake_create_time", 55 constants.INS_KEY_STATUS: "fake_status", 56 constants.INS_KEY_ZONE: "test/zones/fake_zone", 57 "networkInterfaces": [{"accessConfigs": [{"natIP": "1.1.1.1"}]}], 58 "labels": {constants.INS_KEY_AVD_TYPE: "fake_type", 59 constants.INS_KEY_AVD_FLAVOR: "fake_flavor"}, 60 "metadata": { 61 "items":[{"key":constants.INS_KEY_AVD_TYPE, 62 "value":"fake_type"}, 63 {"key":constants.INS_KEY_AVD_FLAVOR, 64 "value":"fake_flavor"}, 65 {"key":constants.INS_KEY_WEBRTC_PORT, 66 "value":"fake_webrtc_port"}]} 67 } 68 69 @staticmethod 70 def _MockCvdRuntimeConfig(): 71 """Create a mock CvdRuntimeConfig.""" 72 return mock.MagicMock( 73 instance_id=2, 74 display_configs=[{'dpi': 480, 'x_res': 1080, 'y_res': 1920}], 75 instance_dir="fake_instance_dir", 76 adb_port=6521, 77 vnc_port=6445, 78 adb_ip_port="127.0.0.1:6521", 79 cvd_tools_path="fake_cvd_tools_path", 80 config_path="fake_config_path", 81 instances={}, 82 root_dir="/tmp/acloud_cvd_temp/local-instance-2/cuttlefish_runtime" 83 ) 84 85 @mock.patch("acloud.list.instance.AdbTools") 86 def testCreateLocalInstance(self, mock_adb_tools): 87 """Test getting local instance info from cvd runtime config.""" 88 mock_adb_tools_object = mock.Mock(device_information={}) 89 mock_adb_tools_object.IsAdbConnected.return_value = True 90 mock_adb_tools.return_value = mock_adb_tools_object 91 self.Patch(cvd_runtime_config, "CvdRuntimeConfig", 92 return_value=self._MockCvdRuntimeConfig()) 93 self.Patch(instance.LocalInstance, "_GetDevidInfoFromCvdStatus", 94 return_value=None) 95 local_instance = instance.LocalInstance("fake_config_path") 96 97 self.assertEqual("local-instance-2", local_instance.name) 98 self.assertEqual(True, local_instance.islocal) 99 self.assertEqual(["1080x1920 (480)"], local_instance.display) 100 expected_full_name = ("device serial: 0.0.0.0:%s %s (%s) elapsed time: %s" 101 % ("6521", 102 "cvd-2", 103 "local-instance-2", 104 "None")) 105 self.assertEqual(expected_full_name, local_instance.fullname) 106 self.assertEqual(6521, local_instance.adb_port) 107 self.assertEqual(6445, local_instance.vnc_port) 108 self.assertEqual(8444, local_instance.webrtc_port) 109 110 # pylint: disable=protected-access 111 def testGetCvdEnv(self): 112 """Test GetCvdEnv.""" 113 self.Patch(cvd_runtime_config, "CvdRuntimeConfig", 114 return_value=self._MockCvdRuntimeConfig()) 115 self.Patch(instance, "_IsProcessRunning", return_value=False) 116 local_instance = instance.LocalInstance("fake_config_path") 117 cvd_env = local_instance._GetCvdEnv() 118 self.assertEqual(cvd_env[constants.ENV_CUTTLEFISH_INSTANCE], "2") 119 self.assertEqual(cvd_env[constants.ENV_CUTTLEFISH_CONFIG_FILE], 120 "fake_config_path") 121 122 # pylint: disable=protected-access 123 def testParsingCvdFleetOutput(self): 124 """Test ParsingCvdFleetOutput.""" 125 cvd_fleet_output = """WARNING: cvd_server client version does not match 126{ 127"adb_serial" : "0.0.0.0:6520", 128"instance_name" : "cvd-1", 129}""" 130 131 expected_result = """{ 132"adb_serial" : "0.0.0.0:6520", 133"instance_name" : "cvd-1", 134}""" 135 136 self.assertEqual( 137 instance.LocalInstance._ParsingCvdFleetOutput(cvd_fleet_output), 138 expected_result) 139 140 # pylint: disable=protected-access 141 def testIsProcessRunning(self): 142 """Test IsProcessRunning.""" 143 process = "cvd_server" 144 self.Patch(utils, "CheckOutput", 145 return_value="/bin/cvd_server -server_fd=4") 146 self.assertEqual(instance._IsProcessRunning(process), True) 147 148 self.Patch(utils, "CheckOutput", return_value="/bin/cvd start") 149 self.assertEqual(instance._IsProcessRunning(process), False) 150 151 @mock.patch("acloud.list.instance.AdbTools") 152 def testDeleteLocalInstance(self, mock_adb_tools): 153 """Test executing 'cvd stop' command.""" 154 self.Patch(cvd_runtime_config, "CvdRuntimeConfig", 155 return_value=self._MockCvdRuntimeConfig()) 156 mock_adb_tools_object = mock.Mock(device_information={}) 157 mock_adb_tools_object.IsAdbConnected.return_value = True 158 mock_adb_tools.return_value = mock_adb_tools_object 159 self.Patch(utils, "AddUserGroupsToCmd", 160 side_effect=lambda cmd, groups: cmd) 161 self.Patch(instance.LocalInstance, "_GetDevidInfoFromCvdStatus", 162 return_value=None) 163 mock_check_call = self.Patch(subprocess, "check_call") 164 mock_check_output = self.Patch( 165 subprocess, "check_output", 166 return_value="cvd_internal_stop E stop cvd failed") 167 168 local_instance = instance.LocalInstance("fake_config_path") 169 with mock.patch.dict("acloud.list.instance.os.environ", clear=True): 170 local_instance.Delete() 171 172 expected_env = { 173 "CUTTLEFISH_INSTANCE": "2", 174 "HOME": "/tmp/acloud_cvd_temp/local-instance-2", 175 "CUTTLEFISH_CONFIG_FILE": "fake_config_path", 176 "ANDROID_HOST_OUT": "", 177 "ANDROID_SOONG_HOST_OUT": "", 178 } 179 mock_check_output.assert_called_with( 180 "/tmp/acloud_cvd_temp/local-instance-2/host_bins/bin/cvd stop", 181 stderr=subprocess.STDOUT, shell=True, env=expected_env, text=True, 182 timeout=instance._CVD_TIMEOUT) 183 mock_check_call.assert_called_with( 184 "/tmp/acloud_cvd_temp/local-instance-2/host_bins/bin/stop_cvd", 185 stderr=subprocess.STDOUT, shell=True, env=expected_env) 186 mock_adb_tools_object.DisconnectAdb.assert_called() 187 188 @mock.patch("acloud.list.instance.tempfile") 189 @mock.patch("acloud.list.instance.AdbTools") 190 def testCreateLocalGoldfishInstance(self, mock_adb_tools, mock_tempfile): 191 """"Test the attributes of LocalGoldfishInstance.""" 192 mock_tempfile.gettempdir.return_value = "/unit/test" 193 mock_adb_tools.return_value = mock.Mock(device_information={}) 194 195 inst = instance.LocalGoldfishInstance(1) 196 197 self.assertEqual(inst.name, "local-goldfish-instance-1") 198 self.assertEqual(inst.avd_type, constants.TYPE_GF) 199 self.assertEqual(inst.adb_port, 5555) 200 self.assertTrue(inst.islocal) 201 self.assertEqual(inst.console_port, 5554) 202 self.assertEqual(inst.device_serial, "emulator-5554") 203 self.assertEqual(inst.instance_dir, 204 "/unit/test/acloud_gf_temp/local-goldfish-instance-1") 205 206 @mock.patch("acloud.list.instance.AdbTools") 207 def testGetLocalGoldfishInstances(self, mock_adb_tools): 208 """Test LocalGoldfishInstance.GetExistingInstances.""" 209 mock_adb_tools.GetDeviceSerials.return_value = [ 210 "127.0.0.1:6520", "emulator-5554", "ABCD", "emulator-5558"] 211 212 instances = instance.LocalGoldfishInstance.GetExistingInstances() 213 214 self.assertEqual(len(instances), 2) 215 self.assertEqual(instances[0].console_port, 5554) 216 self.assertEqual(instances[0].name, "local-goldfish-instance-1") 217 self.assertEqual(instances[1].console_port, 5558) 218 self.assertEqual(instances[1].name, "local-goldfish-instance-3") 219 220 def testGetMaxNumberOfGoldfishInstances(self): 221 """Test LocalGoldfishInstance.GetMaxNumberOfInstances.""" 222 mock_environ = {} 223 with mock.patch.dict("acloud.list.instance.os.environ", 224 mock_environ, clear=True): 225 num = instance.LocalGoldfishInstance.GetMaxNumberOfInstances() 226 self.assertEqual(num, 16) 227 228 mock_environ["ADB_LOCAL_TRANSPORT_MAX_PORT"] = "5565" 229 with mock.patch.dict("acloud.list.instance.os.environ", 230 mock_environ, clear=True): 231 num = instance.LocalGoldfishInstance.GetMaxNumberOfInstances() 232 self.assertEqual(num, 6) 233 234 # pylint: disable=protected-access 235 def testGetElapsedTime(self): 236 """Test _GetElapsedTime""" 237 # Instance time can't parse 238 start_time = "error time" 239 self.assertEqual(instance._MSG_UNABLE_TO_CALCULATE, 240 instance._GetElapsedTime(start_time)) 241 242 # Remote instance elapsed time 243 now = "2019-01-14T13:00:00.000-07:00" 244 start_time = "2019-01-14T03:00:00.000-07:00" 245 self.Patch(instance, "datetime") 246 instance.datetime.datetime.now.return_value = dateutil.parser.parse(now) 247 self.assertEqual( 248 datetime.timedelta(hours=10), instance._GetElapsedTime(start_time)) 249 250 # Local instance elapsed time 251 now = "Mon Jan 14 10:10:10 2019" 252 start_time = "Mon Jan 14 08:10:10 2019" 253 instance.datetime.datetime.now.return_value = dateutil.parser.parse( 254 now).replace(tzinfo=dateutil.tz.tzlocal()) 255 self.assertEqual( 256 datetime.timedelta(hours=2), instance._GetElapsedTime(start_time)) 257 258 # pylint: disable=protected-access 259 def testGetForwardedPortsFromSSHTunnel(self): 260 """"Test Get forwarding adb and vnc port from ssh tunnel.""" 261 self.Patch(subprocess, "check_output", return_value=self.PS_SSH_TUNNEL) 262 self.Patch(instance, "_GetElapsedTime", return_value="fake_time") 263 self.Patch(instance.RemoteInstance, "_GetZoneName", return_value="fake_zone") 264 self.Patch(instance.RemoteInstance, 265 "_GetProjectName", 266 return_value="fake_project") 267 self.Patch(gcompute_client, "GetGCEHostName", return_value="fake_hostname") 268 forwarded_ports = instance.RemoteInstance( 269 mock.MagicMock()).GetForwardedPortsFromSSHTunnel( 270 "1.1.1.1", "fake_hostname", constants.TYPE_CF) 271 self.assertEqual(54321, forwarded_ports.adb_port) 272 self.assertEqual(6789, forwarded_ports.fastboot_port) 273 self.assertEqual(12345, forwarded_ports.vnc_port) 274 275 # If avd_type is undefined in utils.AVD_PORT_DICT. 276 forwarded_ports = instance.RemoteInstance( 277 mock.MagicMock()).GetForwardedPortsFromSSHTunnel( 278 "1.1.1.1", "fake_hostname", "undefined_avd_type") 279 self.assertEqual(None, forwarded_ports.adb_port) 280 self.assertEqual(None, forwarded_ports.fastboot_port) 281 self.assertEqual(None, forwarded_ports.vnc_port) 282 283 # pylint: disable=protected-access 284 def testProcessGceInstance(self): 285 """"Test process instance detail.""" 286 fake_adb = 123456 287 fake_fastboot = 654321 288 fake_vnc = 654321 289 290 self.Patch(instance.RemoteInstance, 291 "_GetProjectName", 292 return_value="fake_project") 293 self.Patch( 294 instance.RemoteInstance, 295 "GetForwardedPortsFromSSHTunnel", 296 return_value=ForwardedPorts(vnc_port=fake_vnc, 297 adb_port=fake_adb, 298 fastboot_port=fake_fastboot)) 299 self.Patch(utils, "GetWebrtcPortFromSSHTunnel", 300 return_value="fake_webrtc_port") 301 self.Patch(instance, "_GetElapsedTime", return_value="fake_time") 302 self.Patch(AdbTools, "IsAdbConnected", return_value=True) 303 304 # test ssh_tunnel_is_connected will be true if ssh tunnel connection is found 305 instance_info = instance.RemoteInstance(self.GCE_INSTANCE) 306 self.assertTrue(instance_info.ssh_tunnel_is_connected) 307 self.assertEqual(instance_info.adb_port, fake_adb) 308 self.assertEqual(instance_info.vnc_port, fake_vnc) 309 self.assertEqual("1.1.1.1", instance_info.ip) 310 self.assertEqual("fake_status", instance_info.status) 311 self.assertEqual("fake_type", instance_info.avd_type) 312 self.assertEqual("fake_flavor", instance_info.avd_flavor) 313 expected_full_name = "device serial: 127.0.0.1:%s (%s) elapsed time: %s" % ( 314 fake_adb, self.GCE_INSTANCE[constants.INS_KEY_NAME], "fake_time") 315 self.assertEqual(expected_full_name, instance_info.fullname) 316 317 # test ssh tunnel is connected but adb is disconnected 318 self.Patch(AdbTools, "IsAdbConnected", return_value=False) 319 instance_info = instance.RemoteInstance(self.GCE_INSTANCE) 320 self.assertTrue(instance_info.ssh_tunnel_is_connected) 321 expected_full_name = "device serial: not connected (%s) elapsed time: %s" % ( 322 instance_info.name, "fake_time") 323 self.assertEqual(expected_full_name, instance_info.fullname) 324 325 # test ssh_tunnel_is_connected will be false if ssh tunnel connection is not found 326 self.Patch( 327 instance.RemoteInstance, 328 "GetForwardedPortsFromSSHTunnel", 329 return_value=ForwardedPorts(vnc_port=None, adb_port=None, fastboot_port=None)) 330 instance_info = instance.RemoteInstance(self.GCE_INSTANCE) 331 self.assertFalse(instance_info.ssh_tunnel_is_connected) 332 expected_full_name = "device serial: not connected (%s) elapsed time: %s" % ( 333 self.GCE_INSTANCE[constants.INS_KEY_NAME], "fake_time") 334 self.assertEqual(expected_full_name, instance_info.fullname) 335 336 def testInstanceSummary(self): 337 """Test instance summary.""" 338 fake_adb = 123456 339 fake_fastboot = 654321 340 fake_vnc = 654321 341 self.Patch(instance.RemoteInstance, 342 "_GetProjectName", 343 return_value="fake_project") 344 self.Patch( 345 instance.RemoteInstance, 346 "GetForwardedPortsFromSSHTunnel", 347 return_value=ForwardedPorts(vnc_port=fake_vnc, 348 adb_port=fake_adb, 349 fastboot_port=fake_fastboot)) 350 self.Patch(utils, "GetWebrtcPortFromSSHTunnel", return_value=8443) 351 self.Patch(instance, "_GetElapsedTime", return_value="fake_time") 352 self.Patch(AdbTools, "IsAdbConnected", return_value=True) 353 remote_instance = instance.RemoteInstance(self.GCE_INSTANCE) 354 result_summary = (" name: fake_ins_name\n " 355 " IP: 1.1.1.1\n " 356 " create time: fake_create_time\n " 357 " elapse time: fake_time\n " 358 " status: fake_status\n " 359 " avd type: fake_type\n " 360 " display: None\n " 361 " vnc: 127.0.0.1:654321\n " 362 " zone: fake_zone\n " 363 " autoconnect: webrtc\n " 364 " webrtc port: fake_webrtc_port\n " 365 " webrtc forward port: 8443\n " 366 " adb serial: 127.0.0.1:123456\n " 367 " product: None\n " 368 " model: None\n " 369 " device: None\n " 370 " transport_id: None") 371 self.assertEqual(remote_instance.Summary(), result_summary) 372 373 self.Patch( 374 instance.RemoteInstance, 375 "GetForwardedPortsFromSSHTunnel", 376 return_value=ForwardedPorts(vnc_port=None, adb_port=None, fastboot_port=None)) 377 self.Patch(instance, "_GetElapsedTime", return_value="fake_time") 378 self.Patch(AdbTools, "IsAdbConnected", return_value=False) 379 remote_instance = instance.RemoteInstance(self.GCE_INSTANCE) 380 result_summary = (" name: fake_ins_name\n " 381 " IP: 1.1.1.1\n " 382 " create time: fake_create_time\n " 383 " elapse time: fake_time\n " 384 " status: fake_status\n " 385 " avd type: fake_type\n " 386 " display: None\n " 387 " vnc: 127.0.0.1:None\n " 388 " zone: fake_zone\n " 389 " autoconnect: webrtc\n " 390 " webrtc port: fake_webrtc_port\n " 391 " webrtc forward port: 8443\n " 392 " adb serial: disconnected") 393 self.assertEqual(remote_instance.Summary(), result_summary) 394 395 def testGetZoneName(self): 396 """Test GetZoneName.""" 397 zone_info = "v1/projects/project/zones/us-central1-c" 398 expected_result = "us-central1-c" 399 self.assertEqual(instance.RemoteInstance._GetZoneName(zone_info), 400 expected_result) 401 # Test can't get zone name from zone info. 402 zone_info = "v1/projects/project/us-central1-c" 403 self.assertEqual(instance.RemoteInstance._GetZoneName(zone_info), None) 404 405 def testGetProjectName(self): 406 """Test GetProjectName.""" 407 zone_info = "v1/projects/fake_project/zones/us-central1-c" 408 expected_result = "fake_project" 409 self.assertEqual(instance.RemoteInstance._GetProjectName(zone_info), 410 expected_result) 411 412 def testGetLocalInstanceConfig(self): 413 """Test GetLocalInstanceConfig.""" 414 self.Patch(instance, "GetLocalInstanceHomeDir", 415 return_value="ins_home") 416 self.Patch(os.path, "isfile", return_value=False) 417 instance_id = 1 418 self.assertEqual(instance.GetLocalInstanceConfig(instance_id), None) 419 420 # Test config in new folder path. 421 self.Patch(os.path, "isfile", return_value=True) 422 expected_result = "ins_home/cuttlefish_assembly/cuttlefish_config.json" 423 self.assertEqual( 424 instance.GetLocalInstanceConfig(instance_id), expected_result) 425 426 def testGetAutoConnect(self): 427 """Test GetAutoConnect.""" 428 name = "ins_name" 429 fullname = "fake_fullname" 430 display = "1080x1920 (480)" 431 ip = "fake_ip" 432 ins_webrtc = instance.Instance( 433 name, fullname, display, ip, webrtc_port=8443) 434 self.assertEqual(ins_webrtc._GetAutoConnect(), constants.INS_KEY_WEBRTC) 435 436 ins_webrtc = instance.Instance( 437 name, fullname, display, ip, vnc_port=6520) 438 self.assertEqual(ins_webrtc._GetAutoConnect(), constants.INS_KEY_VNC) 439 440 ins_webrtc = instance.Instance( 441 name, fullname, display, ip, adb_port=6666) 442 self.assertEqual(ins_webrtc._GetAutoConnect(), constants.INS_KEY_ADB) 443 444 ins_webrtc = instance.Instance( 445 name, fullname, display, ip, fastboot_port=6666) 446 self.assertEqual(ins_webrtc._GetAutoConnect(), constants.INS_KEY_FASTBOOT) 447 448 ins_webrtc = instance.Instance(name, fullname, display, ip) 449 self.assertEqual(ins_webrtc._GetAutoConnect(), None) 450 451 @mock.patch("acloud.list.instance.LocalInstance") 452 def testGetCuttleFishLocalInstances(self, mock_local_instance): 453 """Test GetCuttleFishLocalInstances.""" 454 self.Patch(cvd_runtime_config, "CvdRuntimeConfig", 455 return_value=mock.MagicMock(instance_ids=["2", "3"])) 456 instance.GetCuttleFishLocalInstances("fake_config_path") 457 self.assertEqual(mock_local_instance.call_count, 2) 458 459 def testGetDeviceFullName(self): 460 """Test GetDeviceFullName.""" 461 device_serial = "0.0.0.0:6520" 462 webrtc_device_id = "codelab" 463 instance_name = "local-instance-1" 464 elapsed_time = "10:10:24" 465 466 expected_result = ("device serial: 0.0.0.0:6520 codelab " 467 "(local-instance-1) elapsed time: 10:10:24") 468 self.assertEqual(expected_result, instance._GetDeviceFullName( 469 device_serial, instance_name, elapsed_time, webrtc_device_id)) 470 471 # Test with no webrtc_device_id 472 webrtc_device_id = None 473 expected_result = ("device serial: 0.0.0.0:6520 (local-instance-1) " 474 "elapsed time: 10:10:24") 475 self.assertEqual(expected_result, instance._GetDeviceFullName( 476 device_serial, instance_name, elapsed_time, webrtc_device_id)) 477 478 479if __name__ == "__main__": 480 unittest.main() 481