#!/usr/bin/env python # # Copyright 2016 - 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.internal.lib.gcompute_client.""" import os import apiclient.http import mock import unittest from acloud.internal.lib import driver_test_lib from acloud.internal.lib import gcompute_client from acloud.internal.lib import utils from acloud.public import errors class ComputeClientTest(driver_test_lib.BaseDriverTest): """Test ComputeClient.""" PROJECT = "fake-project" INSTANCE = "fake-instance" IMAGE = "fake-image" IMAGE_URL = "http://fake-image-url" GS_IMAGE_SOURCE_URI = "https://storage.googleapis.com/fake-bucket/fake.tar.gz" MACHINE_TYPE = "fake-machine-type" MACHINE_TYPE_URL = "http://fake-machine-type-url" METADATA = ("metadata_key", "metadata_value") NETWORK = "fake-network" NETWORK_URL = "http://fake-network-url" ZONE = "fake-zone" REGION = "fake-region" OPERATION_NAME = "fake-op" def setUp(self): """Set up test.""" super(ComputeClientTest, self).setUp() self.Patch(gcompute_client.ComputeClient, "InitResourceHandle") fake_cfg = mock.MagicMock() fake_cfg.project = self.PROJECT self.compute_client = gcompute_client.ComputeClient(fake_cfg, mock.MagicMock()) self.compute_client._service = mock.MagicMock() def _SetupMocksForGetOperationStatus(self, mock_result, operation_scope): """A helper class for setting up mocks for testGetOperationStatus*. Args: mock_result: The result to return by _GetOperationStatus. operation_scope: A value of OperationScope. Returns: A mock for Resource object. """ resource_mock = mock.MagicMock() mock_api = mock.MagicMock() if operation_scope == gcompute_client.OperationScope.GLOBAL: self.compute_client._service.globalOperations = mock.MagicMock( return_value=resource_mock) elif operation_scope == gcompute_client.OperationScope.ZONE: self.compute_client._service.zoneOperations = mock.MagicMock( return_value=resource_mock) elif operation_scope == gcompute_client.OperationScope.REGION: self.compute_client._service.regionOperations = mock.MagicMock( return_value=resource_mock) resource_mock.get = mock.MagicMock(return_value=mock_api) mock_api.execute = mock.MagicMock(return_value=mock_result) return resource_mock def testGetOperationStatusGlobal(self): """Test _GetOperationStatus for global.""" resource_mock = self._SetupMocksForGetOperationStatus( {"status": "GOOD"}, gcompute_client.OperationScope.GLOBAL) status = self.compute_client._GetOperationStatus( {"name": self.OPERATION_NAME}, gcompute_client.OperationScope.GLOBAL) self.assertEqual(status, "GOOD") resource_mock.get.assert_called_with( project=self.PROJECT, operation=self.OPERATION_NAME) def testGetOperationStatusZone(self): """Test _GetOperationStatus for zone.""" resource_mock = self._SetupMocksForGetOperationStatus( {"status": "GOOD"}, gcompute_client.OperationScope.ZONE) status = self.compute_client._GetOperationStatus( {"name": self.OPERATION_NAME}, gcompute_client.OperationScope.ZONE, self.ZONE) self.assertEqual(status, "GOOD") resource_mock.get.assert_called_with( project=self.PROJECT, operation=self.OPERATION_NAME, zone=self.ZONE) def testGetOperationStatusRegion(self): """Test _GetOperationStatus for region.""" resource_mock = self._SetupMocksForGetOperationStatus( {"status": "GOOD"}, gcompute_client.OperationScope.REGION) self.compute_client._GetOperationStatus( {"name": self.OPERATION_NAME}, gcompute_client.OperationScope.REGION, self.REGION) resource_mock.get.assert_called_with( project=self.PROJECT, operation=self.OPERATION_NAME, region=self.REGION) def testGetOperationStatusError(self): """Test _GetOperationStatus failed.""" self._SetupMocksForGetOperationStatus( {"error": {"errors": ["error1", "error2"]}}, gcompute_client.OperationScope.GLOBAL) self.assertRaisesRegexp(errors.DriverError, "Get operation state failed.*error1.*error2", self.compute_client._GetOperationStatus, {"name": self.OPERATION_NAME}, gcompute_client.OperationScope.GLOBAL) def testWaitOnOperation(self): """Test WaitOnOperation.""" mock_error = mock.MagicMock() self.Patch(utils, "PollAndWait") self.Patch(errors, "GceOperationTimeoutError", return_value=mock_error) self.compute_client.WaitOnOperation( operation={"name": self.OPERATION_NAME}, operation_scope=gcompute_client.OperationScope.REGION, scope_name=self.REGION) utils.PollAndWait.assert_called_with( func=self.compute_client._GetOperationStatus, expected_return="DONE", timeout_exception=mock_error, timeout_secs=self.compute_client.OPERATION_TIMEOUT_SECS, sleep_interval_secs=self.compute_client. OPERATION_POLL_INTERVAL_SECS, operation={"name": self.OPERATION_NAME}, operation_scope=gcompute_client.OperationScope.REGION, scope_name=self.REGION) def testCreateImage(self): """Test CreateImage.""" self.Patch(gcompute_client.ComputeClient, "WaitOnOperation") resource_mock = mock.MagicMock() self.compute_client._service.images = mock.MagicMock( return_value=resource_mock) resource_mock.insert = mock.MagicMock() expected_body = { "name": self.IMAGE, "rawDisk": { "source": self.GS_IMAGE_SOURCE_URI, }, } self.compute_client.CreateImage( image_name=self.IMAGE, source_uri=self.GS_IMAGE_SOURCE_URI) resource_mock.insert.assert_called_with( project=self.PROJECT, body=expected_body) self.compute_client.WaitOnOperation.assert_called_with( operation=mock.ANY, operation_scope=gcompute_client.OperationScope.GLOBAL) def testCreateImageFail(self): """Test CreateImage fails.""" self.Patch( gcompute_client.ComputeClient, "WaitOnOperation", side_effect=errors.DriverError("Expected fake error")) self.Patch( gcompute_client.ComputeClient, "CheckImageExists", return_value=True) self.Patch(gcompute_client.ComputeClient, "DeleteImage") resource_mock = mock.MagicMock() self.compute_client._service.images = mock.MagicMock( return_value=resource_mock) resource_mock.insert = mock.MagicMock() expected_body = { "name": self.IMAGE, "rawDisk": { "source": self.GS_IMAGE_SOURCE_URI, }, } self.assertRaisesRegexp( errors.DriverError, "Expected fake error", self.compute_client.CreateImage, image_name=self.IMAGE, source_uri=self.GS_IMAGE_SOURCE_URI) resource_mock.insert.assert_called_with( project=self.PROJECT, body=expected_body) self.compute_client.WaitOnOperation.assert_called_with( operation=mock.ANY, operation_scope=gcompute_client.OperationScope.GLOBAL) self.compute_client.CheckImageExists.assert_called_with(self.IMAGE) self.compute_client.DeleteImage.assert_called_with(self.IMAGE) def testCheckImageExistsTrue(self): """Test CheckImageExists return True.""" resource_mock = mock.MagicMock() mock_api = mock.MagicMock() self.compute_client._service.images = mock.MagicMock( return_value=resource_mock) resource_mock.get = mock.MagicMock(return_value=mock_api) mock_api.execute = mock.MagicMock(return_value={"name": self.IMAGE}) self.assertTrue(self.compute_client.CheckImageExists(self.IMAGE)) def testCheckImageExistsFalse(self): """Test CheckImageExists return False.""" resource_mock = mock.MagicMock() mock_api = mock.MagicMock() self.compute_client._service.images = mock.MagicMock( return_value=resource_mock) resource_mock.get = mock.MagicMock(return_value=mock_api) mock_api.execute = mock.MagicMock( side_effect=errors.ResourceNotFoundError(404, "no image")) self.assertFalse(self.compute_client.CheckImageExists(self.IMAGE)) def testDeleteImage(self): """Test DeleteImage.""" self.Patch(gcompute_client.ComputeClient, "WaitOnOperation") resource_mock = mock.MagicMock() self.compute_client._service.images = mock.MagicMock( return_value=resource_mock) resource_mock.delete = mock.MagicMock() self.compute_client.DeleteImage(self.IMAGE) resource_mock.delete.assert_called_with( project=self.PROJECT, image=self.IMAGE) self.assertTrue(self.compute_client.WaitOnOperation.called) def _SetupBatchHttpRequestMock(self): """Setup BatchHttpRequest mock.""" requests = {} def _Add(request, callback, request_id): requests[request_id] = (request, callback) def _Execute(): for rid in requests: _, callback = requests[rid] callback( request_id=rid, response=mock.MagicMock(), exception=None) mock_batch = mock.MagicMock() mock_batch.add = _Add mock_batch.execute = _Execute self.Patch(apiclient.http, "BatchHttpRequest", return_value=mock_batch) def testDeleteImages(self): """Test DeleteImages.""" self._SetupBatchHttpRequestMock() self.Patch(gcompute_client.ComputeClient, "WaitOnOperation") fake_images = ["fake_image_1", "fake_image_2"] mock_api = mock.MagicMock() resource_mock = mock.MagicMock() self.compute_client._service.images = mock.MagicMock( return_value=resource_mock) resource_mock.delete = mock.MagicMock(return_value=mock_api) # Call the API. deleted, failed, error_msgs = self.compute_client.DeleteImages( fake_images) # Verify calls = [mock.call( project=self.PROJECT, image="fake_image_1"), mock.call( project=self.PROJECT, image="fake_image_2")] resource_mock.delete.assert_has_calls(calls, any_order=True) self.assertEqual( gcompute_client.ComputeClient.WaitOnOperation.call_count, 2) self.assertEqual(error_msgs, []) self.assertEqual(failed, []) self.assertEqual(set(deleted), set(fake_images)) def testListImages(self): """Test ListImages.""" fake_token = "fake_next_page_token" image_1 = "image_1" image_2 = "image_2" response_1 = {"items": [image_1], "nextPageToken": fake_token} response_2 = {"items": [image_2]} self.Patch( gcompute_client.ComputeClient, "Execute", side_effect=[response_1, response_2]) resource_mock = mock.MagicMock() self.compute_client._service.images = mock.MagicMock( return_value=resource_mock) resource_mock.list = mock.MagicMock() images = self.compute_client.ListImages() calls = [ mock.call( project=self.PROJECT, filter=None, pageToken=None), mock.call( project=self.PROJECT, filter=None, pageToken=fake_token) ] resource_mock.list.assert_has_calls(calls) self.assertEqual(images, [image_1, image_2]) def testGetInstance(self): """Test GetInstance.""" resource_mock = mock.MagicMock() mock_api = mock.MagicMock() self.compute_client._service.instances = mock.MagicMock( return_value=resource_mock) resource_mock.get = mock.MagicMock(return_value=mock_api) mock_api.execute = mock.MagicMock(return_value={"name": self.INSTANCE}) result = self.compute_client.GetInstance(self.INSTANCE, self.ZONE) self.assertEqual(result, {"name": self.INSTANCE}) resource_mock.get.assert_called_with( project=self.PROJECT, zone=self.ZONE, instance=self.INSTANCE) def testListInstances(self): """Test ListInstances.""" fake_token = "fake_next_page_token" instance_1 = "instance_1" instance_2 = "instance_2" response_1 = {"items": [instance_1], "nextPageToken": fake_token} response_2 = {"items": [instance_2]} self.Patch( gcompute_client.ComputeClient, "Execute", side_effect=[response_1, response_2]) resource_mock = mock.MagicMock() self.compute_client._service.instances = mock.MagicMock( return_value=resource_mock) resource_mock.list = mock.MagicMock() instances = self.compute_client.ListInstances(self.ZONE) calls = [ mock.call( project=self.PROJECT, zone=self.ZONE, filter=None, pageToken=None), mock.call( project=self.PROJECT, zone=self.ZONE, filter=None, pageToken=fake_token), ] resource_mock.list.assert_has_calls(calls) self.assertEqual(instances, [instance_1, instance_2]) def testCreateInstance(self): """Test CreateInstance.""" self.Patch(gcompute_client.ComputeClient, "WaitOnOperation") self.Patch( gcompute_client.ComputeClient, "GetMachineType", return_value={"selfLink": self.MACHINE_TYPE_URL}) self.Patch( gcompute_client.ComputeClient, "GetNetworkUrl", return_value=self.NETWORK_URL) self.Patch( gcompute_client.ComputeClient, "GetImage", return_value={"selfLink": self.IMAGE_URL}) resource_mock = mock.MagicMock() self.compute_client._service.instances = mock.MagicMock( return_value=resource_mock) resource_mock.insert = mock.MagicMock() expected_body = { "machineType": self.MACHINE_TYPE_URL, "name": self.INSTANCE, "networkInterfaces": [ { "network": self.NETWORK_URL, "accessConfigs": [ {"name": "External NAT", "type": "ONE_TO_ONE_NAT"} ], } ], "disks": [ { "type": "PERSISTENT", "boot": True, "mode": "READ_WRITE", "autoDelete": True, "initializeParams": { "diskName": self.INSTANCE, "sourceImage": self.IMAGE_URL, }, } ], "serviceAccounts": [ {"email": "default", "scopes": self.compute_client.DEFAULT_INSTANCE_SCOPE} ], "metadata": { "items": [{"key": self.METADATA[0], "value": self.METADATA[1]}], }, } self.compute_client.CreateInstance( instance=self.INSTANCE, image_name=self.IMAGE, machine_type=self.MACHINE_TYPE, metadata={self.METADATA[0]: self.METADATA[1]}, network=self.NETWORK, zone=self.ZONE) resource_mock.insert.assert_called_with( project=self.PROJECT, zone=self.ZONE, body=expected_body) self.compute_client.WaitOnOperation.assert_called_with( mock.ANY, operation_scope=gcompute_client.OperationScope.ZONE, scope_name=self.ZONE) def testDeleteInstance(self): """Test DeleteInstance.""" self.Patch(gcompute_client.ComputeClient, "WaitOnOperation") resource_mock = mock.MagicMock() self.compute_client._service.instances = mock.MagicMock( return_value=resource_mock) resource_mock.delete = mock.MagicMock() self.compute_client.DeleteInstance( instance=self.INSTANCE, zone=self.ZONE) resource_mock.delete.assert_called_with( project=self.PROJECT, zone=self.ZONE, instance=self.INSTANCE) self.compute_client.WaitOnOperation.assert_called_with( mock.ANY, operation_scope=gcompute_client.OperationScope.ZONE, scope_name=self.ZONE) def testDeleteInstances(self): """Test DeleteInstances.""" self._SetupBatchHttpRequestMock() self.Patch(gcompute_client.ComputeClient, "WaitOnOperation") fake_instances = ["fake_instance_1", "fake_instance_2"] mock_api = mock.MagicMock() resource_mock = mock.MagicMock() self.compute_client._service.instances = mock.MagicMock( return_value=resource_mock) resource_mock.delete = mock.MagicMock(return_value=mock_api) deleted, failed, error_msgs = self.compute_client.DeleteInstances( fake_instances, self.ZONE) calls = [ mock.call( project=self.PROJECT, instance="fake_instance_1", zone=self.ZONE), mock.call( project=self.PROJECT, instance="fake_instance_2", zone=self.ZONE) ] resource_mock.delete.assert_has_calls(calls, any_order=True) self.assertEqual( gcompute_client.ComputeClient.WaitOnOperation.call_count, 2) self.assertEqual(error_msgs, []) self.assertEqual(failed, []) self.assertEqual(set(deleted), set(fake_instances)) def testBatchExecuteOnInstances(self): self._SetupBatchHttpRequestMock() self.Patch(gcompute_client.ComputeClient, "WaitOnOperation") action = mock.MagicMock(return_value=mock.MagicMock()) fake_instances = ["fake_instance_1", "fake_instance_2"] done, failed, error_msgs = self.compute_client._BatchExecuteOnInstances( fake_instances, self.ZONE, action) calls = [mock.call(instance="fake_instance_1"), mock.call(instance="fake_instance_2")] action.assert_has_calls(calls, any_order=True) self.assertEqual( gcompute_client.ComputeClient.WaitOnOperation.call_count, 2) self.assertEqual(set(done), set(fake_instances)) self.assertEqual(error_msgs, []) self.assertEqual(failed, []) def testResetInstance(self): """Test ResetInstance.""" self.Patch(gcompute_client.ComputeClient, "WaitOnOperation") resource_mock = mock.MagicMock() self.compute_client._service.instances = mock.MagicMock( return_value=resource_mock) resource_mock.reset = mock.MagicMock() self.compute_client.ResetInstance( instance=self.INSTANCE, zone=self.ZONE) resource_mock.reset.assert_called_with( project=self.PROJECT, zone=self.ZONE, instance=self.INSTANCE) self.compute_client.WaitOnOperation.assert_called_with( mock.ANY, operation_scope=gcompute_client.OperationScope.ZONE, scope_name=self.ZONE) def _CompareMachineSizeTestHelper(self, machine_info_1, machine_info_2, expected_result=None, expected_error_type=None): """Helper class for testing CompareMachineSize. Args: machine_info_1: A dictionary representing the first machine size. machine_info_2: A dictionary representing the second machine size. expected_result: An integer, 0, 1 or -1, or None if not set. expected_error_type: An exception type, if set will check for exception. """ self.Patch( gcompute_client.ComputeClient, "GetMachineType", side_effect=[machine_info_1, machine_info_2]) if expected_error_type: self.assertRaises(expected_error_type, self.compute_client.CompareMachineSize, "name1", "name2", self.ZONE) else: result = self.compute_client.CompareMachineSize("name1", "name2", self.ZONE) self.assertEqual(result, expected_result) self.compute_client.GetMachineType.assert_has_calls( [mock.call("name1", self.ZONE), mock.call("name2", self.ZONE)]) def testCompareMachineSizeSmall(self): """Test CompareMachineSize where the first one is smaller.""" machine_info_1 = {"guestCpus": 10, "memoryMb": 100} machine_info_2 = {"guestCpus": 10, "memoryMb": 200} self._CompareMachineSizeTestHelper(machine_info_1, machine_info_2, -1) def testCompareMachineSizeLarge(self): """Test CompareMachineSize where the first one is larger.""" machine_info_1 = {"guestCpus": 10, "memoryMb": 200} machine_info_2 = {"guestCpus": 10, "memoryMb": 100} self._CompareMachineSizeTestHelper(machine_info_1, machine_info_2, 1) def testCompareMachineSizeEqual(self): """Test CompareMachineSize where two machine sizes are equal.""" machine_info = {"guestCpus": 10, "memoryMb": 100} self._CompareMachineSizeTestHelper(machine_info, machine_info, 0) def testCompareMachineSizeBadMetric(self): """Test CompareMachineSize with bad metric.""" machine_info = {"unkown_metric": 10, "memoryMb": 100} self._CompareMachineSizeTestHelper( machine_info, machine_info, expected_error_type=errors.DriverError) def testGetMachineType(self): """Test GetMachineType.""" resource_mock = mock.MagicMock() mock_api = mock.MagicMock() self.compute_client._service.machineTypes = mock.MagicMock( return_value=resource_mock) resource_mock.get = mock.MagicMock(return_value=mock_api) mock_api.execute = mock.MagicMock( return_value={"name": self.MACHINE_TYPE}) result = self.compute_client.GetMachineType(self.MACHINE_TYPE, self.ZONE) self.assertEqual(result, {"name": self.MACHINE_TYPE}) resource_mock.get.assert_called_with( project=self.PROJECT, zone=self.ZONE, machineType=self.MACHINE_TYPE) def _GetSerialPortOutputTestHelper(self, response): """Helper function for testing GetSerialPortOutput. Args: response: A dictionary representing a fake response. """ resource_mock = mock.MagicMock() mock_api = mock.MagicMock() self.compute_client._service.instances = mock.MagicMock( return_value=resource_mock) resource_mock.getSerialPortOutput = mock.MagicMock( return_value=mock_api) mock_api.execute = mock.MagicMock(return_value=response) if "contents" in response: result = self.compute_client.GetSerialPortOutput( instance=self.INSTANCE, zone=self.ZONE) self.assertEqual(result, "fake contents") else: self.assertRaisesRegexp( errors.DriverError, "Malformed response.*", self.compute_client.GetSerialPortOutput, instance=self.INSTANCE, zone=self.ZONE) resource_mock.getSerialPortOutput.assert_called_with( project=self.PROJECT, zone=self.ZONE, instance=self.INSTANCE, port=1) def testGetSerialPortOutput(self): response = {"contents": "fake contents"} self._GetSerialPortOutputTestHelper(response) def testGetSerialPortOutputFail(self): response = {"malformed": "fake contents"} self._GetSerialPortOutputTestHelper(response) def testGetInstanceNamesByIPs(self): """Test GetInstanceNamesByIPs.""" good_instance = { "name": "instance_1", "networkInterfaces": [ { "accessConfigs": [ {"natIP": "172.22.22.22"}, ], }, ], } bad_instance = {"name": "instance_2"} self.Patch( gcompute_client.ComputeClient, "ListInstances", return_value=[good_instance, bad_instance]) ip_name_map = self.compute_client.GetInstanceNamesByIPs( ips=["172.22.22.22", "172.22.22.23"], zone=self.ZONE) self.assertEqual(ip_name_map, {"172.22.22.22": "instance_1", "172.22.22.23": None}) def testAddSshRsa(self): """Test AddSshRsa..""" fake_user = "fake_user" sshkey = ( "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQDBkTOTRze9v2VOqkkf7RG" "jSkg6Z2kb9Q9UHsDGatvend3fmjIw1Tugg0O7nnjlPkskmlgyd4a/j99WOeLL" "CPk6xPyoVjrPUVBU/pAk09ORTC4Zqk6YjlW7LOfzvqmXhmIZfYu6Q4Yt50pZzhl" "lllfu26nYjY7Tg12D019nJi/kqPX5+NKgt0LGXTu8T1r2Gav/q4V7QRWQrB8Eiu" "pxXR7I2YhynqovkEt/OXG4qWgvLEXGsWtSQs0CtCzqEVxz0Y9ECr7er4VdjSQxV" "AaeLAsQsK9ROae8hMBFZ3//8zLVapBwpuffCu+fUoql9qeV9xagZcc9zj8XOUOW" "ApiihqNL1111 test@test1.org") project = { "commonInstanceMetadata": { "kind": "compute#metadata", "fingerprint": "a-23icsyx4E=", "items": [ { "key": "sshKeys", "value": "user:key" } ] } } expected = { "kind": "compute#metadata", "fingerprint": "a-23icsyx4E=", "items": [ { "key": "sshKeys", "value": "user:key\n%s:%s" % (fake_user, sshkey) } ] } self.Patch(os.path, "exists", return_value=True) m = mock.mock_open(read_data=sshkey) self.Patch(__builtins__, "open", m, create=True) self.Patch(gcompute_client.ComputeClient, "WaitOnOperation") self.Patch( gcompute_client.ComputeClient, "GetProject", return_value=project) resource_mock = mock.MagicMock() self.compute_client._service.projects = mock.MagicMock( return_value=resource_mock) resource_mock.setCommonInstanceMetadata = mock.MagicMock() self.compute_client.AddSshRsa(fake_user, "/path/to/test_rsa.pub") resource_mock.setCommonInstanceMetadata.assert_called_with( project=self.PROJECT, body=expected) def testAddSshRsaInvalidKey(self): """Test AddSshRsa..""" fake_user = "fake_user" sshkey = "ssh-rsa v2VOqkkf7RGL1111 test@test1.org" project = { "commonInstanceMetadata": { "kind": "compute#metadata", "fingerprint": "a-23icsyx4E=", "items": [ { "key": "sshKeys", "value": "user:key" } ] } } self.Patch(os.path, "exists", return_value=True) m = mock.mock_open(read_data=sshkey) self.Patch(__builtins__, "open", m, create=True) self.Patch(gcompute_client.ComputeClient, "WaitOnOperation") self.Patch( gcompute_client.ComputeClient, "GetProject", return_value=project) self.assertRaisesRegexp(errors.DriverError, "rsa key is invalid:*", self.compute_client.AddSshRsa, fake_user, "/path/to/test_rsa.pub") def testDeleteDisks(self): """Test DeleteDisks.""" self._SetupBatchHttpRequestMock() self.Patch(gcompute_client.ComputeClient, "WaitOnOperation") fake_disks = ["fake_disk_1", "fake_disk_2"] mock_api = mock.MagicMock() resource_mock = mock.MagicMock() self.compute_client._service.disks = mock.MagicMock( return_value=resource_mock) resource_mock.delete = mock.MagicMock(return_value=mock_api) # Call the API. deleted, failed, error_msgs = self.compute_client.DeleteDisks( fake_disks, zone=self.ZONE) # Verify calls = [mock.call( project=self.PROJECT, disk="fake_disk_1", zone=self.ZONE), mock.call( project=self.PROJECT, disk="fake_disk_2", zone=self.ZONE)] resource_mock.delete.assert_has_calls(calls, any_order=True) self.assertEqual( gcompute_client.ComputeClient.WaitOnOperation.call_count, 2) self.assertEqual(error_msgs, []) self.assertEqual(failed, []) self.assertEqual(set(deleted), set(fake_disks)) if __name__ == "__main__": unittest.main()