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