#!/usr/bin/env python # # Copyright 2018 - The Android Open Source Project # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Tests for acloud.public.actions.common_operations.""" from __future__ import absolute_import from __future__ import division import unittest from unittest import mock from acloud import errors from acloud.internal import constants from acloud.internal.lib import android_build_client from acloud.internal.lib import android_compute_client from acloud.internal.lib import auth from acloud.internal.lib import driver_test_lib from acloud.internal.lib import utils from acloud.internal.lib import ssh from acloud.public import report from acloud.public.actions import common_operations class CommonOperationsTest(driver_test_lib.BaseDriverTest): """Test Common Operations.""" maxDiff = None IP = ssh.IP(external="127.0.0.1", internal="10.0.0.1") INSTANCE = "fake-instance" CMD = "test-cmd" AVD_TYPE = "fake-type" BRANCH = "fake-branch" BUILD_TARGET = "fake-target" BUILD_ID = "fake-build-id" LOGS = [{"path": "/log", "type": "TEXT"}] # pylint: disable=protected-access def setUp(self): """Set up the test.""" super().setUp() self.build_client = mock.MagicMock() self.device_factory = mock.MagicMock() self.Patch( android_build_client, "AndroidBuildClient", return_value=self.build_client) self.compute_client = mock.MagicMock() self.compute_client.gce_hostname = None self.Patch( android_compute_client, "AndroidComputeClient", return_value=self.compute_client) self.Patch(auth, "CreateCredentials", return_value=mock.MagicMock()) self.Patch(self.compute_client, "GetInstanceIP", return_value=self.IP) self.Patch( self.device_factory, "CreateInstance", return_value=self.INSTANCE) self.Patch( self.device_factory, "GetComputeClient", return_value=self.compute_client) self.Patch(self.device_factory, "GetVncPorts", return_value=[6444]) self.Patch(self.device_factory, "GetAdbPorts", return_value=[6520]) self.Patch(self.device_factory, "GetBuildInfoDict", return_value={"branch": self.BRANCH, "build_id": self.BUILD_ID, "build_target": self.BUILD_TARGET, "gcs_bucket_build_id": self.BUILD_ID}) self.Patch(self.device_factory, "GetLogs", return_value={self.INSTANCE: self.LOGS}) self.Patch( self.device_factory, "GetFetchCvdWrapperLogIfExist", return_value={}) @staticmethod def _CreateCfg(): """A helper method that creates a mock configuration object.""" cfg = mock.MagicMock() cfg.service_account_name = "fake@service.com" cfg.service_account_private_key_path = "/fake/path/to/key" cfg.zone = "fake_zone" cfg.disk_image_name = "fake_image.tar.gz" cfg.disk_image_mime_type = "fake/type" cfg.ssh_private_key_path = "cfg/private/key" cfg.ssh_public_key_path = "" cfg.extra_args_ssh_tunnel="extra args" return cfg def testDevicePoolCreateDevices(self): """Test Device Pool Create Devices.""" pool = common_operations.DevicePool(self.device_factory) pool.CreateDevices(5) self.assertEqual(self.device_factory.CreateInstance.call_count, 5) self.assertEqual(len(pool.devices), 5) def testCreateDevices(self): """Test Create Devices.""" cfg = self._CreateCfg() _report = common_operations.CreateDevices(self.CMD, cfg, self.device_factory, 1, self.AVD_TYPE) self.assertEqual(_report.command, self.CMD) self.assertEqual(_report.status, report.Status.SUCCESS) self.assertEqual( _report.data, {"devices": [{ "ip": self.IP.external + ":6520", "instance_name": self.INSTANCE, "branch": self.BRANCH, "build_id": self.BUILD_ID, "build_target": self.BUILD_TARGET, "gcs_bucket_build_id": self.BUILD_ID, "logs": self.LOGS }]}) def testCreateDevicesWithAdbPort(self): """Test Create Devices with adb port for cuttlefish avd type.""" forwarded_ports = mock.Mock(adb_port=12345, vnc_port=56789) mock_auto_connect = self.Patch(utils, "AutoConnect", return_value=forwarded_ports) cfg = self._CreateCfg() _report = common_operations.CreateDevices(self.CMD, cfg, self.device_factory, 1, "cuttlefish", autoconnect=True, client_adb_port=12345) mock_auto_connect.assert_called_with( ip_addr="127.0.0.1", rsa_key_file="cfg/private/key", target_vnc_port=6444, target_adb_port=6520, ssh_user=constants.GCE_USER, client_adb_port=12345, extra_args_ssh_tunnel="extra args") self.assertEqual(_report.command, self.CMD) self.assertEqual(_report.status, report.Status.SUCCESS) self.assertEqual( _report.data, {"devices": [{ "ip": self.IP.external + ":6520", "instance_name": self.INSTANCE, "branch": self.BRANCH, "build_id": self.BUILD_ID, "adb_port": 12345, "device_serial": "127.0.0.1:12345", "vnc_port": 56789, "build_target": self.BUILD_TARGET, "gcs_bucket_build_id": self.BUILD_ID, "logs": self.LOGS }]}) def testCreateDevicesMultipleDevices(self): """Test Create Devices with multiple cuttlefish devices.""" forwarded_ports_1 = mock.Mock(adb_port=12345, vnc_port=56789) forwarded_ports_2 = mock.Mock(adb_port=23456, vnc_port=67890) self.Patch(self.device_factory, "GetVncPorts", return_value=[6444, 6445]) self.Patch(self.device_factory, "GetAdbPorts", return_value=[6520, 6521]) self.Patch(utils, "PickFreePort", return_value=12345) mock_auto_connect = self.Patch( utils, "AutoConnect", side_effects=[forwarded_ports_1, forwarded_ports_2]) cfg = self._CreateCfg() _report = common_operations.CreateDevices(self.CMD, cfg, self.device_factory, 1, "cuttlefish", autoconnect=True, client_adb_port=None) self.assertEqual(2, mock_auto_connect.call_count) mock_auto_connect.assert_any_call( ip_addr="127.0.0.1", rsa_key_file="cfg/private/key", target_vnc_port=6444, target_adb_port=6520, ssh_user=constants.GCE_USER, client_adb_port=None, extra_args_ssh_tunnel="extra args") mock_auto_connect.assert_any_call( ip_addr="127.0.0.1", rsa_key_file="cfg/private/key", target_vnc_port=6445, target_adb_port=6521, ssh_user=constants.GCE_USER, client_adb_port=None, extra_args_ssh_tunnel="extra args") self.assertEqual(_report.command, self.CMD) self.assertEqual(_report.status, report.Status.SUCCESS) def testCreateDevicesInternalIP(self): """Test Create Devices and report internal IP.""" cfg = self._CreateCfg() _report = common_operations.CreateDevices(self.CMD, cfg, self.device_factory, 1, self.AVD_TYPE, report_internal_ip=True) self.assertEqual(_report.command, self.CMD) self.assertEqual(_report.status, report.Status.SUCCESS) self.assertEqual( _report.data, {"devices": [{ "ip": self.IP.internal + ":6520", "instance_name": self.INSTANCE, "branch": self.BRANCH, "build_id": self.BUILD_ID, "build_target": self.BUILD_TARGET, "gcs_bucket_build_id": self.BUILD_ID, "logs": self.LOGS }]}) def testCreateDevicesWithSshParameters(self): """Test Create Devices with ssh user and key.""" forwarded_ports = mock.Mock(adb_port=12345, vnc_port=56789) mock_auto_connect = self.Patch(utils, "AutoConnect", return_value=forwarded_ports) mock_establish_webrtc = self.Patch(utils, "EstablishWebRTCSshTunnel") self.Patch(utils, "PickFreePort", return_value=12345) cfg = self._CreateCfg() _report = common_operations.CreateDevices( self.CMD, cfg, self.device_factory, 1, constants.TYPE_CF, autoconnect=True, connect_webrtc=True, ssh_user="user", ssh_private_key_path="private/key") mock_auto_connect.assert_called_with( ip_addr="127.0.0.1", rsa_key_file="private/key", target_vnc_port=6444, target_adb_port=6520, ssh_user="user", client_adb_port=None, extra_args_ssh_tunnel="extra args") mock_establish_webrtc.assert_called_with( ip_addr="127.0.0.1", rsa_key_file="private/key", ssh_user="user", extra_args_ssh_tunnel="extra args", webrtc_local_port=12345) self.assertEqual(_report.status, report.Status.SUCCESS) def testGetErrorType(self): """Test GetErrorType.""" # Test with CheckGCEZonesQuotaError() error = errors.CheckGCEZonesQuotaError() expected_result = constants.GCE_QUOTA_ERROR self.assertEqual(common_operations._GetErrorType(error), expected_result) # Test with DownloadArtifactError() error = errors.DownloadArtifactError() expected_result = constants.ACLOUD_DOWNLOAD_ARTIFACT_ERROR self.assertEqual(common_operations._GetErrorType(error), expected_result) # Test with DeviceConnectionError() error = errors.DeviceConnectionError() expected_result = constants.ACLOUD_SSH_CONNECT_ERROR self.assertEqual(common_operations._GetErrorType(error), expected_result) # Test with ACLOUD_UNKNOWN_ERROR error = errors.DriverError() expected_result = constants.ACLOUD_UNKNOWN_ERROR self.assertEqual(common_operations._GetErrorType(error), expected_result) # Test with error message about GCE quota issue error = errors.DriverError("Quota exceeded for quota read group.") expected_result = constants.GCE_QUOTA_ERROR self.assertEqual(common_operations._GetErrorType(error), expected_result) error = errors.DriverError("ZONE_RESOURCE_POOL_EXHAUSTED_WITH_DETAILS") expected_result = constants.GCE_QUOTA_ERROR self.assertEqual(common_operations._GetErrorType(error), expected_result) def testCreateDevicesWithFetchCvdWrapper(self): """Test Create Devices with FetchCvdWrapper.""" self.Patch( self.device_factory, "GetFetchCvdWrapperLogIfExist", return_value={"fetch_log": "abc"}) cfg = self._CreateCfg() _report = common_operations.CreateDevices(self.CMD, cfg, self.device_factory, 1, constants.TYPE_CF) self.assertEqual(_report.command, self.CMD) self.assertEqual(_report.status, report.Status.SUCCESS) self.assertEqual( _report.data, {"devices": [{ "ip": self.IP.external + ":6520", "instance_name": self.INSTANCE, "branch": self.BRANCH, "build_id": self.BUILD_ID, "build_target": self.BUILD_TARGET, "gcs_bucket_build_id": self.BUILD_ID, "logs": self.LOGS, "fetch_cvd_wrapper_log": { "fetch_log": "abc" }, }]}) if __name__ == "__main__": unittest.main()