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 create_report = mock.Mock() 67 create_report.status = report.Status.SUCCESS 68 self.Patch(common_operations, "CreateDevices", 69 return_value=create_report) 70 self.remote_image_remote_instance._CreateAVD( 71 avd_spec, no_prompts=True) 72 mock_factory.assert_called_once() 73 mock_launch_browser.assert_called_once() 74 75 # Test launch VNC case. 76 avd_spec.connect_webrtc = False 77 avd_spec.connect_vnc = True 78 self.remote_image_remote_instance._CreateAVD( 79 avd_spec, no_prompts=True) 80 mock_launch_vnc.assert_called_once() 81 82 # Test launch with Oxgen case. 83 avd_spec.oxygen = True 84 self.remote_image_remote_instance._CreateAVD( 85 avd_spec, no_prompts=True) 86 mock_lease.assert_called_once() 87 88 def testLeaseOxygenAVD(self): 89 """test LeaseOxygenAVD.""" 90 avd_spec = mock.Mock() 91 avd_spec.oxygen = True 92 avd_spec.remote_image = {constants.BUILD_TARGET: "fake_target", 93 constants.BUILD_ID: "fake_id", 94 constants.BUILD_BRANCH: "fake_branch"} 95 avd_spec.system_build_info = {constants.BUILD_TARGET: "fake_target", 96 constants.BUILD_ID: "fake_id", 97 constants.BUILD_BRANCH: "fake_branch"} 98 avd_spec.kernel_build_info = {constants.BUILD_TARGET: "fake_target", 99 constants.BUILD_ID: "fake_id", 100 constants.BUILD_BRANCH: "fake_branch"} 101 response_fail = "Lease device fail." 102 self.Patch(oxygen_client.OxygenClient, "LeaseDevice", 103 side_effect=[ONE_LINE_LEASE_RESPONSE, response_fail]) 104 expected_status = report.Status.SUCCESS 105 reporter = self.remote_image_remote_instance._LeaseOxygenAVD(avd_spec) 106 self.assertEqual(reporter.status, expected_status) 107 108 expected_status = report.Status.FAIL 109 reporter = self.remote_image_remote_instance._LeaseOxygenAVD(avd_spec) 110 self.assertEqual(reporter.status, expected_status) 111 112 def testOxygenLeaseFailure(self): 113 """test LeaseOxygenAVD when the lease call failed.""" 114 avd_spec = mock.Mock() 115 avd_spec.oxygen = True 116 avd_spec.remote_image = {constants.BUILD_TARGET: "fake_target", 117 constants.BUILD_ID: "fake_id", 118 constants.BUILD_BRANCH: "fake_branch"} 119 avd_spec.system_build_info = {constants.BUILD_TARGET: "fake_target", 120 constants.BUILD_ID: "fake_id", 121 constants.BUILD_BRANCH: "fake_branch"} 122 avd_spec.kernel_build_info = {constants.BUILD_TARGET: "fake_target", 123 constants.BUILD_ID: "fake_id", 124 constants.BUILD_BRANCH: "fake_branch"} 125 response_fail = "Lease device fail." 126 self.Patch(oxygen_client.OxygenClient, "LeaseDevice", 127 side_effect=[LEASE_FAILURE_RESPONSE, response_fail]) 128 expected_status = report.Status.FAIL 129 reporter = self.remote_image_remote_instance._LeaseOxygenAVD(avd_spec) 130 self.assertEqual(reporter.status, expected_status) 131 self.assertEqual(reporter.error_type, constants.ACLOUD_OXYGEN_LEASE_ERROR) 132 self.assertEqual(reporter.errors, ["rpc error: details"]) 133 134 def testGetDeviceInfoFromResponse(self): 135 """test GetDeviceInfoFromResponse.""" 136 expect_session_id = "fake_device" 137 expect_server_url = "10.1.1.1" 138 self.assertEqual( 139 self.remote_image_remote_instance._GetDeviceInfoFromResponse( 140 ONE_LINE_LEASE_RESPONSE), 141 (expect_session_id, expect_server_url)) 142 143 self.assertEqual( 144 self.remote_image_remote_instance._GetDeviceInfoFromResponse( 145 MULTIPLE_LINES_LEASE_RESPONSE), 146 (expect_session_id, expect_server_url)) 147 148 149if __name__ == '__main__': 150 unittest.main() 151