1# Copyright 2021 - The Android Open Source Project 2# 3# Licensed under the Apache License, Version 2.0 (the "License"); 4# you may not use this file except in compliance with the License. 5# You may obtain a copy of the License at 6# 7# http://www.apache.org/licenses/LICENSE-2.0 8# 9# Unless required by applicable law or agreed to in writing, software 10# distributed under the License is distributed on an "AS IS" BASIS, 11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12# See the License for the specific language governing permissions and 13# limitations under the License. 14"""Tests for RemoteImageRemoteInstance.""" 15 16import unittest 17 18from unittest import mock 19 20from acloud.create import remote_image_remote_instance 21from acloud.internal import constants 22from acloud.internal.lib import driver_test_lib 23from acloud.internal.lib import oxygen_client 24from acloud.internal.lib import utils 25from acloud.public import report 26from acloud.public.actions import common_operations 27from acloud.public.actions import remote_instance_cf_device_factory 28 29 30ONE_LINE_LEASE_RESPONSE = ("2021/08/02 11:28:52 session_id:\"fake_device\" " 31 "server_url:\"10.1.1.1\" ports:{type:WATERFALL value:0}") 32MULTIPLE_LINES_LEASE_RESPONSE = """ 332021/08/02 11:28:52 34session_id:"fake_device" 35server_url:"10.1.1.1" 36""" 37LEASE_FAILURE_RESPONSE = """ 382021/08/16 18:07:36 message 1 392021/08/16 18:11:36 Error received while trying to lease device: rpc error: details 40""" 41 42 43class RemoteImageRemoteInstanceTest(driver_test_lib.BaseDriverTest): 44 """Test RemoteImageRemoteInstance method.""" 45 46 def setUp(self): 47 """Initialize new RemoteImageRemoteInstance.""" 48 super().setUp() 49 self.remote_image_remote_instance = remote_image_remote_instance.RemoteImageRemoteInstance() 50 51 # pylint: disable=protected-access 52 @mock.patch.object(utils, "LaunchVNCFromReport") 53 @mock.patch.object(utils, "LaunchBrowserFromReport") 54 @mock.patch.object(remote_image_remote_instance.RemoteImageRemoteInstance, 55 "_LeaseOxygenAVD") 56 @mock.patch.object(remote_instance_cf_device_factory, 57 "RemoteInstanceDeviceFactory") 58 def testCreateAVD(self, mock_factory, mock_lease, mock_launch_browser, 59 mock_launch_vnc): 60 """test CreateAVD.""" 61 avd_spec = mock.Mock() 62 avd_spec.oxygen = False 63 avd_spec.connect_webrtc = True 64 avd_spec.connect_vnc = False 65 avd_spec.gce_only = False 66 avd_spec.avd_type = constants.TYPE_CF 67 create_report = mock.Mock() 68 create_report.status = report.Status.SUCCESS 69 self.Patch(common_operations, "CreateDevices", 70 return_value=create_report) 71 self.remote_image_remote_instance._CreateAVD( 72 avd_spec, no_prompts=True) 73 mock_factory.assert_called_once() 74 mock_launch_browser.assert_called_once() 75 76 # Test launch VNC case. 77 avd_spec.connect_webrtc = False 78 avd_spec.connect_vnc = True 79 self.remote_image_remote_instance._CreateAVD( 80 avd_spec, no_prompts=True) 81 mock_launch_vnc.assert_called_once() 82 83 # Test launch with Oxgen case. 84 avd_spec.oxygen = True 85 self.remote_image_remote_instance._CreateAVD( 86 avd_spec, no_prompts=True) 87 mock_lease.assert_called_once() 88 89 def testLeaseOxygenAVD(self): 90 """test LeaseOxygenAVD.""" 91 avd_spec = mock.Mock() 92 avd_spec.oxygen = True 93 avd_spec.remote_image = {constants.BUILD_TARGET: "fake_target", 94 constants.BUILD_ID: "fake_id", 95 constants.BUILD_BRANCH: "fake_branch"} 96 avd_spec.system_build_info = {constants.BUILD_TARGET: "fake_target", 97 constants.BUILD_ID: "fake_id", 98 constants.BUILD_BRANCH: "fake_branch"} 99 avd_spec.kernel_build_info = {constants.BUILD_TARGET: "fake_target", 100 constants.BUILD_ID: "fake_id", 101 constants.BUILD_BRANCH: "fake_branch"} 102 response_fail = "Lease device fail." 103 self.Patch(oxygen_client.OxygenClient, "LeaseDevice", 104 side_effect=[ONE_LINE_LEASE_RESPONSE, response_fail]) 105 expected_status = report.Status.SUCCESS 106 reporter = self.remote_image_remote_instance._LeaseOxygenAVD(avd_spec) 107 self.assertEqual(reporter.status, expected_status) 108 109 expected_status = report.Status.FAIL 110 reporter = self.remote_image_remote_instance._LeaseOxygenAVD(avd_spec) 111 self.assertEqual(reporter.status, expected_status) 112 113 def testOxygenLeaseFailure(self): 114 """test LeaseOxygenAVD when the lease call failed.""" 115 avd_spec = mock.Mock() 116 avd_spec.oxygen = True 117 avd_spec.remote_image = {constants.BUILD_TARGET: "fake_target", 118 constants.BUILD_ID: "fake_id", 119 constants.BUILD_BRANCH: "fake_branch"} 120 avd_spec.system_build_info = {constants.BUILD_TARGET: "fake_target", 121 constants.BUILD_ID: "fake_id", 122 constants.BUILD_BRANCH: "fake_branch"} 123 avd_spec.kernel_build_info = {constants.BUILD_TARGET: "fake_target", 124 constants.BUILD_ID: "fake_id", 125 constants.BUILD_BRANCH: "fake_branch"} 126 response_fail = "Lease device fail." 127 self.Patch(oxygen_client.OxygenClient, "LeaseDevice", 128 side_effect=[LEASE_FAILURE_RESPONSE, response_fail]) 129 expected_status = report.Status.FAIL 130 reporter = self.remote_image_remote_instance._LeaseOxygenAVD(avd_spec) 131 self.assertEqual(reporter.status, expected_status) 132 self.assertEqual(reporter.error_type, constants.ACLOUD_OXYGEN_LEASE_ERROR) 133 self.assertEqual(reporter.errors, ["rpc error: details"]) 134 135 def testGetDeviceInfoFromResponse(self): 136 """test GetDeviceInfoFromResponse.""" 137 expect_session_id = "fake_device" 138 expect_server_url = "10.1.1.1" 139 self.assertEqual( 140 self.remote_image_remote_instance._GetDeviceInfoFromResponse( 141 ONE_LINE_LEASE_RESPONSE), 142 (expect_session_id, expect_server_url)) 143 144 self.assertEqual( 145 self.remote_image_remote_instance._GetDeviceInfoFromResponse( 146 MULTIPLE_LINES_LEASE_RESPONSE), 147 (expect_session_id, expect_server_url)) 148 149 150if __name__ == '__main__': 151 unittest.main() 152