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