| # Copyright 2019 - 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 remote_instance_cf_device_factory.""" |
| |
| import glob |
| import os |
| import tempfile |
| import unittest |
| import uuid |
| |
| from unittest import mock |
| |
| from acloud.create import avd_spec |
| from acloud.internal import constants |
| from acloud.internal.lib import android_build_client |
| from acloud.internal.lib import auth |
| from acloud.internal.lib import cvd_compute_client_multi_stage |
| from acloud.internal.lib import driver_test_lib |
| from acloud.internal.lib import utils |
| from acloud.list import list as list_instances |
| from acloud.public.actions import remote_instance_cf_device_factory |
| |
| |
| class RemoteInstanceDeviceFactoryTest(driver_test_lib.BaseDriverTest): |
| """Test RemoteInstanceDeviceFactory method.""" |
| |
| def setUp(self): |
| """Set up the test.""" |
| super().setUp() |
| self.Patch(auth, "CreateCredentials", return_value=mock.MagicMock()) |
| self.Patch(android_build_client.AndroidBuildClient, "InitResourceHandle") |
| self.Patch(cvd_compute_client_multi_stage.CvdComputeClient, "InitResourceHandle") |
| self.Patch(cvd_compute_client_multi_stage.CvdComputeClient, "LaunchCvd") |
| self.Patch(cvd_compute_client_multi_stage.CvdComputeClient, "UpdateFetchCvd") |
| self.Patch(cvd_compute_client_multi_stage.CvdComputeClient, "FetchBuild") |
| self.Patch(list_instances, "GetInstancesFromInstanceNames", return_value=mock.MagicMock()) |
| self.Patch(list_instances, "ChooseOneRemoteInstance", return_value=mock.MagicMock()) |
| self.Patch(utils, "GetBuildEnvironmentVariable", |
| return_value="test_env_cf_arm") |
| self.Patch(glob, "glob", return_vale=["fake.img"]) |
| |
| # pylint: disable=protected-access |
| @staticmethod |
| @mock.patch.object(cvd_compute_client_multi_stage.CvdComputeClient, |
| "UpdateCertificate") |
| @mock.patch("acloud.public.actions.remote_instance_cf_device_factory." |
| "cvd_utils") |
| def testProcessArtifacts(mock_cvd_utils, mock_uploadca): |
| """test ProcessArtifacts.""" |
| # Test image source type is local. |
| args = mock.MagicMock() |
| args.config_file = "" |
| args.avd_type = constants.TYPE_CF |
| args.flavor = "phone" |
| args.local_image = constants.FIND_IN_BUILD_ENV |
| args.launch_args = None |
| args.autoconnect = constants.INS_KEY_WEBRTC |
| avd_spec_local_img = avd_spec.AVDSpec(args) |
| mock_cvd_utils.AreTargetFilesRequired.return_value = False |
| fake_image_name = "/fake/aosp_cf_x86_phone-img-eng.username.zip" |
| fake_host_package_name = "/fake/host_package.tar.gz" |
| factory_local_img = remote_instance_cf_device_factory.RemoteInstanceDeviceFactory( |
| avd_spec_local_img, |
| fake_image_name, |
| fake_host_package_name) |
| factory_local_img._ProcessArtifacts() |
| # cf default autoconnect webrtc and should upload certificates |
| mock_uploadca.assert_called_once() |
| mock_uploadca.reset_mock() |
| mock_cvd_utils.UploadArtifacts.assert_called_once_with( |
| mock.ANY, mock_cvd_utils.GCE_BASE_DIR, fake_image_name, |
| fake_host_package_name) |
| mock_cvd_utils.UploadExtraImages.assert_called_once() |
| |
| # given autoconnect to vnc should not upload certificates |
| args.autoconnect = constants.INS_KEY_VNC |
| avd_spec_local_img = avd_spec.AVDSpec(args) |
| factory_local_img = remote_instance_cf_device_factory.RemoteInstanceDeviceFactory( |
| avd_spec_local_img, |
| fake_image_name, |
| fake_host_package_name) |
| factory_local_img._ProcessArtifacts() |
| mock_uploadca.assert_not_called() |
| |
| # Test image source type is remote. |
| args.local_image = None |
| args.build_id = "1234" |
| args.branch = "fake_branch" |
| args.build_target = "fake_target" |
| args.system_build_id = "2345" |
| args.system_branch = "sys_branch" |
| args.system_build_target = "sys_target" |
| args.kernel_build_id = "3456" |
| args.kernel_branch = "kernel_branch" |
| args.kernel_build_target = "kernel_target" |
| avd_spec_remote_img = avd_spec.AVDSpec(args) |
| factory_remote_img = remote_instance_cf_device_factory.RemoteInstanceDeviceFactory( |
| avd_spec_remote_img) |
| factory_remote_img._ProcessArtifacts() |
| |
| compute_client = factory_remote_img.GetComputeClient() |
| compute_client.UpdateFetchCvd.assert_called_once() |
| compute_client.FetchBuild.assert_called_once() |
| |
| # pylint: disable=protected-access |
| @mock.patch.dict(os.environ, {constants.ENV_BUILD_TARGET:'fake-target'}) |
| def testCreateGceInstanceNameMultiStage(self): |
| """test create gce instance.""" |
| # Mock uuid |
| args = mock.MagicMock() |
| args.config_file = "" |
| args.avd_type = constants.TYPE_CF |
| args.flavor = "phone" |
| args.local_image = constants.FIND_IN_BUILD_ENV |
| args.adb_port = None |
| args.launch_args = None |
| fake_avd_spec = avd_spec.AVDSpec(args) |
| fake_avd_spec.cfg.enable_multi_stage = True |
| fake_avd_spec._instance_name_to_reuse = None |
| |
| fake_uuid = mock.MagicMock(hex="1234") |
| self.Patch(uuid, "uuid4", return_value=fake_uuid) |
| self.Patch(cvd_compute_client_multi_stage.CvdComputeClient, "CreateInstance") |
| self.Patch(cvd_compute_client_multi_stage.CvdComputeClient, |
| "GetHostImageName", return_value="fake_image") |
| fake_host_package_name = "/fake/host_package.tar.gz" |
| fake_image_name = "/fake/aosp_cf_x86_phone-img-eng.username.zip" |
| |
| factory = remote_instance_cf_device_factory.RemoteInstanceDeviceFactory( |
| fake_avd_spec, |
| fake_image_name, |
| fake_host_package_name) |
| self.assertEqual(factory.CreateGceInstance(), "ins-1234-userbuild-aosp-cf-x86-phone") |
| |
| # Can't get target name from zip file name. |
| fake_image_name = "/fake/aosp_cf_x86_phone.username.zip" |
| factory = remote_instance_cf_device_factory.RemoteInstanceDeviceFactory( |
| fake_avd_spec, |
| fake_image_name, |
| fake_host_package_name) |
| self.assertEqual(factory.CreateGceInstance(), "ins-1234-userbuild-fake-target") |
| |
| # No image zip path, it uses local build images. |
| fake_image_name = "" |
| factory = remote_instance_cf_device_factory.RemoteInstanceDeviceFactory( |
| fake_avd_spec, |
| fake_image_name, |
| fake_host_package_name) |
| self.assertEqual(factory.CreateGceInstance(), "ins-1234-userbuild-fake-target") |
| |
| def testReuseInstanceNameMultiStage(self): |
| """Test reuse instance name.""" |
| args = mock.MagicMock() |
| args.config_file = "" |
| args.avd_type = constants.TYPE_CF |
| args.flavor = "phone" |
| args.local_image = constants.FIND_IN_BUILD_ENV |
| args.adb_port = None |
| args.launch_args = None |
| fake_avd_spec = avd_spec.AVDSpec(args) |
| fake_avd_spec.cfg.enable_multi_stage = True |
| fake_avd_spec._instance_name_to_reuse = "fake-1234-userbuild-fake-target" |
| fake_uuid = mock.MagicMock(hex="1234") |
| self.Patch(uuid, "uuid4", return_value=fake_uuid) |
| self.Patch(cvd_compute_client_multi_stage.CvdComputeClient, "CreateInstance") |
| self.Patch(cvd_compute_client_multi_stage.CvdComputeClient, |
| "GetHostImageName", return_value="fake_image") |
| fake_host_package_name = "/fake/host_package.tar.gz" |
| fake_image_name = "/fake/aosp_cf_x86_phone-img-eng.username.zip" |
| factory = remote_instance_cf_device_factory.RemoteInstanceDeviceFactory( |
| fake_avd_spec, |
| fake_image_name, |
| fake_host_package_name) |
| self.assertEqual(factory.CreateGceInstance(), "fake-1234-userbuild-fake-target") |
| |
| @mock.patch("acloud.public.actions.remote_instance_cf_device_factory." |
| "cvd_utils") |
| def testGetBuildInfoDict(self, mock_cvd_utils): |
| """Test GetBuildInfoDict.""" |
| fake_host_package_name = "/fake/host_package.tar.gz" |
| fake_image_name = "/fake/aosp_cf_x86_phone-img-eng.username.zip" |
| args = mock.MagicMock() |
| # Test image source type is local. |
| args.config_file = "" |
| args.avd_type = constants.TYPE_CF |
| args.flavor = "phone" |
| args.local_image = "fake_local_image" |
| args.adb_port = None |
| args.cheeps_betty_image = None |
| args.launch_args = None |
| avd_spec_local_image = avd_spec.AVDSpec(args) |
| factory = remote_instance_cf_device_factory.RemoteInstanceDeviceFactory( |
| avd_spec_local_image, |
| fake_image_name, |
| fake_host_package_name) |
| self.assertEqual(factory.GetBuildInfoDict(), None) |
| mock_cvd_utils.GetRemoteBuildInfoDict.assert_not_called() |
| |
| # Test image source type is remote. |
| args.local_image = None |
| args.build_id = "123" |
| args.branch = "fake_branch" |
| args.build_target = "fake_target" |
| avd_spec_remote_image = avd_spec.AVDSpec(args) |
| factory = remote_instance_cf_device_factory.RemoteInstanceDeviceFactory( |
| avd_spec_remote_image, |
| fake_image_name, |
| fake_host_package_name) |
| factory.GetBuildInfoDict() |
| mock_cvd_utils.GetRemoteBuildInfoDict.assert_called() |
| |
| @mock.patch.object(remote_instance_cf_device_factory.RemoteInstanceDeviceFactory, |
| "CreateGceInstance") |
| @mock.patch("acloud.public.actions.remote_instance_cf_device_factory.pull") |
| @mock.patch("acloud.public.actions.remote_instance_cf_device_factory." |
| "cvd_utils") |
| def testLocalImageCreateInstance(self, mock_cvd_utils, mock_pull, |
| mock_create_gce_instance): |
| """Test CreateInstance with local images.""" |
| self.Patch( |
| cvd_compute_client_multi_stage, |
| "CvdComputeClient", |
| return_value=mock.MagicMock()) |
| mock_create_gce_instance.return_value = "instance" |
| fake_avd_spec = mock.MagicMock() |
| fake_avd_spec.image_source = constants.IMAGE_SRC_LOCAL |
| fake_avd_spec._instance_name_to_reuse = None |
| fake_avd_spec.no_pull_log = False |
| fake_avd_spec.base_instance_num = None |
| fake_avd_spec.num_avds_per_instance = None |
| |
| mock_cvd_utils.AreTargetFilesRequired.return_value = False |
| mock_cvd_utils.FindRemoteLogs.return_value = [{"path": "/logcat"}] |
| mock_cvd_utils.UploadExtraImages.return_value = [ |
| ("-boot_image", "/boot/img")] |
| |
| fake_host_package_name = "/fake/host_package.tar.gz" |
| fake_image_name = "" |
| factory = remote_instance_cf_device_factory.RemoteInstanceDeviceFactory( |
| fake_avd_spec, |
| fake_image_name, |
| fake_host_package_name) |
| compute_client = factory.GetComputeClient() |
| compute_client.LaunchCvd.return_value = {"instance": "failure"} |
| factory.CreateInstance() |
| mock_create_gce_instance.assert_called_once() |
| mock_cvd_utils.UploadArtifacts.assert_called_once() |
| mock_cvd_utils.FindRemoteLogs.assert_called_with( |
| mock.ANY, mock_cvd_utils.GCE_BASE_DIR, None, None) |
| compute_client.LaunchCvd.assert_called_once() |
| self.assertIn(["-boot_image", "/boot/img"], |
| compute_client.LaunchCvd.call_args[0]) |
| mock_pull.GetAllLogFilePaths.assert_called_once_with( |
| mock.ANY, constants.REMOTE_LOG_FOLDER) |
| mock_pull.PullLogs.assert_called_once() |
| |
| factory.GetAdbPorts() |
| mock_cvd_utils.GetAdbPorts.assert_called_with(None, None) |
| factory.GetVncPorts() |
| mock_cvd_utils.GetVncPorts.assert_called_with(None, None) |
| self.assertEqual({"instance": "failure"}, factory.GetFailures()) |
| self.assertEqual(2, len(factory.GetLogs().get("instance"))) |
| |
| @mock.patch("acloud.public.actions.remote_instance_cf_device_factory." |
| "cvd_utils") |
| def testExtraImageCreateInstance(self, mock_cvd_utils): |
| """Test CreateInstance with local extra images.""" |
| self.Patch(cvd_compute_client_multi_stage, "CvdComputeClient") |
| self.Patch( |
| remote_instance_cf_device_factory.RemoteInstanceDeviceFactory, |
| "CreateGceInstance", return_value="instance") |
| self.Patch( |
| remote_instance_cf_device_factory.RemoteInstanceDeviceFactory, |
| "_FindLogFiles") |
| |
| mock_cvd_utils.AreTargetFilesRequired.return_value = True |
| |
| with tempfile.TemporaryDirectory() as temp_dir: |
| args = mock.MagicMock() |
| args.config_file = "" |
| args.avd_type = constants.TYPE_CF |
| args.flavor = "phone" |
| args.local_image = temp_dir |
| args.launch_args = None |
| args.no_pull_log = True |
| avd_spec_local_img = avd_spec.AVDSpec(args) |
| factory_local_img = remote_instance_cf_device_factory.RemoteInstanceDeviceFactory( |
| avd_spec_local_img) |
| compute_client = factory_local_img.GetComputeClient() |
| compute_client.LaunchCvd.return_value = {} |
| |
| factory_local_img.CreateInstance() |
| |
| mock_cvd_utils.UploadArtifacts.assert_called_once_with( |
| mock.ANY, mock_cvd_utils.GCE_BASE_DIR, temp_dir, mock.ANY) |
| mock_cvd_utils.UploadExtraImages.assert_called_once_with( |
| mock.ANY, mock_cvd_utils.GCE_BASE_DIR, avd_spec_local_img, |
| args.local_image) |
| |
| @mock.patch.object(remote_instance_cf_device_factory.RemoteInstanceDeviceFactory, |
| "CreateGceInstance") |
| @mock.patch("acloud.public.actions.remote_instance_cf_device_factory.pull") |
| @mock.patch("acloud.public.actions.remote_instance_cf_device_factory." |
| "cvd_utils") |
| def testRemoteImageCreateInstance(self, mock_cvd_utils, mock_pull, |
| mock_create_gce_instance): |
| """Test CreateInstance with remote images.""" |
| self.Patch( |
| cvd_compute_client_multi_stage, |
| "CvdComputeClient", |
| return_value=mock.MagicMock()) |
| mock_create_gce_instance.return_value = "instance" |
| fake_avd_spec = mock.MagicMock() |
| fake_avd_spec.image_source = constants.IMAGE_SRC_REMOTE |
| fake_avd_spec.host_user = None |
| fake_avd_spec.no_pull_log = True |
| fake_avd_spec.base_instance_num = 2 |
| fake_avd_spec.num_avds_per_instance = 3 |
| |
| mock_cvd_utils.AreTargetFilesRequired.return_value = False |
| mock_cvd_utils.FindRemoteLogs.return_value = [{"path": "/logcat"}] |
| mock_cvd_utils.UploadExtraImages.return_value = [] |
| |
| factory = remote_instance_cf_device_factory.RemoteInstanceDeviceFactory( |
| fake_avd_spec) |
| compute_client = factory.GetComputeClient() |
| compute_client.LaunchCvd.return_value = {} |
| factory.CreateInstance() |
| |
| compute_client.FetchBuild.assert_called_once() |
| mock_cvd_utils.FindRemoteLogs.assert_called_with( |
| mock.ANY, mock_cvd_utils.GCE_BASE_DIR, 2, 3) |
| mock_pull.GetAllLogFilePaths.assert_not_called() |
| mock_pull.PullLogs.assert_not_called() |
| |
| factory.GetAdbPorts() |
| mock_cvd_utils.GetAdbPorts.assert_called_with(2, 3) |
| factory.GetVncPorts() |
| mock_cvd_utils.GetVncPorts.assert_called_with(2, 3) |
| self.assertFalse(factory.GetFailures()) |
| self.assertEqual(3, len(factory.GetLogs().get("instance"))) |
| |
| @mock.patch("acloud.public.actions.remote_instance_cf_device_factory." |
| "cvd_utils") |
| def testGetOpenWrtInfoDict(self, mock_cvd_utils): |
| """Test GetOpenWrtInfoDict.""" |
| self.Patch(cvd_compute_client_multi_stage.CvdComputeClient, |
| "GetSshConnectCmd", return_value="fake_ssh_command") |
| args = mock.MagicMock() |
| args.config_file = "" |
| args.avd_type = constants.TYPE_CF |
| args.flavor = "phone" |
| args.local_image = "fake_local_image" |
| args.launch_args = None |
| args.openwrt = False |
| avd_spec_no_openwrt = avd_spec.AVDSpec(args) |
| factory = remote_instance_cf_device_factory.RemoteInstanceDeviceFactory( |
| avd_spec_no_openwrt) |
| self.assertIsNone(factory.GetOpenWrtInfoDict()) |
| |
| args.openwrt = True |
| avd_spec_openwrt = avd_spec.AVDSpec(args) |
| factory = remote_instance_cf_device_factory.RemoteInstanceDeviceFactory( |
| avd_spec_openwrt) |
| self.assertIsNotNone(factory.GetOpenWrtInfoDict()) |
| mock_cvd_utils.GetOpenWrtInfoDict.assert_called() |
| |
| |
| if __name__ == "__main__": |
| unittest.main() |
