Add unittests.
Change-Id: I55e7397c44605719bb01aa1b6d81eed2fa743cf4
diff --git a/internal/lib/gcompute_client_test.py b/internal/lib/gcompute_client_test.py
new file mode 100644
index 0000000..2382eb1
--- /dev/null
+++ b/internal/lib/gcompute_client_test.py
@@ -0,0 +1,730 @@
+#!/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()