| # Licensed to the Apache Software Foundation (ASF) under one or more |
| # contributor license agreements. See the NOTICE file distributed with |
| # this work for additional information regarding copyright ownership. |
| # The ASF licenses this file to You 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 Google Compute Engine Driver |
| """ |
| |
| import sys |
| import datetime |
| import unittest |
| from unittest import mock |
| |
| from libcloud.test import MockHttp |
| from libcloud.utils.py3 import httplib |
| from libcloud.compute.base import Node, StorageVolume |
| from libcloud.test.compute import TestCaseMixin |
| from libcloud.test.secrets import GCE_PARAMS, GCE_KEYWORD_PARAMS |
| from libcloud.common.google import ( |
| GoogleBaseError, |
| ResourceExistsError, |
| ResourceNotFoundError, |
| GoogleBaseAuthConnection, |
| ) |
| from libcloud.compute.types import NodeState |
| from libcloud.test.file_fixtures import ComputeFileFixtures |
| from libcloud.compute.drivers.gce import ( |
| API_VERSION, |
| GCEZone, |
| GCERoute, |
| GCERegion, |
| GCEUrlMap, |
| GCEAddress, |
| GCEBackend, |
| GCENetwork, |
| GCEProject, |
| GCEFirewall, |
| GCENodeImage, |
| GCENodeDriver, |
| GCESubnetwork, |
| GCEHealthCheck, |
| GCEBackendService, |
| GCEForwardingRule, |
| GCETargetHttpProxy, |
| timestamp_to_datetime, |
| ) |
| from libcloud.test.common.test_google import GoogleTestCase, GoogleAuthMockHttp |
| |
| |
| class GCENodeDriverTest(GoogleTestCase, TestCaseMixin): |
| """ |
| Google Compute Engine Test Class. |
| """ |
| |
| # Mock out a few specific calls that interact with the user, system or |
| # environment. |
| GCEZone._now = lambda x: datetime.datetime(2013, 6, 26, 19, 0, 0) |
| datacenter = "us-central1-a" |
| |
| def setUp(self): |
| GCEMockHttp.test = self |
| GCENodeDriver.connectionCls.conn_class = GCEMockHttp |
| GoogleBaseAuthConnection.conn_class = GoogleAuthMockHttp |
| GCEMockHttp.type = None |
| kwargs = GCE_KEYWORD_PARAMS.copy() |
| kwargs["auth_type"] = "IA" |
| kwargs["datacenter"] = self.datacenter |
| self.driver = GCENodeDriver(*GCE_PARAMS, **kwargs) |
| |
| def test_default_scopes(self): |
| self.assertIsNone(self.driver.scopes) |
| |
| def test_build_service_account_gce_struct_default_service_account(self): |
| result = self.driver._build_service_accounts_gce_list(service_accounts=None) |
| self.assertEqual( |
| result, |
| [ |
| { |
| "email": "default", |
| "scopes": ["https://www.googleapis.com/auth/devstorage.read_only"], |
| } |
| ], |
| ) |
| |
| def test_build_service_account_gce_struct_no_service_account(self): |
| result = self.driver._build_service_accounts_gce_list(service_accounts=[]) |
| self.assertEqual(result, []) |
| |
| def test_build_service_account_gce_struct_custom_service_account(self): |
| data = [{"email": "1", "scopes": ["a"]}, {"email": "2", "scopes": ["b"]}] |
| expected_result = [ |
| {"email": "1", "scopes": ["https://www.googleapis.com/auth/a"]}, |
| {"email": "2", "scopes": ["https://www.googleapis.com/auth/b"]}, |
| ] |
| |
| result = self.driver._build_service_accounts_gce_list(service_accounts=data) |
| self.assertEqual(result, expected_result) |
| |
| def test_timestamp_to_datetime(self): |
| timestamp1 = "2013-06-26T10:05:19.340-07:00" |
| datetime1 = datetime.datetime(2013, 6, 26, 17, 5, 19) |
| self.assertEqual(timestamp_to_datetime(timestamp1), datetime1) |
| timestamp2 = "2013-06-26T17:43:15.000-00:00" |
| datetime2 = datetime.datetime(2013, 6, 26, 17, 43, 15) |
| self.assertEqual(timestamp_to_datetime(timestamp2), datetime2) |
| |
| def test_get_object_by_kind(self): |
| obj = self.driver._get_object_by_kind(None) |
| self.assertIsNone(obj) |
| obj = self.driver._get_object_by_kind("") |
| self.assertIsNone(obj) |
| obj = self.driver._get_object_by_kind( |
| "https://www.googleapis.com/compute/v1/projects/project_name/" |
| "global/targetHttpProxies/web-proxy" |
| ) |
| self.assertEqual(obj.name, "web-proxy") |
| |
| def test_get_region_from_zone(self): |
| zone1 = self.driver.ex_get_zone("us-central1-a") |
| expected_region1 = "us-central1" |
| region1 = self.driver._get_region_from_zone(zone1) |
| self.assertEqual(region1.name, expected_region1) |
| zone2 = self.driver.ex_get_zone("europe-west1-b") |
| expected_region2 = "europe-west1" |
| region2 = self.driver._get_region_from_zone(zone2) |
| self.assertEqual(region2.name, expected_region2) |
| |
| def test_get_volume(self): |
| volume_name = "lcdisk" |
| volume = self.driver.ex_get_volume(volume_name) |
| self.assertTrue(isinstance(volume, StorageVolume)) |
| self.assertEqual(volume.name, volume_name) |
| |
| def test_get_volume_location(self): |
| volume_name = "lcdisk" |
| location = self.driver.zone |
| volume = self.driver.ex_get_volume(volume_name, zone=location) |
| self.assertTrue(isinstance(volume, StorageVolume)) |
| self.assertEqual(volume.name, volume_name) |
| |
| def test_get_volume_location_name(self): |
| volume_name = "lcdisk" |
| location = self.driver.zone |
| volume = self.driver.ex_get_volume(volume_name, zone=location.name) |
| self.assertTrue(isinstance(volume, StorageVolume)) |
| self.assertEqual(volume.name, volume_name) |
| |
| def test_find_zone_or_region(self): |
| zone1 = self.driver._find_zone_or_region("libcloud-demo-np-node", "instances") |
| self.assertEqual(zone1.name, "us-central2-a") |
| zone2 = self.driver._find_zone_or_region("libcloud-demo-europe-np-node", "instances") |
| self.assertEqual(zone2.name, "europe-west1-a") |
| region = self.driver._find_zone_or_region("libcloud-demo-address", "addresses", region=True) |
| self.assertEqual(region.name, "us-central1") |
| |
| def test_match_images(self): |
| project = "debian-cloud" |
| image = self.driver._match_images(project, "debian-7") |
| self.assertEqual(image.name, "debian-7-wheezy-v20131120") |
| image = self.driver._match_images(project, "backports") |
| self.assertEqual(image.name, "backports-debian-7-wheezy-v20131127") |
| |
| def test_build_disk_gce_struct(self): |
| device_name = "disk_name" |
| disk_name = None |
| source = self.driver.ex_get_volume("lcdisk") |
| is_boot = True |
| # source as input |
| d = self.driver._build_disk_gce_struct( |
| device_name=device_name, source=source, disk_name=disk_name, is_boot=is_boot |
| ) |
| self.assertEqual(source.extra["selfLink"], d["source"]) |
| self.assertTrue(d["boot"]) |
| self.assertTrue(d["autoDelete"]) |
| self.assertEqual("READ_WRITE", d["mode"]) |
| self.assertFalse("initializeParams" in d) |
| |
| # image as input |
| device_name = "disk_name" |
| disk_type = self.driver.ex_get_disktype("pd-ssd", "us-central1-a") |
| image = self.driver.ex_get_image("debian-7") |
| source = None |
| is_boot = True |
| d = self.driver._build_disk_gce_struct( |
| device_name=device_name, disk_type=disk_type, image=image, is_boot=is_boot |
| ) |
| self.assertEqual("READ_WRITE", d["mode"]) |
| self.assertEqual("PERSISTENT", d["type"]) |
| self.assertTrue("initializeParams" in d and isinstance(d["initializeParams"], dict)) |
| self.assertTrue( |
| all(k in d["initializeParams"] for k in ["sourceImage", "diskType", "diskName"]) |
| ) |
| self.assertTrue(d["initializeParams"]["sourceImage"].startswith("https://")) |
| self.assertTrue(d["autoDelete"]) |
| self.assertTrue(d["boot"]) |
| |
| def test_build_network_gce_struct(self): |
| network = self.driver.ex_get_network("lcnetwork") |
| address = self.driver.ex_get_address("lcaddress") |
| internalip = self.driver.ex_get_address("testaddress") |
| subnetwork_name = "cf-972cf02e6ad49112" |
| subnetwork = self.driver.ex_get_subnetwork(subnetwork_name) |
| d = self.driver._build_network_gce_struct(network, subnetwork, address) |
| self.assertTrue("network" in d) |
| self.assertTrue("subnetwork" in d) |
| self.assertTrue("kind" in d and d["kind"] == "compute#instanceNetworkInterface") |
| self.assertEqual(d["accessConfigs"][0]["natIP"], address.address) |
| # test with internal IP |
| d = self.driver._build_network_gce_struct( |
| network, subnetwork, address, internal_ip=internalip |
| ) |
| self.assertTrue("network" in d) |
| self.assertTrue("subnetwork" in d) |
| self.assertTrue("kind" in d and d["kind"] == "compute#instanceNetworkInterface") |
| self.assertEqual(d["accessConfigs"][0]["natIP"], address.address) |
| self.assertEqual(d["networkIP"], internalip) |
| network = self.driver.ex_get_network("default") |
| d = self.driver._build_network_gce_struct(network) |
| self.assertTrue("network" in d) |
| self.assertFalse("subnetwork" in d) |
| self.assertTrue("kind" in d and d["kind"] == "compute#instanceNetworkInterface") |
| |
| def test_build_scheduling_gce_struct(self): |
| self.assertFalse(self.driver._build_scheduling_gce_struct(None, None, None)) |
| # on_host_maintenance bad value should raise a Valueerror |
| self.assertRaises( |
| ValueError, |
| self.driver._build_service_account_gce_struct, |
| 'on_host_maintenance="foobar"', |
| ) |
| # on_host_maintenance is 'MIGRATE' and prempt is True |
| self.assertRaises( |
| ValueError, |
| self.driver._build_service_account_gce_struct, |
| 'on_host_maintenance="MIGRATE"', |
| "preemptible=True", |
| ) |
| # automatic_restart is True and prempt is True |
| self.assertRaises( |
| ValueError, |
| self.driver._build_service_account_gce_struct, |
| 'automatic_restart="True"', |
| "preemptible=True", |
| ) |
| |
| actual = self.driver._build_scheduling_gce_struct("TERMINATE", True, False) |
| self.assertTrue("automaticRestart" in actual and actual["automaticRestart"] is True) |
| self.assertTrue( |
| "onHostMaintenance" in actual and actual["onHostMaintenance"] == "TERMINATE" |
| ) |
| self.assertTrue("preemptible" in actual) |
| self.assertFalse(actual["preemptible"]) |
| |
| def test_build_service_account_gce_struct(self): |
| self.assertRaises(ValueError, self.driver._build_service_account_gce_struct, None) |
| input = {"scopes": ["compute-ro"]} |
| actual = self.driver._build_service_account_gce_struct(input) |
| self.assertTrue("email" in actual) |
| self.assertTrue("scopes" in actual) |
| |
| input = {"scopes": ["compute-ro"], "email": "test@test.com"} |
| actual = self.driver._build_service_account_gce_struct(input) |
| self.assertTrue("email" in actual) |
| self.assertEqual(actual["email"], "test@test.com") |
| self.assertTrue("scopes" in actual) |
| |
| def test_build_service_account_gce_list(self): |
| # ensure we have a list |
| self.assertRaises(ValueError, self.driver._build_service_accounts_gce_list, "foo") |
| # no input |
| actual = self.driver._build_service_accounts_gce_list() |
| self.assertTrue(len(actual) == 1) |
| self.assertTrue("email" in actual[0]) |
| self.assertTrue("scopes" in actual[0]) |
| |
| def test_get_selflink_or_name(self): |
| network = self.driver.ex_get_network("lcnetwork") |
| |
| # object as input |
| actual = self.driver._get_selflink_or_name(network, False, "network") |
| self.assertEqual("lcnetwork", actual) |
| actual = self.driver._get_selflink_or_name(network, True, "network") |
| self.assertTrue(actual.startswith("https://")) |
| |
| # name-only as input |
| actual = self.driver._get_selflink_or_name("lcnetwork", True, "network") |
| self.assertTrue(actual.startswith("https://")) |
| |
| actual = self.driver._get_selflink_or_name("lcnetwork", False, "network") |
| self.assertTrue("lcnetwork", actual) |
| |
| # if selflinks is true, we need objname |
| self.assertRaises(ValueError, self.driver._get_selflink_or_name, "lcnetwork", True) |
| |
| def test_ex_get_serial_output(self): |
| self.assertRaises(ValueError, self.driver.ex_get_serial_output, "foo") |
| node = self.driver.ex_get_node("node-name", "us-central1-a") |
| self.assertTrue( |
| self.driver.ex_get_serial_output(node), |
| "This is some serial\r\noutput for you.", |
| ) |
| |
| def test_ex_list(self): |
| d = self.driver |
| # Test the default case for all list methods |
| # (except list_volume_snapshots, which requires an arg) |
| for list_fn in ( |
| d.ex_list_addresses, |
| d.ex_list_backendservices, |
| d.ex_list_disktypes, |
| d.ex_list_firewalls, |
| d.ex_list_forwarding_rules, |
| d.ex_list_healthchecks, |
| d.ex_list_networks, |
| d.ex_list_subnetworks, |
| d.ex_list_project_images, |
| d.ex_list_regions, |
| d.ex_list_routes, |
| d.ex_list_snapshots, |
| d.ex_list_targethttpproxies, |
| d.ex_list_targetinstances, |
| d.ex_list_targetpools, |
| d.ex_list_urlmaps, |
| d.ex_list_zones, |
| d.list_images, |
| d.list_locations, |
| d.list_nodes, |
| d.list_sizes, |
| d.list_volumes, |
| ): |
| full_list = [item.name for item in list_fn()] |
| li = d.ex_list(list_fn) |
| iter_list = [item.name for sublist in li for item in sublist] |
| self.assertEqual(full_list, iter_list) |
| |
| # Test paging & filtering with a single list function as they require |
| # additional test fixtures |
| list_fn = d.ex_list_regions |
| for count, sublist in zip((2, 1), d.ex_list(list_fn).page(2)): |
| self.assertTrue(len(sublist) == count) |
| for sublist in d.ex_list(list_fn).filter("name eq us-central1"): |
| self.assertTrue(len(sublist) == 1) |
| self.assertEqual(sublist[0].name, "us-central1") |
| |
| def test_ex_list_addresses(self): |
| address_list = self.driver.ex_list_addresses() |
| address_list_all = self.driver.ex_list_addresses("all") |
| address_list_uc1 = self.driver.ex_list_addresses("us-central1") |
| address_list_global = self.driver.ex_list_addresses("global") |
| self.assertEqual(len(address_list), 2) |
| self.assertEqual(len(address_list_all), 5) |
| self.assertEqual(len(address_list_global), 1) |
| self.assertEqual(address_list[0].name, "libcloud-demo-address") |
| self.assertEqual(address_list_uc1[0].name, "libcloud-demo-address") |
| self.assertEqual(address_list_global[0].name, "lcaddressglobal") |
| names = [a.name for a in address_list_all] |
| self.assertTrue("libcloud-demo-address" in names) |
| |
| def test_ex_list_backendservices(self): |
| self.backendservices_mock = "empty" |
| backendservices_list = self.driver.ex_list_backendservices() |
| self.assertListEqual(backendservices_list, []) |
| |
| self.backendservices_mock = "web-service" |
| backendservices_list = self.driver.ex_list_backendservices() |
| web_service = backendservices_list[0] |
| self.assertEqual(web_service.name, "web-service") |
| self.assertEqual(len(web_service.healthchecks), 1) |
| self.assertEqual(len(web_service.backends), 2) |
| |
| def test_ex_list_healthchecks(self): |
| healthchecks = self.driver.ex_list_healthchecks() |
| self.assertEqual(len(healthchecks), 3) |
| self.assertEqual(healthchecks[0].name, "basic-check") |
| |
| def test_ex_list_firewalls(self): |
| firewalls = self.driver.ex_list_firewalls() |
| self.assertEqual(len(firewalls), 5) |
| self.assertEqual(firewalls[0].name, "default-allow-internal") |
| |
| def test_ex_list_forwarding_rules(self): |
| forwarding_rules = self.driver.ex_list_forwarding_rules() |
| forwarding_rules_all = self.driver.ex_list_forwarding_rules("all") |
| forwarding_rules_uc1 = self.driver.ex_list_forwarding_rules("us-central1") |
| self.assertEqual(len(forwarding_rules), 2) |
| self.assertEqual(len(forwarding_rules_all), 2) |
| self.assertEqual(forwarding_rules[0].name, "lcforwardingrule") |
| self.assertEqual(forwarding_rules_uc1[0].name, "lcforwardingrule") |
| names = [f.name for f in forwarding_rules_all] |
| self.assertTrue("lcforwardingrule" in names) |
| |
| def test_ex_list_forwarding_rules_global(self): |
| forwarding_rules = self.driver.ex_list_forwarding_rules(global_rules=True) |
| self.assertEqual(len(forwarding_rules), 2) |
| self.assertEqual(forwarding_rules[0].name, "http-rule") |
| names = [f.name for f in forwarding_rules] |
| self.assertListEqual(names, ["http-rule", "http-rule2"]) |
| |
| def test_list_images(self): |
| local_images = self.driver.list_images() |
| all_deprecated_images = self.driver.list_images(ex_include_deprecated=True) |
| debian_images = self.driver.list_images(ex_project="debian-cloud") |
| local_plus_deb = self.driver.list_images(["debian-cloud", "project_name"]) |
| self.assertEqual(len(local_images), 52) |
| self.assertEqual(len(all_deprecated_images), 180) |
| self.assertEqual(len(debian_images), 2) |
| self.assertEqual(len(local_plus_deb), 6) |
| self.assertEqual(local_images[0].name, "custom-image1") |
| self.assertEqual(local_images[2].name, "custom-image") |
| self.assertEqual(debian_images[1].name, "debian-7-wheezy-v20131120") |
| |
| def test_ex_destroy_instancegroup(self): |
| name = "myname" |
| zone = "us-central1-a" |
| uig = self.driver.ex_get_instancegroup(name, zone) |
| self.assertTrue(self.driver.ex_destroy_instancegroup(uig)) |
| |
| def test_ex_get_instancegroup(self): |
| name = "myname" |
| loc = "us-central1-a" |
| actual = self.driver.ex_get_instancegroup(name, loc) |
| self.assertEqual(actual.name, name) |
| self.assertEqual(actual.zone.name, loc) |
| |
| def test_ex_create_instancegroup(self): |
| name = "myname" |
| loc = "us-central1-a" |
| actual = self.driver.ex_create_instancegroup(name, loc) |
| self.assertEqual(actual.name, name) |
| self.assertEqual(actual.zone.name, loc) |
| |
| def test_ex_list_instancegroups(self): |
| loc = "us-central1-a" |
| actual = self.driver.ex_list_instancegroups(loc) |
| self.assertTrue(len(actual) == 2) |
| self.assertEqual(actual[0].name, "myname") |
| self.assertEqual(actual[1].name, "myname2") |
| |
| def test_ex_list_instancegroups_zone_attribute_not_present_in_response(self): |
| GCEMockHttp.type = "zone_attribute_not_present" |
| loc = "us-central1-a" |
| actual = self.driver.ex_list_instancegroups(loc) |
| self.assertTrue(len(actual) == 2) |
| self.assertEqual(actual[0].name, "myname") |
| self.assertEqual(actual[1].name, "myname2") |
| |
| def test_ex_instancegroup_list_instances(self): |
| name = "myname" |
| loc = "us-central1-a" |
| gceobj = self.driver.ex_get_instancegroup(name, loc) |
| actual = self.driver.ex_instancegroup_list_instances(gceobj) |
| self.assertTrue(len(actual) == 2) |
| for node in actual: |
| self.assertTrue(isinstance(node, Node)) |
| self.assertEqual(loc, node.extra["zone"].name) |
| |
| def test_ex_instancegroup_add_instances(self): |
| name = "myname" |
| loc = "us-central1-a" |
| gceobj = self.driver.ex_get_instancegroup(name, loc) |
| node_name = self.driver.ex_get_node("node-name", loc) |
| lcnode = self.driver.ex_get_node("lcnode-001", loc) |
| node_list = [node_name, lcnode] |
| self.assertTrue(self.driver.ex_instancegroup_add_instances(gceobj, node_list)) |
| |
| def test_ex_instancegroup_remove_instances(self): |
| name = "myname" |
| loc = "us-central1-a" |
| gceobj = self.driver.ex_get_instancegroup(name, loc) |
| node_name = self.driver.ex_get_node("node-name", loc) |
| lcnode = self.driver.ex_get_node("lcnode-001", loc) |
| node_list = [node_name, lcnode] |
| self.assertTrue(self.driver.ex_instancegroup_remove_instances(gceobj, node_list)) |
| |
| def test_ex_instancegroup_set_named_ports(self): |
| name = "myname" |
| loc = "us-central1-a" |
| gceobj = self.driver.ex_get_instancegroup(name, loc) |
| named_ports = [{"name": "foo", "port": 4444}] |
| # base case |
| self.assertTrue(self.driver.ex_instancegroup_set_named_ports(gceobj, named_ports)) |
| # specify nothing, default is empty list |
| self.assertTrue(self.driver.ex_instancegroup_set_named_ports(gceobj)) |
| # specify empty list |
| self.assertTrue(self.driver.ex_instancegroup_set_named_ports(gceobj, [])) |
| # raise valueerror if string is passed in |
| self.assertRaises( |
| ValueError, self.driver.ex_instancegroup_set_named_ports, gceobj, "foobar" |
| ) |
| # raise valueerror if dictionary is passed in |
| self.assertRaises( |
| ValueError, |
| self.driver.ex_instancegroup_set_named_ports, |
| gceobj, |
| {"name": "foo", "port": 4444}, |
| ) |
| |
| def test_ex_instancegroupmanager_set_autohealing_policies(self): |
| kwargs = { |
| "host": "lchost", |
| "path": "/lc", |
| "port": 8000, |
| "interval": 10, |
| "timeout": 10, |
| "unhealthy_threshold": 4, |
| "healthy_threshold": 3, |
| "description": "test healthcheck", |
| } |
| healthcheck_name = "lchealthcheck" |
| hc = self.driver.ex_create_healthcheck(healthcheck_name, **kwargs) |
| |
| ig_name = "myinstancegroup" |
| ig_zone = "us-central1-a" |
| manager = self.driver.ex_get_instancegroupmanager(ig_name, ig_zone) |
| |
| res = self.driver.ex_instancegroupmanager_set_autohealingpolicies( |
| manager=manager, healthcheck=hc, initialdelaysec=2 |
| ) |
| self.assertTrue(res) |
| |
| res = manager.set_autohealingpolicies(healthcheck=hc, initialdelaysec=2) |
| self.assertTrue(res) |
| |
| def test_ex_create_instancegroupmanager(self): |
| name = "myinstancegroup" |
| zone = "us-central1-a" |
| size = 4 |
| template_name = "my-instance-template1" |
| template = self.driver.ex_get_instancetemplate(template_name) |
| mig = self.driver.ex_create_instancegroupmanager( |
| name, zone, template, size, base_instance_name="base-foo" |
| ) |
| |
| self.assertEqual(mig.name, name) |
| self.assertEqual(mig.size, size) |
| self.assertEqual(mig.zone.name, zone) |
| |
| def test_ex_create_instancegroupmanager_shared_network(self): |
| name = "myinstancegroup-shared-network" |
| zone = "us-central1-a" |
| size = 4 |
| template_name = "my-instance-template-shared-network" |
| template = self.driver.ex_get_instancetemplate(template_name) |
| mig = self.driver.ex_create_instancegroupmanager( |
| name, zone, template, size, base_instance_name="base-foo" |
| ) |
| |
| self.assertEqual(mig.name, name) |
| self.assertEqual(mig.size, size) |
| self.assertEqual(mig.zone.name, zone) |
| |
| def test_ex_create_instancetemplate(self): |
| name = "my-instance-template1" |
| actual = self.driver.ex_create_instancetemplate( |
| name, size="n1-standard-1", image="debian-7", network="default" |
| ) |
| self.assertEqual(actual.name, name) |
| self.assertEqual(actual.extra["properties"]["machineType"], "n1-standard-1") |
| |
| def test_list_locations(self): |
| locations = self.driver.list_locations() |
| self.assertEqual(len(locations), 6) |
| self.assertEqual(locations[0].name, "asia-east1-a") |
| |
| def test_ex_list_routes(self): |
| routes = self.driver.ex_list_routes() |
| self.assertEqual(len(routes), 3) |
| self.assertTrue("lcdemoroute" in [route.name for route in routes]) |
| |
| def test_ex_list_sslcertificate(self): |
| ssl_name = "example" |
| certs = self.driver.ex_list_sslcertificates() |
| self.assertEqual(certs[0].name, ssl_name) |
| self.assertTrue(len(certs) == 1) |
| |
| def test_ex_list_subnetworks(self): |
| subnetworks = self.driver.ex_list_subnetworks() |
| self.assertEqual(len(subnetworks), 1) |
| self.assertEqual(subnetworks[0].name, "cf-972cf02e6ad49112") |
| self.assertEqual(subnetworks[0].cidr, "10.128.0.0/20") |
| subnetworks = self.driver.ex_list_subnetworks("all") |
| self.assertEqual(len(subnetworks), 4) |
| |
| def test_ex_create_sslcertificate(self): |
| ssl_name = "example" |
| private_key = "-----BEGIN RSA PRIVATE KEY-----\nfoobar==\n-----END RSA PRIVATE KEY-----\n" |
| certificate = "-----BEGIN CERTIFICATE-----\nfoobar==\n-----END CERTIFICATE-----\n" |
| ssl = self.driver.ex_create_sslcertificate( |
| ssl_name, certificate=certificate, private_key=private_key |
| ) |
| self.assertEqual(ssl_name, ssl.name) |
| self.assertEqual(certificate, ssl.certificate) |
| |
| def test_ex_create_subnetwork(self): |
| name = "cf-972cf02e6ad49112" |
| cidr = "10.128.0.0/20" |
| network_name = "cf" |
| network = self.driver.ex_get_network(network_name) |
| region_name = "us-central1" |
| region = self.driver.ex_get_region(region_name) |
| description = "LCTestSubnet" |
| privateipgoogleaccess = True |
| secondaryipranges = [{"rangeName": "secondary", "ipCidrRange": "192.168.168.0/24"}] |
| # test by network/region name |
| subnet = self.driver.ex_create_subnetwork( |
| name, |
| cidr, |
| network_name, |
| region_name, |
| description=description, |
| privateipgoogleaccess=privateipgoogleaccess, |
| secondaryipranges=secondaryipranges, |
| ) |
| self.assertTrue(isinstance(subnet, GCESubnetwork)) |
| self.assertTrue(isinstance(subnet.region, GCERegion)) |
| self.assertTrue(isinstance(subnet.network, GCENetwork)) |
| self.assertEqual(subnet.name, name) |
| self.assertEqual(subnet.cidr, cidr) |
| self.assertEqual(subnet.extra["privateIpGoogleAccess"], privateipgoogleaccess) |
| self.assertEqual(subnet.extra["secondaryIpRanges"], secondaryipranges) |
| # test by network/region object |
| subnet = self.driver.ex_create_subnetwork(name, cidr, network, region) |
| self.assertTrue(isinstance(subnet, GCESubnetwork)) |
| self.assertTrue(isinstance(subnet.region, GCERegion)) |
| self.assertTrue(isinstance(subnet.network, GCENetwork)) |
| self.assertEqual(subnet.name, name) |
| self.assertEqual(subnet.cidr, cidr) |
| self.assertEqual(subnet.extra["privateIpGoogleAccess"], privateipgoogleaccess) |
| self.assertEqual(subnet.extra["secondaryIpRanges"], secondaryipranges) |
| |
| def test_ex_destroy_subnetwork(self): |
| name = "cf-972cf02e6ad49112" |
| region_name = "us-central1" |
| region = self.driver.ex_get_region(region_name) |
| # delete with no region |
| self.assertTrue(self.driver.ex_destroy_subnetwork(name)) |
| # delete with region name |
| self.assertTrue(self.driver.ex_destroy_subnetwork(name, region_name)) |
| # delete with region object |
| self.assertTrue(self.driver.ex_destroy_subnetwork(name, region)) |
| |
| def test_ex_get_sslcertificate(self): |
| ssl_name = "example" |
| ssl = self.driver.ex_get_sslcertificate(ssl_name) |
| self.assertEqual(ssl.name, ssl_name) |
| self.assertTrue(hasattr(ssl, "certificate")) |
| self.assertTrue(len(ssl.certificate)) |
| |
| def test_ex_get_accelerator_type(self): |
| name = "nvidia-tesla-k80" |
| zone = self.driver.ex_get_zone("us-central1-a") |
| accelerator_type = self.driver.ex_get_accelerator_type(name, zone) |
| self.assertEqual(accelerator_type.name, name) |
| self.assertEqual(accelerator_type.zone, zone) |
| |
| def test_ex_get_subnetwork(self): |
| name = "cf-972cf02e6ad49112" |
| region_name = "us-central1" |
| region = self.driver.ex_get_region(region_name) |
| # fetch by no region |
| subnetwork = self.driver.ex_get_subnetwork(name) |
| self.assertEqual(subnetwork.name, name) |
| # fetch by region name |
| subnetwork = self.driver.ex_get_subnetwork(name, region_name) |
| self.assertEqual(subnetwork.name, name) |
| # fetch by region object |
| subnetwork = self.driver.ex_get_subnetwork(name, region) |
| self.assertEqual(subnetwork.name, name) |
| # do the same but this time by resource URL |
| url = "https://www.googleapis.com/compute/v1/projects/project_name/regions/us-central1/subnetworks/cf-972cf02e6ad49112" |
| # fetch by no region |
| subnetwork = self.driver.ex_get_subnetwork(url) |
| self.assertEqual(subnetwork.name, name) |
| self.assertEqual(subnetwork.region.name, region_name) |
| # test with a subnetwork that is under a different project |
| url_other = "https://www.googleapis.com/compute/v1/projects/other_name/regions/us-central1/subnetworks/cf-972cf02e6ad49114" |
| subnetwork = self.driver.ex_get_subnetwork(url_other) |
| self.assertEqual(subnetwork.name, "cf-972cf02e6ad49114") |
| |
| def test_ex_list_networks(self): |
| networks = self.driver.ex_list_networks() |
| self.assertEqual(len(networks), 3) |
| self.assertEqual(networks[0].name, "cf") |
| self.assertEqual(networks[0].mode, "auto") |
| self.assertEqual(len(networks[0].subnetworks), 4) |
| self.assertEqual(networks[1].name, "custom") |
| self.assertEqual(networks[1].mode, "custom") |
| self.assertEqual(len(networks[1].subnetworks), 1) |
| self.assertEqual(networks[2].name, "default") |
| self.assertEqual(networks[2].mode, "legacy") |
| |
| def test_list_nodes(self): |
| nodes = self.driver.list_nodes() |
| nodes_all = self.driver.list_nodes(ex_zone="all") |
| nodes_uc1a = self.driver.list_nodes(ex_zone="us-central1-a") |
| nodes_uc1b = self.driver.list_nodes(ex_zone="us-central1-b") |
| self.assertEqual(len(nodes), 1) |
| self.assertEqual(len(nodes_all), 8) |
| self.assertEqual(len(nodes_uc1a), 1) |
| self.assertEqual(len(nodes_uc1b), 0) |
| self.assertEqual(nodes[0].name, "node-name") |
| self.assertEqual(nodes_uc1a[0].name, "node-name") |
| self.assertEqual(nodes_uc1a[0].extra["cpuPlatform"], "Intel Skylake") |
| self.assertEqual(nodes_uc1a[0].extra["minCpuPlatform"], "Intel Skylake") |
| |
| names = [n.name for n in nodes_all] |
| self.assertTrue("node-name" in names) |
| |
| states = [n.state for n in nodes_all] |
| self.assertTrue(NodeState.SUSPENDED in states) |
| |
| def test_ex_list_regions(self): |
| regions = self.driver.ex_list_regions() |
| self.assertEqual(len(regions), 3) |
| self.assertEqual(regions[0].name, "europe-west1") |
| |
| def test_ex_list_snapshots(self): |
| snapshots = self.driver.ex_list_snapshots() |
| self.assertEqual(len(snapshots), 2) |
| self.assertEqual(snapshots[0].name, "lcsnapshot") |
| |
| def test_ex_list_targethttpproxies(self): |
| target_proxies = self.driver.ex_list_targethttpproxies() |
| self.assertEqual(len(target_proxies), 2) |
| self.assertEqual(target_proxies[0].name, "web-proxy") |
| names = [t.name for t in target_proxies] |
| self.assertListEqual(names, ["web-proxy", "web-proxy2"]) |
| |
| def test_ex_list_targetinstances(self): |
| target_instances = self.driver.ex_list_targetinstances() |
| target_instances_all = self.driver.ex_list_targetinstances("all") |
| target_instances_uc1 = self.driver.ex_list_targetinstances("us-central1-a") |
| self.assertEqual(len(target_instances), 2) |
| self.assertEqual(len(target_instances_all), 2) |
| self.assertEqual(len(target_instances_uc1), 2) |
| self.assertEqual(target_instances[0].name, "hello") |
| self.assertEqual(target_instances_uc1[0].name, "hello") |
| names = [t.name for t in target_instances_all] |
| self.assertTrue("lctargetinstance" in names) |
| |
| def test_ex_list_targetpools(self): |
| target_pools = self.driver.ex_list_targetpools() |
| target_pools_all = self.driver.ex_list_targetpools("all") |
| target_pools_uc1 = self.driver.ex_list_targetpools("us-central1") |
| self.assertEqual(len(target_pools), 4) |
| self.assertEqual(len(target_pools_all), 5) |
| self.assertEqual(len(target_pools_uc1), 4) |
| self.assertEqual(target_pools[0].name, "lctargetpool") |
| self.assertEqual(target_pools_uc1[0].name, "lctargetpool") |
| names = [t.name for t in target_pools_all] |
| self.assertTrue("www-pool" in names) |
| |
| def test_list_sizes(self): |
| sizes = self.driver.list_sizes() |
| sizes_all = self.driver.list_sizes("all") |
| self.assertEqual(len(sizes), 23) |
| self.assertEqual(len(sizes_all), 100) |
| self.assertEqual(sizes[0].name, "f1-micro") |
| self.assertEqual(sizes[0].extra["zone"].name, "us-central1-a") |
| self.assertEqual(sizes[1].name, "a2-highgpu-1g") |
| self.assertEqual(len(sizes[1].extra["accelerators"]), 1) |
| self.assertEqual( |
| sizes[1].extra["accelerators"][0]["guestAcceleratorType"], |
| "nvidia-tesla-v100", |
| ) |
| names = [s.name for s in sizes_all] |
| self.assertEqual(names.count("n1-standard-1"), 5) |
| |
| def test_ex_get_license(self): |
| license = self.driver.ex_get_license("suse-cloud", "sles-12") |
| self.assertEqual(license.name, "sles-12") |
| self.assertTrue(license.charges_use_fee) |
| |
| def test_list_disktypes(self): |
| disktypes = self.driver.ex_list_disktypes() |
| disktypes_all = self.driver.ex_list_disktypes("all") |
| disktypes_uc1a = self.driver.ex_list_disktypes("us-central1-a") |
| self.assertEqual(len(disktypes), 2) |
| self.assertEqual(len(disktypes_all), 9) |
| self.assertEqual(len(disktypes_uc1a), 2) |
| self.assertEqual(disktypes[0].name, "pd-ssd") |
| self.assertEqual(disktypes_uc1a[0].name, "pd-ssd") |
| names = [v.name for v in disktypes_all] |
| self.assertTrue("pd-standard" in names) |
| self.assertTrue("local-ssd" in names) |
| |
| def test_ex_list_instancegroupmanagers(self): |
| instancegroupmanagers = self.driver.ex_list_instancegroupmanagers() |
| instancegroupmanagers_all = self.driver.ex_list_instancegroupmanagers("all") |
| instancegroupmanagers_ue1b = self.driver.ex_list_instancegroupmanagers("us-east1-b") |
| self.assertEqual(len(instancegroupmanagers), 1) |
| self.assertEqual(len(instancegroupmanagers_all), 2) |
| self.assertEqual(len(instancegroupmanagers_ue1b), 1) |
| |
| def test_ex_instancegroupmanager_list_managed_instances(self): |
| ig_name = "myinstancegroup" |
| ig_zone = "us-central1-a" |
| mig = self.driver.ex_get_instancegroupmanager(ig_name, ig_zone) |
| instances = mig.list_managed_instances() |
| self.assertTrue(all([x["currentAction"] == "NONE" for x in instances])) |
| self.assertTrue("base-foo-2vld" in [x["name"] for x in instances]) |
| self.assertEqual(len(instances), 4) |
| |
| def test_ex_list_instancetemplates(self): |
| instancetemplates = self.driver.ex_list_instancetemplates() |
| self.assertEqual(len(instancetemplates), 1) |
| self.assertEqual(instancetemplates[0].name, "my-instance-template1") |
| |
| def test_ex_list_autoscalers(self): |
| autoscalers = self.driver.ex_list_autoscalers("all") |
| self.assertEqual(len(autoscalers), 1) |
| self.assertEqual(autoscalers[0].name, "my-autoscaler") |
| |
| def test_ex_list_urlmaps(self): |
| urlmaps_list = self.driver.ex_list_urlmaps() |
| web_map = urlmaps_list[0] |
| self.assertEqual(web_map.name, "web-map") |
| self.assertEqual(len(web_map.host_rules), 0) |
| self.assertEqual(len(web_map.path_matchers), 0) |
| self.assertEqual(len(web_map.tests), 0) |
| |
| def test_list_volumes(self): |
| volumes = self.driver.list_volumes() |
| volumes_all = self.driver.list_volumes("all") |
| volumes_uc1a = self.driver.list_volumes("us-central1-a") |
| self.assertEqual(len(volumes), 2) |
| self.assertEqual(len(volumes_all), 17) |
| self.assertEqual(len(volumes_uc1a), 2) |
| self.assertEqual(volumes[0].name, "lcdisk") |
| self.assertEqual(volumes_uc1a[0].name, "lcdisk") |
| names = [v.name for v in volumes_all] |
| self.assertTrue("libcloud-demo-europe-boot-disk" in names) |
| |
| def test_ex_list_zones(self): |
| zones = self.driver.ex_list_zones() |
| self.assertEqual(len(zones), 6) |
| self.assertEqual(zones[0].name, "asia-east1-a") |
| |
| def test_ex_create_address_global(self): |
| address_name = "lcaddressglobal" |
| address = self.driver.ex_create_address(address_name, "global") |
| self.assertTrue(isinstance(address, GCEAddress)) |
| self.assertEqual(address.name, address_name) |
| self.assertEqual(address.region, "global") |
| |
| def test_ex_create_address(self): |
| address_name = "lcaddress" |
| address = self.driver.ex_create_address(address_name) |
| self.assertTrue(isinstance(address, GCEAddress)) |
| self.assertEqual(address.name, address_name) |
| |
| def test_ex_create_address_internal(self): |
| address_name = "lcaddressinternal" |
| address = self.driver.ex_create_address( |
| address_name, |
| region="us-central1", |
| address="10.128.0.12", |
| address_type="INTERNAL", |
| subnetwork="subnet-1", |
| ) |
| self.assertTrue(isinstance(address, GCEAddress)) |
| self.assertEqual(address.name, address_name) |
| self.assertEqual(address.address, "10.128.0.12") |
| |
| self.assertRaises( |
| ValueError, |
| self.driver.ex_create_address, |
| address_name, |
| address_type="WRONG", |
| ) |
| self.assertRaises( |
| ValueError, |
| self.driver.ex_create_address, |
| address_name, |
| address_type="EXTERNAL", |
| subnetwork="subnet-1", |
| ) |
| |
| def test_ex_create_backend(self): |
| # Note: this is an internal object, no API call is made |
| # and no fixture is needed specifically for GCEBackend, however |
| # it does rely on an InstanceGroup object. |
| ig = self.driver.ex_get_instancegroup("myinstancegroup", "us-central1-a") |
| |
| backend = self.driver.ex_create_backend(ig) |
| |
| self.assertTrue(isinstance(backend, GCEBackend)) |
| self.assertEqual(backend.name, "{}/instanceGroups/{}".format(ig.zone.name, ig.name)) |
| self.assertEqual(backend.instance_group.name, ig.name) |
| self.assertEqual(backend.balancing_mode, "UTILIZATION") |
| |
| def test_ex_create_backendservice(self): |
| backendservice_name = "web-service" |
| |
| ig1 = self.driver.ex_get_instancegroup("myinstancegroup", "us-central1-a") |
| backend1 = self.driver.ex_create_backend(ig1) |
| ig2 = self.driver.ex_get_instancegroup("myinstancegroup2", "us-central1-a") |
| backend2 = self.driver.ex_create_backend(ig2) |
| |
| backendservice = self.driver.ex_create_backendservice( |
| name=backendservice_name, |
| healthchecks=["lchealthcheck"], |
| backends=[backend1, backend2], |
| ) |
| self.assertTrue(isinstance(backendservice, GCEBackendService)) |
| self.assertEqual(backendservice.name, backendservice_name) |
| self.assertEqual(len(backendservice.backends), 2) |
| ig_links = [ig1.extra["selfLink"], ig2.extra["selfLink"]] |
| for be in backendservice.backends: |
| self.assertTrue(be["group"] in ig_links) |
| |
| def test_ex_create_healthcheck(self): |
| healthcheck_name = "lchealthcheck" |
| kwargs = { |
| "host": "lchost", |
| "path": "/lc", |
| "port": 8000, |
| "interval": 10, |
| "timeout": 10, |
| "unhealthy_threshold": 4, |
| "healthy_threshold": 3, |
| "description": "test healthcheck", |
| } |
| hc = self.driver.ex_create_healthcheck(healthcheck_name, **kwargs) |
| self.assertTrue(isinstance(hc, GCEHealthCheck)) |
| self.assertEqual(hc.name, healthcheck_name) |
| self.assertEqual(hc.path, "/lc") |
| self.assertEqual(hc.port, 8000) |
| self.assertEqual(hc.interval, 10) |
| self.assertEqual(hc.extra["host"], "lchost") |
| self.assertEqual(hc.extra["description"], "test healthcheck") |
| |
| def test_ex_create_image(self): |
| volume = self.driver.ex_get_volume("lcdisk") |
| description = "CoreOS, CoreOS stable, 1520.6.0, amd64-usr published on 2017-10-12" |
| name = "coreos" |
| family = "coreos-stable" |
| licenses = ["projects/coreos-cloud/global/licenses/coreos-stable"] |
| guest_os_features = ["VIRTIO_SCSI_MULTIQUEUE"] |
| expected_features = [{"type": "VIRTIO_SCSI_MULTIQUEUE"}] |
| mock_request = mock.Mock() |
| mock_request.side_effect = self.driver.connection.async_request |
| self.driver.connection.async_request = mock_request |
| |
| image = self.driver.ex_create_image( |
| name, |
| volume, |
| description=description, |
| family=family, |
| guest_os_features=guest_os_features, |
| ex_licenses=licenses, |
| ) |
| self.assertTrue(isinstance(image, GCENodeImage)) |
| self.assertTrue(image.name.startswith(name)) |
| self.assertEqual(image.extra["description"], description) |
| self.assertEqual(image.extra["family"], family) |
| self.assertEqual(image.extra["guestOsFeatures"], expected_features) |
| self.assertEqual(image.extra["licenses"][0].name, licenses[0].split("/")[-1]) |
| expected_data = { |
| "description": description, |
| "family": family, |
| "guestOsFeatures": expected_features, |
| "name": name, |
| "licenses": licenses, |
| "sourceDisk": volume.extra["selfLink"], |
| "zone": volume.extra["zone"].name, |
| } |
| mock_request.assert_called_once_with("/global/images", data=expected_data, method="POST") |
| |
| def test_ex_copy_image(self): |
| name = "coreos" |
| url = "gs://storage.core-os.net/coreos/amd64-generic/247.0.0/coreos_production_gce.tar.gz" |
| description = "CoreOS, CoreOS stable, 1520.6.0, amd64-usr published on 2017-10-12" |
| family = "coreos-stable" |
| guest_os_features = ["VIRTIO_SCSI_MULTIQUEUE"] |
| expected_features = [{"type": "VIRTIO_SCSI_MULTIQUEUE"}] |
| image = self.driver.ex_copy_image( |
| name, |
| url, |
| description=description, |
| family=family, |
| guest_os_features=guest_os_features, |
| ) |
| self.assertTrue(image.name.startswith(name)) |
| self.assertEqual(image.extra["description"], description) |
| self.assertEqual(image.extra["family"], family) |
| self.assertEqual(image.extra["guestOsFeatures"], expected_features) |
| |
| def test_ex_create_firewall(self): |
| name = "lcfirewall" |
| priority = 900 |
| description = "Libcloud Test Firewall" |
| allowed = [{"IPProtocol": "tcp", "ports": ["4567"]}] |
| source_service_accounts = ["lcsource@gserviceaccount.com"] |
| target_tags = ["libcloud"] |
| network = "default" |
| firewall = self.driver.ex_create_firewall( |
| name, |
| allowed, |
| description=description, |
| network=network, |
| priority=priority, |
| target_tags=target_tags, |
| source_service_accounts=source_service_accounts, |
| ) |
| self.assertTrue(isinstance(firewall, GCEFirewall)) |
| self.assertEqual(firewall.name, name) |
| |
| def test_ex_create_firewall_egress(self): |
| name = "lcfirewall-egress" |
| priority = 900 |
| direction = "EGRESS" |
| description = "Libcloud Egress Firewall" |
| allowed = [{"IPProtocol": "tcp", "ports": ["4567"]}] |
| target_service_accounts = ["lctarget@gserviceaccount.com"] |
| target_ranges = ["8.8.8.8/32"] |
| network = "default" |
| firewall = self.driver.ex_create_firewall( |
| name, |
| allowed, |
| description=description, |
| network=network, |
| priority=priority, |
| direction=direction, |
| target_ranges=target_ranges, |
| target_service_accounts=target_service_accounts, |
| ) |
| self.assertTrue(isinstance(firewall, GCEFirewall)) |
| self.assertEqual(firewall.name, name) |
| |
| def test_ex_create_firewall_deny(self): |
| name = "lcfirewall-deny" |
| priority = 900 |
| denied = [{"IPProtocol": "tcp", "ports": ["4567"]}] |
| description = "Libcloud Deny Firewall" |
| source_ranges = ["10.240.100.0/24"] |
| source_tags = ["libcloud"] |
| network = "default" |
| firewall = self.driver.ex_create_firewall( |
| name, |
| denied=denied, |
| description=description, |
| network=network, |
| priority=priority, |
| source_tags=source_tags, |
| source_ranges=source_ranges, |
| ) |
| self.assertTrue(isinstance(firewall, GCEFirewall)) |
| self.assertEqual(firewall.name, name) |
| |
| def test_ex_create_forwarding_rule(self): |
| fwr_name = "lcforwardingrule" |
| targetpool = "lctargetpool" |
| region = "us-central1" |
| address = "lcaddress" |
| port_range = "8000-8500" |
| description = "test forwarding rule" |
| fwr = self.driver.ex_create_forwarding_rule( |
| fwr_name, |
| targetpool, |
| region=region, |
| address=address, |
| port_range=port_range, |
| description=description, |
| ) |
| self.assertTrue(isinstance(fwr, GCEForwardingRule)) |
| self.assertEqual(fwr.name, fwr_name) |
| self.assertEqual(fwr.region.name, region) |
| self.assertEqual(fwr.protocol, "TCP") |
| self.assertEqual(fwr.extra["portRange"], port_range) |
| self.assertEqual(fwr.extra["description"], description) |
| |
| def test_ex_create_forwarding_rule_global(self): |
| fwr_name = "http-rule" |
| target_name = "web-proxy" |
| address = "lcaddressglobal" |
| port_range = "80-80" |
| description = "global forwarding rule" |
| for target in (target_name, self.driver.ex_get_targethttpproxy(target_name)): |
| fwr = self.driver.ex_create_forwarding_rule( |
| fwr_name, |
| target, |
| global_rule=True, |
| address=address, |
| port_range=port_range, |
| description=description, |
| ) |
| self.assertTrue(isinstance(fwr, GCEForwardingRule)) |
| self.assertEqual(fwr.name, fwr_name) |
| self.assertEqual(fwr.extra["portRange"], port_range) |
| self.assertEqual(fwr.extra["description"], description) |
| |
| def test_ex_create_forwarding_rule_targetpool_keyword(self): |
| """Test backwards compatibility with the targetpool kwarg.""" |
| fwr_name = "lcforwardingrule" |
| targetpool = "lctargetpool" |
| region = "us-central1" |
| address = self.driver.ex_get_address("lcaddress") |
| port_range = "8000-8500" |
| description = "test forwarding rule" |
| fwr = self.driver.ex_create_forwarding_rule( |
| fwr_name, |
| targetpool=targetpool, |
| region=region, |
| address=address, |
| port_range=port_range, |
| description=description, |
| ) |
| self.assertTrue(isinstance(fwr, GCEForwardingRule)) |
| self.assertEqual(fwr.name, fwr_name) |
| self.assertEqual(fwr.region.name, region) |
| self.assertEqual(fwr.protocol, "TCP") |
| self.assertEqual(fwr.extra["portRange"], port_range) |
| self.assertEqual(fwr.extra["description"], description) |
| |
| def test_ex_create_route(self): |
| route_name = "lcdemoroute" |
| dest_range = "192.168.25.0/24" |
| priority = 1000 |
| route = self.driver.ex_create_route(route_name, dest_range) |
| self.assertTrue(isinstance(route, GCERoute)) |
| self.assertEqual(route.name, route_name) |
| self.assertEqual(route.priority, priority) |
| self.assertTrue("tag1" in route.tags) |
| self.assertTrue(route.extra["nextHopInstance"].endswith("libcloud-100")) |
| self.assertEqual(route.dest_range, dest_range) |
| |
| def test_ex_create_network(self): |
| network_name = "lcnetwork" |
| cidr = "10.11.0.0/16" |
| routing_mode = "REGIONAL" |
| network = self.driver.ex_create_network(network_name, cidr, routing_mode="regional") |
| self.assertTrue(isinstance(network, GCENetwork)) |
| self.assertEqual(network.name, network_name) |
| self.assertEqual(network.cidr, cidr) |
| |
| # Test using more options |
| description = "A custom network" |
| network = self.driver.ex_create_network( |
| network_name, cidr, description=description, routing_mode=routing_mode |
| ) |
| self.assertEqual(network.extra["description"], description) |
| self.assertEqual(network.extra["routingConfig"]["routingMode"], routing_mode) |
| |
| def test_ex_create_network_bad_options(self): |
| network_name = "lcnetwork" |
| cidr = "10.11.0.0/16" |
| self.assertRaises( |
| ValueError, self.driver.ex_create_network, network_name, cidr, mode="auto" |
| ) |
| self.assertRaises( |
| ValueError, self.driver.ex_create_network, network_name, cidr, mode="foobar" |
| ) |
| self.assertRaises( |
| ValueError, self.driver.ex_create_network, network_name, None, mode="legacy" |
| ) |
| self.assertRaises( |
| ValueError, |
| self.driver.ex_create_network, |
| network_name, |
| cidr, |
| routing_mode="universal", |
| ) |
| |
| def test_ex_set_machine_type_notstopped(self): |
| # get running node, change machine type |
| zone = "us-central1-a" |
| node = self.driver.ex_get_node("node-name", zone) |
| self.assertRaises(GoogleBaseError, self.driver.ex_set_machine_type, node, "custom-4-61440") |
| |
| def test_ex_set_machine_type(self): |
| # get stopped node, change machine type |
| zone = "us-central1-a" |
| node = self.driver.ex_get_node("stopped-node", zone) |
| self.assertEqual(node.size, "n1-standard-1") |
| self.assertEqual(node.extra["status"], "TERMINATED") |
| self.assertTrue(self.driver.ex_set_machine_type(node, "custom-4-11264")) |
| |
| def test_ex_node_start(self): |
| zone = "us-central1-a" |
| node = self.driver.ex_get_node("stopped-node", zone) |
| self.assertTrue(self.driver.ex_start_node(node)) |
| |
| def test_ex_node_stop(self): |
| zone = "us-central1-a" |
| node = self.driver.ex_get_node("node-name", zone) |
| self.assertTrue(self.driver.ex_stop_node(node)) |
| |
| # try and stop a stopped node (should work) |
| zone = "us-central1-a" |
| node = self.driver.ex_get_node("stopped-node", zone) |
| self.assertTrue(self.driver.ex_stop_node(node)) |
| |
| def test_create_node_req(self): |
| image = self.driver.ex_get_image("debian-7") |
| size = self.driver.ex_get_size("n1-standard-1") |
| location = self.driver.zone |
| network = self.driver.ex_get_network("default") |
| tags = ["libcloud"] |
| metadata = [{"key": "test_key", "value": "test_value"}] |
| boot_disk = self.driver.ex_get_volume("lcdisk") |
| node_request, node_data = self.driver._create_node_req( |
| "lcnode", size, image, location, network, tags, metadata, boot_disk |
| ) |
| self.assertEqual(node_request, "/zones/%s/instances" % location.name) |
| self.assertEqual(node_data["metadata"]["items"][0]["key"], "test_key") |
| self.assertEqual(node_data["tags"]["items"][0], "libcloud") |
| self.assertEqual(node_data["name"], "lcnode") |
| self.assertTrue(node_data["disks"][0]["boot"]) |
| self.assertIsInstance(node_data["serviceAccounts"], list) |
| self.assertIsInstance(node_data["serviceAccounts"][0], dict) |
| self.assertEqual(node_data["serviceAccounts"][0]["email"], "default") |
| self.assertIsInstance(node_data["serviceAccounts"][0]["scopes"], list) |
| self.assertEqual(len(node_data["serviceAccounts"][0]["scopes"]), 1) |
| self.assertEqual(len(node_data["networkInterfaces"]), 1) |
| self.assertTrue(node_data["networkInterfaces"][0]["network"].startswith("https://")) |
| |
| def test_create_node_location_not_provided(self): |
| # location is not specified neither via datacenter constructor argument |
| # nor via location method argument |
| node_name = "node-name" |
| size = self.driver.ex_get_size("n1-standard-1") |
| del size.extra["zone"] |
| image = self.driver.ex_get_image("debian-7") |
| self.driver.zone = None |
| |
| expected_msg = "Zone not provided" |
| self.assertRaisesRegex( |
| ValueError, expected_msg, self.driver.create_node, node_name, size, image |
| ) |
| |
| def test_create_node_location_not_provided_inferred_from_size(self): |
| # test a scenario where node location is inferred from NodeSize zone attribute |
| node_name = "node-name" |
| size = self.driver.ex_get_size("n1-standard-1") |
| image = self.driver.ex_get_image("debian-7") |
| zone = self.driver.ex_list_zones()[0] |
| zone = self.driver.ex_get_zone("us-central1-a") |
| |
| self.driver.zone = None |
| size.extra["zone"] = zone |
| |
| node = self.driver.create_node(node_name, size, image) |
| self.assertTrue(node) |
| |
| def test_create_node_network_opts(self): |
| node_name = "node-name" |
| size = self.driver.ex_get_size("n1-standard-1") |
| image = self.driver.ex_get_image("debian-7") |
| zone = self.driver.ex_get_zone("us-central1-a") |
| network = self.driver.ex_get_network("lcnetwork") |
| address = self.driver.ex_get_address("lcaddress") |
| ex_nic_gce_struct = [ |
| { |
| "network": "global/networks/lcnetwork", |
| "accessConfigs": [{"name": "lcnetwork-test", "type": "ONE_TO_ONE_NAT"}], |
| } |
| ] |
| # Test using default |
| node = self.driver.create_node(node_name, size, image) |
| self.assertEqual(node.extra["networkInterfaces"][0]["name"], "nic0") |
| |
| # Test using just the network |
| node = self.driver.create_node(node_name, size, image, location=zone, ex_network=network) |
| self.assertEqual(node.extra["networkInterfaces"][0]["name"], "nic0") |
| |
| # Test using just the struct |
| node = self.driver.create_node( |
| node_name, size, image, location=zone, ex_nic_gce_struct=ex_nic_gce_struct |
| ) |
| self.assertEqual(node.extra["networkInterfaces"][0]["name"], "nic0") |
| |
| # Test both address and struct, should fail |
| self.assertRaises( |
| ValueError, |
| self.driver.create_node, |
| node_name, |
| size, |
| image, |
| location=zone, |
| external_ip=address, |
| ex_nic_gce_struct=ex_nic_gce_struct, |
| ) |
| |
| # Test both ex_network and struct, should fail |
| self.assertRaises( |
| ValueError, |
| self.driver.create_node, |
| node_name, |
| size, |
| image, |
| location=zone, |
| ex_network=network, |
| ex_nic_gce_struct=ex_nic_gce_struct, |
| ) |
| |
| def test_create_node_subnetwork_opts(self): |
| node_name = "sn-node-name" |
| size = self.driver.ex_get_size("n1-standard-1") |
| image = self.driver.ex_get_image("debian-7") |
| zone = self.driver.ex_get_zone("us-central1-a") |
| network = self.driver.ex_get_network("custom-network") |
| subnetwork = self.driver.ex_get_subnetwork("cf-972cf02e6ad49112") |
| |
| ex_nic_gce_struct = [ |
| { |
| "network": "global/networks/custom-network", |
| "subnetwork": "projects/project_name/regions/us-central1/subnetworks/cf-972cf02e6ad49112", |
| "accessConfigs": [{"name": "External NAT", "type": "ONE_TO_ONE_NAT"}], |
| } |
| ] |
| # Test using just the network and subnetwork |
| node = self.driver.create_node( |
| node_name, |
| size, |
| image, |
| location=zone, |
| ex_network=network, |
| ex_subnetwork=subnetwork, |
| ) |
| self.assertEqual(node.extra["networkInterfaces"][0]["name"], "nic0") |
| self.assertEqual( |
| node.extra["networkInterfaces"][0]["subnetwork"].split("/")[-1], |
| "cf-972cf02e6ad49112", |
| ) |
| |
| # Test using just the struct |
| node = self.driver.create_node( |
| node_name, size, image, location=zone, ex_nic_gce_struct=ex_nic_gce_struct |
| ) |
| self.assertEqual(node.extra["networkInterfaces"][0]["name"], "nic0") |
| self.assertEqual( |
| node.extra["networkInterfaces"][0]["subnetwork"].split("/")[-1], |
| "cf-972cf02e6ad49112", |
| ) |
| |
| # Test using subnetwork selfLink |
| node = self.driver.create_node( |
| node_name, |
| size, |
| image, |
| location=zone, |
| ex_network=network, |
| ex_subnetwork=subnetwork.extra["selfLink"], |
| ) |
| self.assertEqual(node.extra["networkInterfaces"][0]["name"], "nic0") |
| self.assertEqual( |
| node.extra["networkInterfaces"][0]["subnetwork"].split("/")[-1], |
| "cf-972cf02e6ad49112", |
| ) |
| |
| def test_create_node_disk_opts(self): |
| node_name = "node-name" |
| size = self.driver.ex_get_size("n1-standard-1") |
| image = self.driver.ex_get_image("debian-7") |
| boot_disk = self.driver.ex_get_volume("lcdisk") |
| disk_type = self.driver.ex_get_disktype("pd-ssd", "us-central1-a") |
| DEMO_BASE_NAME = "lc-test" |
| gce_disk_struct = [ |
| { |
| "type": "PERSISTENT", |
| "deviceName": "%s-gstruct" % DEMO_BASE_NAME, |
| "initializeParams": { |
| "diskName": "%s-gstruct" % DEMO_BASE_NAME, |
| "sourceImage": image.extra["selfLink"], |
| }, |
| "boot": True, |
| "autoDelete": True, |
| }, |
| { |
| "type": "SCRATCH", |
| "deviceName": "%s-gstruct-lssd" % DEMO_BASE_NAME, |
| "initializeParams": {"diskType": disk_type.extra["selfLink"]}, |
| "autoDelete": True, |
| }, |
| ] |
| |
| self.assertRaises(ValueError, self.driver.create_node, node_name, size, None) |
| node = self.driver.create_node(node_name, size, image) |
| self.assertTrue(isinstance(node, Node)) |
| node = self.driver.create_node(node_name, size, None, ex_boot_disk=boot_disk) |
| self.assertTrue(isinstance(node, Node)) |
| node = self.driver.create_node(node_name, size, None, ex_disks_gce_struct=gce_disk_struct) |
| self.assertTrue(isinstance(node, Node)) |
| self.assertRaises( |
| ValueError, |
| self.driver.create_node, |
| node_name, |
| size, |
| None, |
| ex_boot_disk=boot_disk, |
| ex_disks_gce_struct=gce_disk_struct, |
| ) |
| |
| def test_create_node(self): |
| node_name = "node-name" |
| image = self.driver.ex_get_image("debian-7") |
| size = self.driver.ex_get_size("n1-standard-1") |
| node = self.driver.create_node(node_name, size, image) |
| self.assertTrue(isinstance(node, Node)) |
| self.assertEqual(node.name, node_name) |
| |
| def test_create_node_disk_size(self): |
| node_name = "node-name" |
| image = self.driver.ex_get_image("debian-7") |
| size = self.driver.ex_get_size("n1-standard-1") |
| disk_size = 25 |
| node = self.driver.create_node(node_name, size, image, ex_disk_size=disk_size) |
| self.assertTrue(isinstance(node, Node)) |
| self.assertEqual(node.name, node_name) |
| self.assertEqual(node.extra["boot_disk"].size, str(disk_size)) |
| |
| def test_create_node_image_family(self): |
| node_name = "node-name" |
| size = self.driver.ex_get_size("n1-standard-1") |
| node = self.driver.create_node(node_name, size, image=None, ex_image_family="coreos-stable") |
| self.assertTrue(isinstance(node, Node)) |
| self.assertEqual(node.name, node_name) |
| |
| image = self.driver.ex_get_image("debian-7") |
| self.assertRaises( |
| ValueError, |
| self.driver.create_node, |
| node_name, |
| size, |
| image, |
| ex_image_family="coreos-stable", |
| ) |
| |
| def test_create_node_req_with_serviceaccounts(self): |
| image = self.driver.ex_get_image("debian-7") |
| size = self.driver.ex_get_size("n1-standard-1") |
| location = self.driver.zone |
| network = self.driver.ex_get_network("default") |
| # ex_service_accounts with specific scopes, default 'email' |
| ex_sa = [{"scopes": ["compute-ro", "pubsub", "storage-ro"]}] |
| node_request, node_data = self.driver._create_node_req( |
| "lcnode", size, image, location, network, ex_service_accounts=ex_sa |
| ) |
| self.assertIsInstance(node_data["serviceAccounts"], list) |
| self.assertIsInstance(node_data["serviceAccounts"][0], dict) |
| self.assertEqual(node_data["serviceAccounts"][0]["email"], "default") |
| self.assertIsInstance(node_data["serviceAccounts"][0]["scopes"], list) |
| self.assertEqual(len(node_data["serviceAccounts"][0]["scopes"]), 3) |
| self.assertTrue( |
| "https://www.googleapis.com/auth/devstorage.read_only" |
| in node_data["serviceAccounts"][0]["scopes"] |
| ) |
| self.assertTrue( |
| "https://www.googleapis.com/auth/compute.readonly" |
| in node_data["serviceAccounts"][0]["scopes"] |
| ) |
| |
| def test_format_metadata(self): |
| in_md = [{"key": "k0", "value": "v0"}, {"key": "k1", "value": "v1"}] |
| out_md = self.driver._format_metadata("fp", in_md) |
| self.assertTrue("fingerprint" in out_md) |
| self.assertEqual(out_md["fingerprint"], "fp") |
| self.assertTrue("items" in out_md) |
| self.assertEqual(len(out_md["items"]), 2) |
| self.assertTrue(out_md["items"][0]["key"] in ["k0", "k1"]) |
| self.assertTrue(out_md["items"][0]["value"] in ["v0", "v1"]) |
| |
| in_md = [{"k0": "v0"}, {"k1": "v1"}] |
| out_md = self.driver._format_metadata("fp", in_md) |
| self.assertTrue("fingerprint" in out_md) |
| self.assertEqual(out_md["fingerprint"], "fp") |
| self.assertTrue("items" in out_md) |
| self.assertEqual(len(out_md["items"]), 2) |
| self.assertTrue(out_md["items"][0]["key"] in ["k0", "k1"]) |
| self.assertTrue(out_md["items"][0]["value"] in ["v0", "v1"]) |
| |
| in_md = {"key": "k0", "value": "v0"} |
| out_md = self.driver._format_metadata("fp", in_md) |
| self.assertTrue("fingerprint" in out_md) |
| self.assertEqual(out_md["fingerprint"], "fp") |
| self.assertTrue("items" in out_md) |
| self.assertEqual(len(out_md["items"]), 1, out_md) |
| self.assertEqual(out_md["items"][0]["key"], "k0") |
| self.assertEqual(out_md["items"][0]["value"], "v0") |
| |
| in_md = {"k0": "v0"} |
| out_md = self.driver._format_metadata("fp", in_md) |
| self.assertTrue("fingerprint" in out_md) |
| self.assertEqual(out_md["fingerprint"], "fp") |
| self.assertTrue("items" in out_md) |
| self.assertEqual(len(out_md["items"]), 1) |
| self.assertEqual(out_md["items"][0]["key"], "k0") |
| self.assertEqual(out_md["items"][0]["value"], "v0") |
| |
| in_md = {"k0": "v0", "k1": "v1", "k2": "v2"} |
| out_md = self.driver._format_metadata("fp", in_md) |
| self.assertTrue("fingerprint" in out_md) |
| self.assertEqual(out_md["fingerprint"], "fp") |
| self.assertTrue("items" in out_md) |
| self.assertEqual(len(out_md["items"]), 3) |
| keys = [x["key"] for x in out_md["items"]] |
| vals = [x["value"] for x in out_md["items"]] |
| keys.sort() |
| vals.sort() |
| self.assertEqual(keys, ["k0", "k1", "k2"]) |
| self.assertEqual(vals, ["v0", "v1", "v2"]) |
| |
| in_md = {"items": [{"key": "k0", "value": "v0"}, {"key": "k1", "value": "v1"}]} |
| out_md = self.driver._format_metadata("fp", in_md) |
| self.assertTrue("fingerprint" in out_md) |
| self.assertEqual(out_md["fingerprint"], "fp") |
| self.assertTrue("items" in out_md) |
| self.assertEqual(len(out_md["items"]), 2) |
| self.assertTrue(out_md["items"][0]["key"] in ["k0", "k1"]) |
| self.assertTrue(out_md["items"][0]["value"] in ["v0", "v1"]) |
| |
| in_md = {"items": "foo"} |
| self.assertRaises(ValueError, self.driver._format_metadata, "fp", in_md) |
| in_md = {"items": {"key": "k1", "value": "v0"}} |
| self.assertRaises(ValueError, self.driver._format_metadata, "fp", in_md) |
| in_md = ["k0", "v1"] |
| self.assertRaises(ValueError, self.driver._format_metadata, "fp", in_md) |
| |
| def test_create_node_with_metadata(self): |
| node_name = "node-name" |
| image = self.driver.ex_get_image("debian-7") |
| size = self.driver.ex_get_size("n1-standard-1") |
| zone = self.driver.ex_get_zone("us-central1-a") |
| |
| # md is a list of dicts, each with 'key' and 'value' for |
| # backwards compatibility |
| md = [{"key": "k0", "value": "v0"}, {"key": "k1", "value": "v1"}] |
| request, data = self.driver._create_node_req(node_name, size, image, zone, metadata=md) |
| self.assertTrue("items" in data["metadata"]) |
| self.assertEqual(len(data["metadata"]["items"]), 2) |
| |
| # md doesn't contain "items" key |
| md = {"key": "key1", "value": "value1"} |
| request, data = self.driver._create_node_req(node_name, size, image, zone, metadata=md) |
| self.assertTrue("items" in data["metadata"]) |
| self.assertEqual(len(data["metadata"]["items"]), 1) |
| |
| # md contains "items" key |
| md = {"items": [{"key": "k0", "value": "v0"}]} |
| request, data = self.driver._create_node_req(node_name, size, image, zone, metadata=md) |
| self.assertTrue("items" in data["metadata"]) |
| self.assertEqual(len(data["metadata"]["items"]), 1) |
| self.assertEqual(data["metadata"]["items"][0]["key"], "k0") |
| self.assertEqual(data["metadata"]["items"][0]["value"], "v0") |
| |
| def test_create_node_with_accelerator(self): |
| node_name = "node-name" |
| image = self.driver.ex_get_image("debian-7") |
| size = self.driver.ex_get_size("n1-standard-1") |
| zone = self.driver.ex_get_zone("us-central1-a") |
| request, data = self.driver._create_node_req( |
| node_name, |
| size, |
| image, |
| zone, |
| ex_accelerator_type="nvidia-tesla-k80", |
| ex_accelerator_count=3, |
| ) |
| self.assertTrue("guestAccelerators" in data) |
| self.assertEqual(len(data["guestAccelerators"]), 1) |
| self.assertTrue("nvidia-tesla-k80" in data["guestAccelerators"][0]["acceleratorType"]) |
| self.assertEqual(data["guestAccelerators"][0]["acceleratorCount"], 3) |
| |
| def test_create_node_with_labels(self): |
| node_name = "node-name" |
| image = self.driver.ex_get_image("debian-7") |
| size = self.driver.ex_get_size("n1-standard-1") |
| zone = self.driver.ex_get_zone("us-central1-a") |
| |
| # labels is a dict |
| labels = {"label1": "v1", "label2": "v2"} |
| request, data = self.driver._create_node_req(node_name, size, image, zone, ex_labels=labels) |
| self.assertTrue(data["labels"] is not None) |
| self.assertEqual(len(data["labels"]), 2) |
| self.assertEqual(data["labels"]["label1"], "v1") |
| self.assertEqual(data["labels"]["label2"], "v2") |
| |
| def test_create_node_existing(self): |
| node_name = "libcloud-demo-europe-np-node" |
| image = self.driver.ex_get_image("debian-7") |
| size = self.driver.ex_get_size("n1-standard-1", zone="europe-west1-a") |
| self.assertRaises( |
| ResourceExistsError, |
| self.driver.create_node, |
| node_name, |
| size, |
| image, |
| location="europe-west1-a", |
| ) |
| |
| def test_ex_create_multiple_nodes(self): |
| base_name = "lcnode" |
| image = self.driver.ex_get_image("debian-7") |
| size = self.driver.ex_get_size("n1-standard-1") |
| number = 2 |
| disk_size = "25" |
| # NOTE: We use small poll_interval to speed up the tests |
| nodes = self.driver.ex_create_multiple_nodes( |
| base_name, size, image, number, ex_disk_size=disk_size, poll_interval=0.1 |
| ) |
| self.assertEqual(len(nodes), 2) |
| self.assertTrue(isinstance(nodes[0], Node)) |
| self.assertTrue(isinstance(nodes[1], Node)) |
| self.assertEqual(nodes[0].name, "%s-000" % base_name) |
| self.assertEqual(nodes[1].name, "%s-001" % base_name) |
| self.assertEqual(nodes[0].extra["boot_disk"].size, disk_size) |
| self.assertEqual(nodes[1].extra["boot_disk"].size, disk_size) |
| |
| def test_ex_create_multiple_nodes_image_family(self): |
| base_name = "lcnode" |
| image = None |
| size = self.driver.ex_get_size("n1-standard-1") |
| number = 2 |
| # NOTE: We use small poll_interval to speed up the tests |
| nodes = self.driver.ex_create_multiple_nodes( |
| base_name, |
| size, |
| image, |
| number, |
| ex_image_family="coreos-stable", |
| poll_interval=0.1, |
| ) |
| self.assertEqual(len(nodes), 2) |
| self.assertTrue(isinstance(nodes[0], Node)) |
| self.assertTrue(isinstance(nodes[1], Node)) |
| self.assertEqual(nodes[0].name, "%s-000" % base_name) |
| self.assertEqual(nodes[1].name, "%s-001" % base_name) |
| |
| image = self.driver.ex_get_image("debian-7") |
| self.assertRaises( |
| ValueError, |
| self.driver.ex_create_multiple_nodes, |
| base_name, |
| size, |
| image, |
| number, |
| ex_image_family="coreos-stable", |
| ) |
| |
| def test_ex_create_targethttpproxy(self): |
| proxy_name = "web-proxy" |
| urlmap_name = "web-map" |
| for urlmap in (urlmap_name, self.driver.ex_get_urlmap(urlmap_name)): |
| proxy = self.driver.ex_create_targethttpproxy(proxy_name, urlmap) |
| self.assertTrue(isinstance(proxy, GCETargetHttpProxy)) |
| self.assertEqual(proxy_name, proxy.name) |
| |
| def test_ex_create_targetinstance(self): |
| targetinstance_name = "lctargetinstance" |
| zone = "us-central1-a" |
| node = self.driver.ex_get_node("node-name", zone) |
| targetinstance = self.driver.ex_create_targetinstance( |
| targetinstance_name, zone=zone, node=node |
| ) |
| self.assertEqual(targetinstance.name, targetinstance_name) |
| self.assertEqual(targetinstance.zone.name, zone) |
| |
| def test_ex_create_targetpool(self): |
| targetpool_name = "lctargetpool" |
| region = "us-central1" |
| healthchecks = ["libcloud-lb-demo-healthcheck"] |
| node1 = self.driver.ex_get_node("libcloud-lb-demo-www-000", "us-central1-b") |
| node2 = self.driver.ex_get_node("libcloud-lb-demo-www-001", "us-central1-b") |
| nodes = [node1, node2] |
| targetpool = self.driver.ex_create_targetpool( |
| targetpool_name, region=region, healthchecks=healthchecks, nodes=nodes |
| ) |
| self.assertEqual(targetpool.name, targetpool_name) |
| self.assertEqual(len(targetpool.nodes), len(nodes)) |
| self.assertEqual(targetpool.region.name, region) |
| |
| def test_ex_create_targetpool_session_affinity(self): |
| targetpool_name = "lctargetpool-sticky" |
| region = "us-central1" |
| session_affinity = "CLIENT_IP_PROTO" |
| targetpool = self.driver.ex_create_targetpool( |
| targetpool_name, region=region, session_affinity=session_affinity |
| ) |
| self.assertEqual(targetpool.name, targetpool_name) |
| self.assertEqual(targetpool.extra.get("sessionAffinity"), session_affinity) |
| |
| def test_ex_create_urlmap(self): |
| urlmap_name = "web-map" |
| for service in ( |
| "web-service", |
| self.driver.ex_get_backendservice("web-service"), |
| ): |
| urlmap = self.driver.ex_create_urlmap(urlmap_name, service) |
| self.assertTrue(isinstance(urlmap, GCEUrlMap)) |
| self.assertEqual(urlmap_name, urlmap.name) |
| |
| def test_create_volume_image_family(self): |
| volume_name = "lcdisk" |
| size = 10 |
| volume = self.driver.create_volume(size, volume_name, ex_image_family="coreos-stable") |
| self.assertTrue(isinstance(volume, StorageVolume)) |
| self.assertEqual(volume.name, volume_name) |
| |
| image = self.driver.ex_get_image("debian-7") |
| self.assertRaises( |
| ValueError, |
| self.driver.create_volume, |
| size, |
| volume_name, |
| image=image, |
| ex_image_family="coreos-stable", |
| ) |
| |
| def test_create_volume_location(self): |
| volume_name = "lcdisk" |
| size = 10 |
| zone = self.driver.zone |
| volume = self.driver.create_volume(size, volume_name, location=zone) |
| self.assertTrue(isinstance(volume, StorageVolume)) |
| self.assertEqual(volume.name, volume_name) |
| |
| def test_ex_create_volume_snapshot(self): |
| snapshot_name = "lcsnapshot" |
| volume = self.driver.ex_get_volume("lcdisk") |
| snapshot = volume.snapshot(snapshot_name) |
| |
| self.assertEqual(snapshot.name, snapshot_name) |
| self.assertEqual(snapshot.size, "10") |
| |
| def test_create_volume_ssd(self): |
| volume_name = "lcdisk" |
| size = 10 |
| volume = self.driver.create_volume(size, volume_name, ex_disk_type="pd-ssd") |
| self.assertTrue(isinstance(volume, StorageVolume)) |
| self.assertEqual(volume.extra["type"], "pd-ssd") |
| |
| def test_create_volume(self): |
| volume_name = "lcdisk" |
| size = 10 |
| volume = self.driver.create_volume(size, volume_name) |
| self.assertTrue(isinstance(volume, StorageVolume)) |
| self.assertEqual(volume.name, volume_name) |
| |
| def test_ex_set_volume_labels(self): |
| volume_name = "lcdisk" |
| zone = self.driver.zone |
| volume_labels = {"one": "1", "two": "2", "three": "3"} |
| size = 10 |
| new_vol = self.driver.create_volume(size, volume_name, location=zone) |
| self.assertTrue(self.driver.ex_set_volume_labels(new_vol, labels=volume_labels)) |
| exist_vol = self.driver.ex_get_volume(volume_name, self.driver.zone) |
| self.assertEqual(exist_vol.extra["labels"], volume_labels) |
| |
| def test_ex_update_healthcheck(self): |
| healthcheck_name = "lchealthcheck" |
| healthcheck = self.driver.ex_get_healthcheck(healthcheck_name) |
| healthcheck.port = 9000 |
| healthcheck2 = self.driver.ex_update_healthcheck(healthcheck) |
| self.assertTrue(isinstance(healthcheck2, GCEHealthCheck)) |
| |
| def test_ex_update_firewall(self): |
| firewall_name = "lcfirewall" |
| firewall = self.driver.ex_get_firewall(firewall_name) |
| firewall.source_ranges = ["10.0.0.0/16"] |
| firewall.description = "LCFirewall-2" |
| firewall2 = self.driver.ex_update_firewall(firewall) |
| self.assertTrue(isinstance(firewall2, GCEFirewall)) |
| |
| def test_ex_targetpool_gethealth(self): |
| targetpool = self.driver.ex_get_targetpool("lb-pool") |
| health = targetpool.get_health("libcloud-lb-demo-www-000") |
| self.assertEqual(len(health), 1) |
| self.assertTrue("node" in health[0]) |
| self.assertTrue("health" in health[0]) |
| self.assertEqual(health[0]["health"], "UNHEALTHY") |
| |
| def test_ex_targetpool_with_backup_pool(self): |
| targetpool = self.driver.ex_get_targetpool("lb-pool") |
| self.assertTrue("backupPool" in targetpool.extra) |
| self.assertTrue("failoverRatio" in targetpool.extra) |
| |
| def test_ex_targetpool_setbackup(self): |
| targetpool = self.driver.ex_get_targetpool("lb-pool") |
| backup_targetpool = self.driver.ex_get_targetpool("backup-pool") |
| self.assertTrue(targetpool.set_backup_targetpool(backup_targetpool, 0.1)) |
| |
| def test_ex_targetpool_remove_add_node(self): |
| targetpool = self.driver.ex_get_targetpool("lctargetpool") |
| node = self.driver.ex_get_node("libcloud-lb-demo-www-001", "us-central1-b") |
| remove_node = self.driver.ex_targetpool_remove_node(targetpool, node) |
| self.assertTrue(remove_node) |
| self.assertEqual(len(targetpool.nodes), 1) |
| |
| add_node = self.driver.ex_targetpool_add_node(targetpool, node.extra["selfLink"]) |
| self.assertTrue(add_node) |
| self.assertEqual(len(targetpool.nodes), 2) |
| |
| remove_node = self.driver.ex_targetpool_remove_node(targetpool, node.extra["selfLink"]) |
| self.assertTrue(remove_node) |
| self.assertEqual(len(targetpool.nodes), 1) |
| |
| add_node = self.driver.ex_targetpool_add_node(targetpool, node) |
| self.assertTrue(add_node) |
| self.assertEqual(len(targetpool.nodes), 2) |
| |
| # check that duplicates are filtered |
| add_node = self.driver.ex_targetpool_add_node(targetpool, node.extra["selfLink"]) |
| self.assertTrue(add_node) |
| self.assertEqual(len(targetpool.nodes), 2) |
| |
| def test_ex_targetpool_remove_add_healthcheck(self): |
| targetpool = self.driver.ex_get_targetpool("lctargetpool") |
| healthcheck = self.driver.ex_get_healthcheck("libcloud-lb-demo-healthcheck") |
| remove_healthcheck = self.driver.ex_targetpool_remove_healthcheck(targetpool, healthcheck) |
| self.assertTrue(remove_healthcheck) |
| self.assertEqual(len(targetpool.healthchecks), 0) |
| |
| add_healthcheck = self.driver.ex_targetpool_add_healthcheck(targetpool, healthcheck) |
| self.assertTrue(add_healthcheck) |
| self.assertEqual(len(targetpool.healthchecks), 1) |
| |
| def test_reboot_node(self): |
| node = self.driver.ex_get_node("node-name") |
| reboot = self.driver.reboot_node(node) |
| self.assertTrue(reboot) |
| |
| def test_ex_set_node_tags(self): |
| new_tags = ["libcloud"] |
| node = self.driver.ex_get_node("node-name") |
| set_tags = self.driver.ex_set_node_tags(node, new_tags) |
| self.assertTrue(set_tags) |
| |
| def test_attach_volume_invalid_usecase(self): |
| node = self.driver.ex_get_node("node-name") |
| self.assertRaises(ValueError, self.driver.attach_volume, node, None) |
| self.assertRaises( |
| ValueError, |
| self.driver.attach_volume, |
| node, |
| None, |
| ex_source="foo/bar", |
| device=None, |
| ) |
| |
| def test_attach_volume(self): |
| volume = self.driver.ex_get_volume("lcdisk") |
| node = self.driver.ex_get_node("node-name") |
| attach = volume.attach(node) |
| self.assertTrue(attach) |
| |
| def test_ex_resize_volume(self): |
| volume = self.driver.ex_get_volume("lcdisk") |
| desired_size = int(volume.size) + 8 |
| resize = self.driver.ex_resize_volume(volume, desired_size) |
| self.assertTrue(resize) |
| |
| def test_detach_volume(self): |
| volume = self.driver.ex_get_volume("lcdisk") |
| node = self.driver.ex_get_node("node-name") |
| # This fails since the node is required |
| detach = volume.detach() |
| self.assertFalse(detach) |
| # This should pass |
| detach = self.driver.detach_volume(volume, node) |
| self.assertTrue(detach) |
| |
| def test_ex_destroy_address_global(self): |
| address = self.driver.ex_get_address("lcaddressglobal", "global") |
| self.assertEqual(address.name, "lcaddressglobal") |
| self.assertEqual(address.region, "global") |
| destroyed = address.destroy() |
| self.assertTrue(destroyed) |
| |
| def test_ex_destroy_address(self): |
| address = self.driver.ex_get_address("lcaddress") |
| destroyed = address.destroy() |
| self.assertTrue(destroyed) |
| |
| def test_ex_destroy_backendservice(self): |
| backendservice = self.driver.ex_get_backendservice("web-service") |
| destroyed = backendservice.destroy() |
| self.assertTrue(destroyed) |
| |
| def test_ex_destroy_healthcheck(self): |
| hc = self.driver.ex_get_healthcheck("lchealthcheck") |
| destroyed = hc.destroy() |
| self.assertTrue(destroyed) |
| |
| def test_ex_delete_image(self): |
| self.assertRaises(ResourceNotFoundError, self.driver.ex_get_image, "missing-image") |
| self.assertRaises(ResourceNotFoundError, self.driver.ex_delete_image, "missing-image") |
| |
| image = self.driver.ex_get_image("debian-7") |
| deleted = self.driver.ex_delete_image(image) |
| self.assertTrue(deleted) |
| |
| def test_ex_deprecate_image(self): |
| dep_ts = "2064-03-11T20:18:36.194-07:00" |
| obs_ts = "2074-03-11T20:18:36.194-07:00" |
| del_ts = "2084-03-11T20:18:36.194-07:00" |
| image = self.driver.ex_get_image("debian-7-wheezy-v20131014") |
| deprecated = image.deprecate( |
| "debian-7", "DEPRECATED", deprecated=dep_ts, obsolete=obs_ts, deleted=del_ts |
| ) |
| self.assertTrue(deprecated) |
| self.assertEqual(image.extra["deprecated"]["deprecated"], dep_ts) |
| self.assertEqual(image.extra["deprecated"]["obsolete"], obs_ts) |
| self.assertEqual(image.extra["deprecated"]["deleted"], del_ts) |
| |
| def test_ex_destroy_firewall(self): |
| firewall = self.driver.ex_get_firewall("lcfirewall") |
| destroyed = firewall.destroy() |
| self.assertTrue(destroyed) |
| |
| def test_ex_destroy_forwarding_rule(self): |
| fwr = self.driver.ex_get_forwarding_rule("lcforwardingrule") |
| destroyed = fwr.destroy() |
| self.assertTrue(destroyed) |
| |
| def test_ex_destroy_forwarding_rule_global(self): |
| fwr = self.driver.ex_get_forwarding_rule("http-rule", global_rule=True) |
| destroyed = fwr.destroy() |
| self.assertTrue(destroyed) |
| |
| def test_ex_destroy_route(self): |
| route = self.driver.ex_get_route("lcdemoroute") |
| destroyed = route.destroy() |
| self.assertTrue(destroyed) |
| |
| def test_ex_destroy_network(self): |
| network = self.driver.ex_get_network("lcnetwork") |
| destroyed = network.destroy() |
| self.assertTrue(destroyed) |
| |
| def test_destroy_node(self): |
| node = self.driver.ex_get_node("node-name") |
| destroyed = node.destroy() |
| self.assertTrue(destroyed) |
| |
| def test_ex_destroy_multiple_nodes(self): |
| nodes = [] |
| nodes.append(self.driver.ex_get_node("lcnode-000")) |
| nodes.append(self.driver.ex_get_node("lcnode-001")) |
| destroyed = self.driver.ex_destroy_multiple_nodes(nodes) |
| for d in destroyed: |
| self.assertTrue(d) |
| |
| def test_destroy_targethttpproxy(self): |
| proxy = self.driver.ex_get_targethttpproxy("web-proxy") |
| destroyed = proxy.destroy() |
| self.assertTrue(destroyed) |
| |
| def test_destroy_targetinstance(self): |
| targetinstance = self.driver.ex_get_targetinstance("lctargetinstance") |
| self.assertEqual(targetinstance.name, "lctargetinstance") |
| destroyed = targetinstance.destroy() |
| self.assertTrue(destroyed) |
| |
| def test_destroy_targetpool(self): |
| targetpool = self.driver.ex_get_targetpool("lctargetpool") |
| destroyed = targetpool.destroy() |
| self.assertTrue(destroyed) |
| |
| def test_destroy_urlmap(self): |
| urlmap = self.driver.ex_get_urlmap("web-map") |
| destroyed = urlmap.destroy() |
| self.assertTrue(destroyed) |
| |
| def test_destroy_volume(self): |
| disk = self.driver.ex_get_volume("lcdisk") |
| destroyed = disk.destroy() |
| self.assertTrue(destroyed) |
| |
| def test_ex_set_volume_auto_delete(self): |
| node = self.driver.ex_get_node("node-name") |
| volume = node.extra["boot_disk"] |
| auto_delete = self.driver.ex_set_volume_auto_delete(volume, node) |
| self.assertTrue(auto_delete) |
| |
| def test_destroy_volume_snapshot(self): |
| snapshot = self.driver.ex_get_snapshot("lcsnapshot") |
| destroyed = snapshot.destroy() |
| self.assertTrue(destroyed) |
| |
| def test_ex_get_address_global(self): |
| address_name = "lcaddressglobal" |
| address = self.driver.ex_get_address(address_name, "global") |
| self.assertEqual(address.name, address_name) |
| self.assertEqual(address.address, "173.99.99.99") |
| self.assertEqual(address.region, "global") |
| self.assertEqual(address.extra["status"], "RESERVED") |
| |
| def test_ex_get_address(self): |
| address_name = "lcaddress" |
| address = self.driver.ex_get_address(address_name) |
| self.assertEqual(address.name, address_name) |
| self.assertEqual(address.address, "173.255.113.20") |
| self.assertEqual(address.region.name, "us-central1") |
| self.assertEqual(address.extra["status"], "RESERVED") |
| |
| def test_ex_get_backendservice(self): |
| web_service = self.driver.ex_get_backendservice("web-service") |
| self.assertEqual(web_service.name, "web-service") |
| self.assertEqual(web_service.protocol, "HTTP") |
| self.assertEqual(web_service.port, 80) |
| self.assertEqual(web_service.timeout, 30) |
| self.assertEqual(web_service.healthchecks[0].name, "basic-check") |
| self.assertEqual(len(web_service.healthchecks), 1) |
| backends = web_service.backends |
| self.assertEqual(len(backends), 2) |
| self.assertEqual(backends[0]["balancingMode"], "RATE") |
| self.assertEqual(backends[0]["maxRate"], 100) |
| self.assertEqual(backends[0]["capacityScaler"], 1.0) |
| |
| web_service = self.driver.ex_get_backendservice("no-backends") |
| self.assertEqual(web_service.name, "web-service") |
| self.assertEqual(web_service.healthchecks[0].name, "basic-check") |
| self.assertEqual(len(web_service.healthchecks), 1) |
| self.assertEqual(len(web_service.backends), 0) |
| |
| def test_ex_get_healthcheck(self): |
| healthcheck_name = "lchealthcheck" |
| healthcheck = self.driver.ex_get_healthcheck(healthcheck_name) |
| self.assertEqual(healthcheck.name, healthcheck_name) |
| self.assertEqual(healthcheck.port, 8000) |
| self.assertEqual(healthcheck.path, "/lc") |
| |
| def test_ex_get_firewall(self): |
| firewall_name = "lcfirewall" |
| firewall = self.driver.ex_get_firewall(firewall_name) |
| self.assertEqual(firewall.name, firewall_name) |
| self.assertEqual(firewall.network.name, "default") |
| self.assertEqual(firewall.target_tags, ["libcloud"]) |
| |
| def test_ex_get_forwarding_rule(self): |
| fwr_name = "lcforwardingrule" |
| fwr = self.driver.ex_get_forwarding_rule(fwr_name) |
| self.assertEqual(fwr.name, fwr_name) |
| self.assertEqual(fwr.extra["portRange"], "8000-8500") |
| self.assertEqual(fwr.targetpool.name, "lctargetpool") |
| self.assertEqual(fwr.protocol, "TCP") |
| |
| def test_ex_get_forwarding_rule_global(self): |
| fwr_name = "http-rule" |
| fwr = self.driver.ex_get_forwarding_rule(fwr_name, global_rule=True) |
| self.assertEqual(fwr.name, fwr_name) |
| self.assertEqual(fwr.extra["portRange"], "80-80") |
| self.assertEqual(fwr.targetpool.name, "web-proxy") |
| self.assertEqual(fwr.protocol, "TCP") |
| self.assertEqual(fwr.address, "192.0.2.1") |
| self.assertEqual(fwr.targetpool.name, "web-proxy") |
| |
| def test_ex_get_image_license(self): |
| image = self.driver.ex_get_image("sles-12-v20141023") |
| self.assertTrue("licenses" in image.extra) |
| self.assertEqual(image.extra["licenses"][0].name, "sles-12") |
| self.assertTrue(image.extra["licenses"][0].charges_use_fee) |
| |
| def test_ex_get_image(self): |
| partial_name = "debian-7" |
| image = self.driver.ex_get_image(partial_name) |
| self.assertEqual(image.name, "debian-7-wheezy-v20131120") |
| # A 'debian-7' image exists in the local project |
| self.assertTrue(image.extra["description"].startswith("Debian")) |
| |
| partial_name = "debian-6" |
| image = self.driver.ex_get_image(partial_name) |
| self.assertEqual(image.name, "debian-6-squeeze-v20130926") |
| self.assertTrue(image.extra["description"].startswith("Debian")) |
| |
| partial_name = "debian-7" |
| image = self.driver.ex_get_image(partial_name, ["debian-cloud"]) |
| self.assertEqual(image.name, "debian-7-wheezy-v20131120") |
| |
| partial_name = "debian-7" |
| self.assertRaises( |
| ResourceNotFoundError, |
| self.driver.ex_get_image, |
| partial_name, |
| "suse-cloud", |
| ex_standard_projects=False, |
| ) |
| |
| def test_ex_get_image_from_family(self): |
| family = "coreos-beta" |
| description = "CoreOS beta 522.3.0" |
| image = self.driver.ex_get_image_from_family(family) |
| self.assertEqual(image.name, "coreos-beta-522-3-0-v20141226") |
| self.assertEqual(image.extra["description"], description) |
| self.assertEqual(image.extra["family"], family) |
| |
| url = ( |
| "https://www.googleapis.com/compute/v1/projects/coreos-cloud/" |
| "global/images/family/coreos-beta" |
| ) |
| image = self.driver.ex_get_image_from_family(url) |
| self.assertEqual(image.name, "coreos-beta-522-3-0-v20141226") |
| self.assertEqual(image.extra["description"], description) |
| self.assertEqual(image.extra["family"], family) |
| |
| project_list = ["coreos-cloud"] |
| image = self.driver.ex_get_image_from_family( |
| family, ex_project_list=project_list, ex_standard_projects=False |
| ) |
| self.assertEqual(image.name, "coreos-beta-522-3-0-v20141226") |
| self.assertEqual(image.extra["description"], description) |
| self.assertEqual(image.extra["family"], family) |
| |
| self.assertRaises(ResourceNotFoundError, self.driver.ex_get_image_from_family, "nofamily") |
| |
| def test_ex_get_route(self): |
| route_name = "lcdemoroute" |
| route = self.driver.ex_get_route(route_name) |
| self.assertEqual(route.name, route_name) |
| self.assertEqual(route.dest_range, "192.168.25.0/24") |
| self.assertEqual(route.priority, 1000) |
| |
| def test_ex_get_network(self): |
| network_name = "lcnetwork" |
| network = self.driver.ex_get_network(network_name) |
| self.assertEqual(network.name, network_name) |
| self.assertEqual(network.cidr, "10.11.0.0/16") |
| self.assertEqual(network.extra["gatewayIPv4"], "10.11.0.1") |
| self.assertEqual(network.extra["description"], "A custom network") |
| # do the same but this time with URL |
| url = ( |
| "https://www.googleapis.com/compute/v1/projects/project_name/global/networks/lcnetwork" |
| ) |
| network = self.driver.ex_get_network(url) |
| self.assertEqual(network.name, network_name) |
| self.assertEqual(network.cidr, "10.11.0.0/16") |
| self.assertEqual(network.extra["gatewayIPv4"], "10.11.0.1") |
| self.assertEqual(network.extra["description"], "A custom network") |
| # do the same but with a network under a different project |
| url_other = ( |
| "https://www.googleapis.com/compute/v1/projects/other_name/global/networks/lcnetwork" |
| ) |
| network = self.driver.ex_get_network(url_other) |
| self.assertEqual(network.name, network_name) |
| self.assertEqual(network.cidr, "10.11.0.0/16") |
| self.assertEqual(network.extra["gatewayIPv4"], "10.11.0.1") |
| self.assertEqual(network.extra["description"], "A custom network") |
| |
| def test_ex_get_node(self): |
| node_name = "node-name" |
| zone = "us-central1-a" |
| node = self.driver.ex_get_node(node_name, zone) |
| self.assertEqual(node.name, node_name) |
| self.assertEqual(node.size, "n1-standard-1") |
| removed_node = "libcloud-lb-demo-www-002" |
| self.assertRaises( |
| ResourceNotFoundError, |
| self.driver.ex_get_node, |
| removed_node, |
| "us-central1-b", |
| ) |
| missing_node = "dummy-node" |
| self.assertRaises(ResourceNotFoundError, self.driver.ex_get_node, missing_node, "all") |
| |
| def test_ex_get_project(self): |
| project = self.driver.ex_get_project() |
| self.assertEqual(project.name, "project_name") |
| networks_quota = project.quotas[1] |
| self.assertEqual(networks_quota["usage"], 3) |
| self.assertEqual(networks_quota["limit"], 5) |
| self.assertEqual(networks_quota["metric"], "NETWORKS") |
| self.assertTrue("fingerprint" in project.extra["commonInstanceMetadata"]) |
| self.assertTrue("items" in project.extra["commonInstanceMetadata"]) |
| self.assertTrue("usageExportLocation" in project.extra) |
| self.assertTrue("bucketName" in project.extra["usageExportLocation"]) |
| self.assertEqual( |
| project.extra["usageExportLocation"]["bucketName"], |
| "gs://graphite-usage-reports", |
| ) |
| |
| def test_ex_add_access_config(self): |
| self.assertRaises(ValueError, self.driver.ex_add_access_config, "node", "name", "nic") |
| node = self.driver.ex_get_node("node-name", "us-central1-a") |
| self.assertTrue(self.driver.ex_add_access_config(node, "foo", "bar")) |
| |
| def test_ex_delete_access_config(self): |
| self.assertRaises(ValueError, self.driver.ex_add_access_config, "node", "name", "nic") |
| node = self.driver.ex_get_node("node-name", "us-central1-a") |
| self.assertTrue(self.driver.ex_delete_access_config(node, "foo", "bar")) |
| |
| def test_ex_set_usage_export_bucket(self): |
| self.assertRaises(ValueError, self.driver.ex_set_usage_export_bucket, "foo") |
| bucket_name = "gs://foo" |
| self.driver.ex_set_usage_export_bucket(bucket_name) |
| |
| bucket_name = "https://www.googleapis.com/foo" |
| self.driver.ex_set_usage_export_bucket(bucket_name) |
| |
| project = GCEProject(id=None, name=None, metadata=None, quotas=None, driver=self.driver) |
| project.set_usage_export_bucket(bucket=bucket_name) |
| |
| def test__set_project_metadata(self): |
| self.assertEqual(len(self.driver._set_project_metadata(None, False, "")), 0) |
| |
| # 'delete' metadata, but retain current sshKeys |
| md = self.driver._set_project_metadata(None, False, "this is a test") |
| self.assertEqual(len(md), 1) |
| self.assertEqual(md[0]["key"], "sshKeys") |
| self.assertEqual(md[0]["value"], "this is a test") |
| |
| # 'delete' metadata *and* any existing sshKeys |
| md = self.driver._set_project_metadata(None, True, "this is a test") |
| self.assertEqual(len(md), 0) |
| |
| # add new metadata, keep existing sshKeys, since the new value also |
| # has 'sshKeys', we want the final struct to only have one ke/value |
| # of sshKeys and it should be the "current_keys" |
| gce_md = { |
| "items": [ |
| {"key": "foo", "value": "one"}, |
| {"key": "sshKeys", "value": "another test"}, |
| ] |
| } |
| md = self.driver._set_project_metadata(gce_md, False, "this is a test") |
| self.assertEqual(len(md), 2, str(md)) |
| sshKeys = "" |
| count = 0 |
| for d in md: |
| if d["key"] == "sshKeys": |
| count += 1 |
| sshKeys = d["value"] |
| self.assertEqual(sshKeys, "this is a test") |
| self.assertEqual(count, 1) |
| |
| # add new metadata, overwrite existing sshKeys, in this case, the |
| # existing 'sshKeys' value should be replaced |
| gce_md = { |
| "items": [ |
| {"key": "foo", "value": "one"}, |
| {"key": "sshKeys", "value": "another test"}, |
| ] |
| } |
| md = self.driver._set_project_metadata(gce_md, True, "this is a test") |
| self.assertEqual(len(md), 2, str(md)) |
| sshKeys = "" |
| count = 0 |
| for d in md: |
| if d["key"] == "sshKeys": |
| count += 1 |
| sshKeys = d["value"] |
| self.assertEqual(sshKeys, "another test") |
| self.assertEqual(count, 1) |
| |
| # add new metadata, remove existing sshKeys. in this case, we had an |
| # 'sshKeys' entry, but it will be removed entirely |
| gce_md = {"items": [{"key": "foo", "value": "one"}, {"key": "nokeys", "value": "two"}]} |
| md = self.driver._set_project_metadata(gce_md, True, "this is a test") |
| self.assertEqual(len(md), 2, str(md)) |
| sshKeys = "" |
| count = 0 |
| for d in md: |
| if d["key"] == "sshKeys": |
| count += 1 |
| sshKeys = d["value"] |
| self.assertEqual(sshKeys, "") |
| self.assertEqual(count, 0) |
| |
| def test_ex_set_common_instance_metadata(self): |
| # test non-dict |
| self.assertRaises(ValueError, self.driver.ex_set_common_instance_metadata, ["bad", "type"]) |
| # test standard python dict |
| pydict = {"key": "pydict", "value": 1} |
| self.driver.ex_set_common_instance_metadata(pydict) |
| # test GCE badly formatted dict |
| bad_gcedict = {"items": "foo"} |
| self.assertRaises(ValueError, self.driver.ex_set_common_instance_metadata, bad_gcedict) |
| # test gce formatted dict |
| gcedict = { |
| "items": [ |
| {"key": "gcedict1", "value": "v1"}, |
| {"key": "gcedict2", "value": "v2"}, |
| ] |
| } |
| self.driver.ex_set_common_instance_metadata(gcedict) |
| |
| # Verify project notation works |
| project = GCEProject(id=None, name=None, metadata=None, quotas=None, driver=self.driver) |
| project.set_common_instance_metadata(metadata=gcedict) |
| |
| def test_ex_set_node_metadata(self): |
| node = self.driver.ex_get_node("node-name", "us-central1-a") |
| # test non-dict |
| self.assertRaises(ValueError, self.driver.ex_set_node_metadata, node, ["bad", "type"]) |
| # test standard python dict |
| pydict = {"key": "pydict", "value": 1} |
| self.driver.ex_set_node_metadata(node, pydict) |
| # test GCE badly formatted dict |
| bad_gcedict = {"items": "foo"} |
| self.assertRaises(ValueError, self.driver.ex_set_node_metadata, node, bad_gcedict) |
| # test gce formatted dict |
| gcedict = { |
| "items": [ |
| {"key": "gcedict1", "value": "v1"}, |
| {"key": "gcedict2", "value": "v2"}, |
| ] |
| } |
| self.driver.ex_set_node_metadata(node, gcedict) |
| |
| def test_ex_set_node_labels(self): |
| node = self.driver.ex_get_node("node-name", "us-central1-a") |
| # Test basic values |
| simplelabel = {"key": "value"} |
| self.driver.ex_set_node_labels(node, simplelabel) |
| # Test multiple values |
| multilabels = {"item1": "val1", "item2": "val2"} |
| self.driver.ex_set_node_labels(node, multilabels) |
| |
| def test_ex_set_image_labels(self): |
| image = self.driver.ex_get_image("custom-image") |
| # Test basic values |
| simplelabel = {"foo": "bar"} |
| self.driver.ex_set_image_labels(image, simplelabel) |
| image = self.driver.ex_get_image("custom-image") |
| self.assertTrue("foo" in image.extra["labels"]) |
| # Test multiple values |
| multilabels = {"one": "1", "two": "two"} |
| self.driver.ex_set_image_labels(image, multilabels) |
| image = self.driver.ex_get_image("custom-image") |
| self.assertEqual(len(image.extra["labels"]), 3) |
| self.assertTrue("two" in image.extra["labels"]) |
| self.assertTrue("two" in image.extra["labels"]) |
| |
| def test_ex_get_region(self): |
| region_name = "us-central1" |
| region = self.driver.ex_get_region(region_name) |
| self.assertEqual(region.name, region_name) |
| self.assertEqual(region.status, "UP") |
| self.assertEqual(region.zones[0].name, "us-central1-a") |
| |
| def test_ex_get_size(self): |
| size_name = "n1-standard-1" |
| size = self.driver.ex_get_size(size_name) |
| self.assertEqual(size.name, size_name) |
| self.assertEqual(size.extra["zone"].name, "us-central1-a") |
| self.assertEqual(size.disk, 10) |
| self.assertEqual(size.ram, 3840) |
| self.assertEqual(size.extra["guestCpus"], 1) |
| |
| def test_ex_get_targethttpproxy(self): |
| targethttpproxy_name = "web-proxy" |
| targethttpproxy = self.driver.ex_get_targethttpproxy(targethttpproxy_name) |
| self.assertEqual(targethttpproxy.name, targethttpproxy_name) |
| self.assertEqual(targethttpproxy.urlmap.name, "web-map") |
| |
| def test_ex_get_targetinstance(self): |
| targetinstance_name = "lctargetinstance" |
| targetinstance = self.driver.ex_get_targetinstance(targetinstance_name) |
| self.assertEqual(targetinstance.name, targetinstance_name) |
| self.assertEqual(targetinstance.zone.name, "us-central1-a") |
| |
| def test_ex_get_targetpool(self): |
| targetpool_name = "lctargetpool" |
| targetpool = self.driver.ex_get_targetpool(targetpool_name) |
| self.assertEqual(targetpool.name, targetpool_name) |
| self.assertEqual(len(targetpool.nodes), 2) |
| self.assertEqual(targetpool.region.name, "us-central1") |
| |
| def test_ex_get_instancegroupmanager(self): |
| igmgr_name = "myinstancegroup" |
| igmgr = self.driver.ex_get_instancegroupmanager(igmgr_name, "us-central1-b") |
| self.assertEqual(igmgr.name, igmgr_name) |
| self.assertEqual(igmgr.size, 4) |
| self.assertEqual(igmgr.zone.name, "us-central1-b") |
| |
| # search all zones |
| igmgr = self.driver.ex_get_instancegroupmanager(igmgr_name) |
| self.assertEqual(igmgr.name, igmgr_name) |
| self.assertEqual(igmgr.size, 4) |
| self.assertEqual(igmgr.zone.name, "us-central1-a") |
| |
| def test_ex_get_instancetemplate(self): |
| instancetemplate_name = "my-instance-template1" |
| instancetemplate = self.driver.ex_get_instancetemplate(instancetemplate_name) |
| self.assertEqual(instancetemplate.name, instancetemplate_name) |
| self.assertEqual(instancetemplate.extra["properties"]["machineType"], "n1-standard-1") |
| |
| def test_ex_get_snapshot(self): |
| snapshot_name = "lcsnapshot" |
| snapshot = self.driver.ex_get_snapshot(snapshot_name) |
| self.assertEqual(snapshot.name, snapshot_name) |
| self.assertEqual(snapshot.size, "10") |
| self.assertEqual(snapshot.status, "READY") |
| |
| def test_ex_get_urlmap(self): |
| urlmap_name = "web-map" |
| urlmap = self.driver.ex_get_urlmap(urlmap_name) |
| self.assertEqual(urlmap.name, urlmap_name) |
| self.assertEqual(urlmap.default_service.name, "web-service") |
| |
| def test_ex_get_volume(self): |
| volume_name = "lcdisk" |
| volume = self.driver.ex_get_volume(volume_name) |
| self.assertEqual(volume.name, volume_name) |
| self.assertEqual(volume.size, "10") |
| self.assertEqual(volume.extra["status"], "READY") |
| self.assertEqual(volume.extra["type"], "pd-ssd") |
| |
| def test_ex_get_disktype(self): |
| disktype_name = "pd-ssd" |
| disktype_zone = "us-central1-a" |
| disktype = self.driver.ex_get_disktype(disktype_name, disktype_zone) |
| self.assertEqual(disktype.name, disktype_name) |
| self.assertEqual(disktype.zone.name, disktype_zone) |
| self.assertEqual(disktype.extra["description"], "SSD Persistent Disk") |
| self.assertEqual(disktype.extra["valid_disk_size"], "10GB-10240GB") |
| self.assertEqual(disktype.extra["default_disk_size_gb"], "100") |
| |
| # zone not set |
| disktype_name = "pd-ssd" |
| disktype = self.driver.ex_get_disktype(disktype_name) |
| self.assertEqual(disktype.name, disktype_name) |
| |
| def test_ex_get_zone(self): |
| zone_name = "us-central1-b" |
| zone = self.driver.ex_get_zone(zone_name) |
| self.assertEqual(zone.name, zone_name) |
| self.assertFalse(zone.time_until_mw) |
| self.assertFalse(zone.next_mw_duration) |
| |
| zone_no_mw = self.driver.ex_get_zone("us-central1-a") |
| self.assertIsNone(zone_no_mw.time_until_mw) |
| |
| def test_driver_zone_attributes(self): |
| zones = self.driver.ex_list_zones() |
| self.assertEqual(len(self.driver.zone_dict), len(zones)) |
| self.assertEqual(len(self.driver.zone_list), len(zones)) |
| for zone, fetched_zone in zip(self.driver.zone_list, zones): |
| self.assertEqual(zone.id, fetched_zone.id) |
| self.assertEqual(zone.name, fetched_zone.name) |
| self.assertEqual(zone.status, fetched_zone.status) |
| |
| def test_driver_region_attributes(self): |
| regions = self.driver.ex_list_regions() |
| self.assertEqual(len(self.driver.region_dict), len(regions)) |
| self.assertEqual(len(self.driver.region_list), len(regions)) |
| for region, fetched_region in zip(self.driver.region_list, regions): |
| self.assertEqual(region.id, fetched_region.id) |
| self.assertEqual(region.name, fetched_region.name) |
| self.assertEqual(region.status, fetched_region.status) |
| |
| |
| class GCENodeDriverTest2(GoogleTestCase): |
| """ |
| GCE Test Class, test node driver without passing `datacenter` parameter on initialization. |
| """ |
| |
| def setUp(self): |
| GCEMockHttp.test = self |
| GCENodeDriver.connectionCls.conn_class = GCEMockHttp |
| GoogleBaseAuthConnection.conn_class = GoogleAuthMockHttp |
| GCEMockHttp.type = None |
| kwargs = GCE_KEYWORD_PARAMS.copy() |
| kwargs["auth_type"] = "IA" |
| self.driver = GCENodeDriver(*GCE_PARAMS, **kwargs) |
| |
| def test_zone_attributes(self): |
| self.assertIsNone(self.driver._zone_dict) |
| self.assertIsNone(self.driver._zone_list) |
| |
| zones = self.driver.ex_list_zones() |
| |
| self.assertEqual(len(self.driver.zone_list), len(zones)) |
| self.assertEqual(len(self.driver.zone_dict), len(zones)) |
| for zone, fetched_zone in zip(self.driver.zone_list, zones): |
| self.assertEqual(zone.id, fetched_zone.id) |
| self.assertEqual(zone.name, fetched_zone.name) |
| self.assertEqual(zone.status, fetched_zone.status) |
| |
| def test_region_attributes(self): |
| self.assertIsNone(self.driver._region_dict) |
| self.assertIsNone(self.driver._region_list) |
| |
| regions = self.driver.ex_list_regions() |
| |
| self.assertEqual(len(self.driver.region_list), len(regions)) |
| self.assertEqual(len(self.driver.region_dict), len(regions)) |
| for region, fetched_region in zip(self.driver.region_list, regions): |
| self.assertEqual(region.id, fetched_region.id) |
| self.assertEqual(region.name, fetched_region.name) |
| self.assertEqual(region.status, fetched_region.status) |
| |
| |
| class GCEMockHttp(MockHttp, unittest.TestCase): |
| fixtures = ComputeFileFixtures("gce") |
| json_hdr = {"content-type": "application/json; charset=UTF-8"} |
| |
| def _get_method_name(self, type, use_param, qs, path): |
| api_path = "/compute/%s" % API_VERSION |
| project_path = "/projects/%s" % GCE_KEYWORD_PARAMS["project"] |
| path = path.replace(api_path, "") |
| # This replace is separate, since there is a call with a different |
| # project name |
| path = path.replace(project_path, "") |
| # The path to get project information is the base path, so use a fake |
| # '/project' path instead |
| if not path: |
| path = "/project" |
| method_name = super()._get_method_name(type, use_param, qs, path) |
| return method_name |
| |
| def _setUsageExportBucket(self, method, url, body, headers): |
| if method == "POST": |
| body = self.fixtures.load("setUsageExportBucket_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instances_custom_node(self, method, url, body, header): |
| body = self.fixtures.load("zones_us_central1_a_instances_custom_node.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instances_node_name_setMachineType(self, method, url, body, header): |
| body = self.fixtures.load("zones_us_central1_a_instances_node_name_setMachineType.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_operations_operation_setMachineType_notstopped( |
| self, method, url, body, header |
| ): |
| body = self.fixtures.load( |
| "zones_us_central1_a_operations_operation_setMachineType_notstopped.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instances_custom_node_setMachineType(self, method, url, body, header): |
| body = { |
| "error": { |
| "errors": [ |
| { |
| "domain": "global", |
| "reason": "invalid", |
| "message": "Invalid value for field 'resource.machineTypes': " |
| "'projects/project_name/zones/us-central1-a/machineTypes/custom-1-61440'. Resource was not found.", |
| } |
| ], |
| "code": 400, |
| "message": "Invalid value for field 'resource.machineTypes': " |
| "'projects/project_name/zones/us-central1-a/machineTypes/custom-1-61440'. Resource was not found.", |
| } |
| } |
| return ( |
| httplib.BAD_REQUEST, |
| body, |
| self.json_hdr, |
| httplib.responses[httplib.BAD_REQUEST], |
| ) |
| |
| def _zones_us_central1_a_instances_stopped_node_setMachineType(self, method, url, body, header): |
| body = self.fixtures.load("zones_us_central1_a_instances_stopped_node_setMachineType.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_operations_operation_setMachineType(self, method, url, body, header): |
| body = self.fixtures.load("zones_us_central1_a_operations_operation_setMachineType.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_operations_operation_startnode(self, method, url, body, header): |
| body = self.fixtures.load("zones_us_central1_a_operations_operation_startnode.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instances_stopped_node_start(self, method, url, body, header): |
| body = self.fixtures.load("zones_us_central1_a_instances_stopped_node_start.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instances_stopped_node_stop(self, method, url, body, header): |
| body = self.fixtures.load("zones_us_central1_a_instances_stopped_node_stop.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instances_stopped_node(self, method, url, body, headers): |
| body = self.fixtures.load("zones_us_central1_a_instances_stopped_node.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_operations_operation_stopnode(self, method, url, body, headers): |
| body = self.fixtures.load("zones_us_central1_a_operations_operation_stopnode.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instances_node_name_stop(self, method, url, body, headers): |
| body = self.fixtures.load("zones_us_central1_a_instances_node_name_stop.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_acceleratorTypes_nvidia_tesla_k80(self, method, url, body, headers): |
| body = self.fixtures.load("zones_us_central1_a_acceleratorTypes_nvidia_tesla_k80.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instances_node_name_setMetadata(self, method, url, body, headers): |
| body = self.fixtures.load("zones_us_central1_a_instances_node_name_setMetadata_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instances_node_name_setLabels(self, method, url, body, headers): |
| body = self.fixtures.load("zones_us_central1_a_instances_node_name_setLabels_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_images_custom_image_setLabels(self, method, url, body, headers): |
| self.assertTrue("global/images/custom-image/setLabels" in url) |
| body = self.fixtures.load("global_custom_image_setLabels_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _setCommonInstanceMetadata(self, method, url, body, headers): |
| if method == "POST": |
| body = self.fixtures.load("setCommonInstanceMetadata_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _aggregated_subnetworks(self, method, url, body, headers): |
| body = self.fixtures.load("aggregated_subnetworks.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _aggregated_addresses(self, method, url, body, headers): |
| body = self.fixtures.load("aggregated_addresses.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _aggregated_diskTypes(self, method, url, body, headers): |
| body = self.fixtures.load("aggregated_disktypes.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _aggregated_disks(self, method, url, body, headers): |
| body = self.fixtures.load("aggregated_disks.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _aggregated_forwardingRules(self, method, url, body, headers): |
| body = self.fixtures.load("aggregated_forwardingRules.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _aggregated_instances(self, method, url, body, headers): |
| body = self.fixtures.load("aggregated_instances.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _aggregated_instanceGroupManagers(self, method, url, body, headers): |
| body = self.fixtures.load("aggregated_instanceGroupManagers.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _aggregated_machineTypes(self, method, url, body, headers): |
| body = self.fixtures.load("aggregated_machineTypes.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _aggregated_targetInstances(self, method, url, body, headers): |
| body = self.fixtures.load("aggregated_targetInstances.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _aggregated_targetPools(self, method, url, body, headers): |
| body = self.fixtures.load("aggregated_targetPools.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_backendServices(self, method, url, body, headers): |
| if method == "POST": |
| body = self.fixtures.load("global_backendServices_post.json") |
| else: |
| backend_name = getattr(self.test, "backendservices_mock", "web-service") |
| body = self.fixtures.load("global_backendServices-%s.json" % backend_name) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_backendServices_no_backends(self, method, url, body, headers): |
| body = self.fixtures.load("global_backendServices_no_backends.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_backendServices_web_service(self, method, url, body, headers): |
| if method == "DELETE": |
| body = self.fixtures.load("global_backendServices_web_service_delete.json") |
| else: |
| body = self.fixtures.load("global_backendServices_web_service.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_forwardingRules(self, method, url, body, headers): |
| if method == "POST": |
| body = self.fixtures.load("global_forwardingRules_post.json") |
| else: |
| body = self.fixtures.load("global_forwardingRules.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_forwardingRules_http_rule(self, method, url, body, headers): |
| if method == "DELETE": |
| body = self.fixtures.load("global_forwardingRules_http_rule_delete.json") |
| else: |
| body = self.fixtures.load("global_forwardingRules_http_rule.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_httpHealthChecks(self, method, url, body, headers): |
| if method == "POST": |
| body = self.fixtures.load("global_httpHealthChecks_post.json") |
| else: |
| body = self.fixtures.load("global_httpHealthChecks.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_httpHealthChecks_default_health_check(self, method, url, body, headers): |
| body = self.fixtures.load("global_httpHealthChecks_basic-check.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_httpHealthChecks_basic_check(self, method, url, body, headers): |
| body = self.fixtures.load("global_httpHealthChecks_basic-check.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_httpHealthChecks_libcloud_lb_demo_healthcheck(self, method, url, body, headers): |
| body = self.fixtures.load("global_httpHealthChecks_libcloud-lb-demo-healthcheck.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_httpHealthChecks_lchealthcheck(self, method, url, body, headers): |
| if method == "DELETE": |
| body = self.fixtures.load("global_httpHealthChecks_lchealthcheck_delete.json") |
| elif method == "PUT": |
| body = self.fixtures.load("global_httpHealthChecks_lchealthcheck_put.json") |
| else: |
| body = self.fixtures.load("global_httpHealthChecks_lchealthcheck.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_firewalls(self, method, url, body, headers): |
| if method == "POST": |
| body = self.fixtures.load("global_firewalls_post.json") |
| else: |
| body = self.fixtures.load("global_firewalls.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_firewalls_lcfirewall(self, method, url, body, headers): |
| if method == "DELETE": |
| body = self.fixtures.load("global_firewalls_lcfirewall_delete.json") |
| elif method == "PUT": |
| body = self.fixtures.load("global_firewalls_lcfirewall_put.json") |
| else: |
| body = self.fixtures.load("global_firewalls_lcfirewall.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_firewalls_lcfirewall_egress(self, method, url, body, headers): |
| body = self.fixtures.load("global_firewalls_lcfirewall-egress.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_firewalls_lcfirewall_deny(self, method, url, body, headers): |
| body = self.fixtures.load("global_firewalls_lcfirewall-deny.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_images(self, method, url, body, headers): |
| if method == "POST": |
| body = self.fixtures.load("global_images_post.json") |
| elif "maxResults" in url and "pageToken" not in url: |
| body = self.fixtures.load("global_images_paged.json") |
| else: |
| if "maxResults" in url: |
| self.assertIn("pageToken=token", url) |
| body = self.fixtures.load("global_images.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_images_debian_7_wheezy_v20131120(self, method, url, body, headers): |
| body = self.fixtures.load("global_images_debian_7_wheezy_v20131120_delete.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_images_debian_7_wheezy_v20131014_deprecate(self, method, url, body, headers): |
| body = self.fixtures.load("global_images_debian_7_wheezy_v20131014_deprecate.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_images_family_coreos_beta(self, method, url, body, headers): |
| body = self.fixtures.load("global_images_family_notfound.json") |
| return ( |
| httplib.NOT_FOUND, |
| body, |
| self.json_hdr, |
| httplib.responses[httplib.NOT_FOUND], |
| ) |
| |
| def _global_images_family_coreos_stable(self, method, url, body, headers): |
| body = self.fixtures.load("global_images_family_notfound.json") |
| return ( |
| httplib.NOT_FOUND, |
| body, |
| self.json_hdr, |
| httplib.responses[httplib.NOT_FOUND], |
| ) |
| |
| def _global_images_family_nofamily(self, method, url, body, headers): |
| body = self.fixtures.load("global_images_family_notfound.json") |
| return ( |
| httplib.NOT_FOUND, |
| body, |
| self.json_hdr, |
| httplib.responses[httplib.NOT_FOUND], |
| ) |
| |
| def _global_routes(self, method, url, body, headers): |
| if method == "POST": |
| body = self.fixtures.load("global_routes_post.json") |
| else: |
| body = self.fixtures.load("global_routes.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_networks(self, method, url, body, headers): |
| if method == "POST": |
| body = self.fixtures.load("global_networks_post.json") |
| else: |
| body = self.fixtures.load("global_networks.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_networks_custom_network(self, method, url, body, headers): |
| body = self.fixtures.load("global_networks_custom_network.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_networks_cf(self, method, url, body, headers): |
| body = self.fixtures.load("global_networks_cf.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_networks_default(self, method, url, body, headers): |
| body = self.fixtures.load("global_networks_default.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_networks_libcloud_demo_network(self, method, url, body, headers): |
| body = self.fixtures.load("global_networks_libcloud-demo-network.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_networks_libcloud_demo_europe_network(self, method, url, body, headers): |
| body = self.fixtures.load("global_networks_libcloud-demo-europe-network.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_routes_lcdemoroute(self, method, url, body, headers): |
| if method == "DELETE": |
| body = self.fixtures.load("global_routes_lcdemoroute_delete.json") |
| else: |
| body = self.fixtures.load("global_routes_lcdemoroute.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_networks_lcnetwork(self, method, url, body, headers): |
| if method == "DELETE": |
| body = self.fixtures.load("global_networks_lcnetwork_delete.json") |
| else: |
| body = self.fixtures.load("global_networks_lcnetwork.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_snapshots(self, method, url, body, headers): |
| body = self.fixtures.load("global_snapshots.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_snapshots_lcsnapshot(self, method, url, body, headers): |
| if method == "DELETE": |
| body = self.fixtures.load("global_snapshots_lcsnapshot_delete.json") |
| else: |
| body = self.fixtures.load("global_snapshots_lcsnapshot.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_operations_operation_setUsageExportBucket(self, method, url, body, headers): |
| body = self.fixtures.load("operations_operation_setUsageExportBucket.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_operations_operation_setCommonInstanceMetadata(self, method, url, body, headers): |
| body = self.fixtures.load("operations_operation_setCommonInstanceMetadata.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_operations_operation_global_backendServices_post(self, method, url, body, headers): |
| body = self.fixtures.load("operations_operation_global_backendServices_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_operations_operation_global_backendServices_web_service_delete( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_global_backendServices_web_service_delete" ".json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_operations_operation_global_forwardingRules_http_rule_delete( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_global_forwardingRules_http_rule_delete.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_operations_operation_global_forwardingRules_post(self, method, url, body, headers): |
| body = self.fixtures.load("operations_operation_global_forwardingRules_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_operations_operation_global_httpHealthChecks_lchealthcheck_delete( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_global_httpHealthChecks_lchealthcheck_delete.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_operations_operation_global_images_debian7_delete(self, method, url, body, headers): |
| body = self.fixtures.load("operations_operation_global_images_debian7_delete.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_operations_operation_global_httpHealthChecks_lchealthcheck_put( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_global_httpHealthChecks_lchealthcheck_delete.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_operations_operation_global_httpHealthChecks_post(self, method, url, body, headers): |
| body = self.fixtures.load("operations_operation_global_httpHealthChecks_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_operations_operation_global_firewalls_lcfirewall_delete( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load("operations_operation_global_firewalls_lcfirewall_delete.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_operations_operation_global_firewalls_lcfirewall_put( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load("operations_operation_global_firewalls_lcfirewall_put.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_operations_operation_global_firewalls_post(self, method, url, body, headers): |
| body = self.fixtures.load("operations_operation_global_firewalls_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_operations_operation_global_routes_lcdemoroute_delete( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load("operations_operation_global_routes_lcdemoroute_delete.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_operations_operation_global_networks_lcnetwork_delete( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load("operations_operation_global_networks_lcnetwork_delete.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_operations_operation_global_routes_post(self, method, url, body, headers): |
| body = self.fixtures.load("operations_operation_global_routes_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_operations_operation_global_networks_post(self, method, url, body, headers): |
| body = self.fixtures.load("operations_operation_global_networks_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_operations_operation_global_snapshots_lcsnapshot_delete( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load("operations_operation_global_snapshots_lcsnapshot_delete.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_operations_operation_global_image_post(self, method, url, body, headers): |
| body = self.fixtures.load("operations_operation_global_image_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_operations_operation_global_addresses_lcaddressglobal_delete( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_global_addresses_lcaddressglobal_delete.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_operations_operation_global_targetHttpProxies_post( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load("operations_operation_global_targetHttpProxies_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_operations_operation_global_targetHttpProxies_web_proxy_delete( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_global_targetHttpProxies_web_proxy_delete" ".json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_operations_operation_global_urlMaps_post(self, method, url, body, headers): |
| body = self.fixtures.load("operations_operation_global_urlMaps_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_operations_operation_global_urlMaps_web_map_delete( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load("operations_operation_global_urlMaps_web_map_delete.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_targetHttpProxies(self, method, url, body, headers): |
| if method == "POST": |
| body = self.fixtures.load("global_targetHttpProxies_post.json") |
| else: |
| body = self.fixtures.load("global_targetHttpProxies.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_targetHttpProxies_web_proxy(self, method, url, body, headers): |
| if method == "DELETE": |
| body = self.fixtures.load("global_targetHttpProxies_web_proxy_delete.json") |
| else: |
| body = self.fixtures.load("global_targetHttpProxies_web_proxy.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_urlMaps(self, method, url, body, headers): |
| if method == "POST": |
| body = self.fixtures.load("global_urlMaps_post.json") |
| else: |
| body = self.fixtures.load("global_urlMaps.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_urlMaps_web_map(self, method, url, body, headers): |
| if method == "DELETE": |
| body = self.fixtures.load("global_urlMaps_web_map_delete.json") |
| else: |
| body = self.fixtures.load("global_urlMaps_web_map.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_east1_subnetworks_cf_972cf02e6ad49113(self, method, url, body, headers): |
| body = self.fixtures.load("regions_us-east1_subnetworks_cf_972cf02e6ad49113.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_subnetworks_cf_972cf02e6ad49112(self, method, url, body, headers): |
| body = self.fixtures.load("regions_us-central1_subnetworks_cf_972cf02e6ad49112.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _projects_other_name_regions_us_central1(self, method, url, body, headers): |
| body = self.fixtures.load("projects_other_name_regions_us-central1.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _projects_other_name_global_networks_lcnetwork(self, method, url, body, headers): |
| body = self.fixtures.load("projects_other_name_global_networks_lcnetwork.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _projects_other_name_global_networks_cf(self, method, url, body, headers): |
| body = self.fixtures.load("projects_other_name_global_networks_cf.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _projects_other_name_global_networks_shared_network_for_mig( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load("projects_other_name_global_networks_shared_network_for_mig.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _projects_other_name_regions_us_central1_subnetworks_cf_972cf02e6ad49114( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "projects_other_name_regions_us-central1_subnetworks_cf_972cf02e6ad49114.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _projects_other_name_regions_us_central1_subnetworks_shared_subnetwork_for_mig( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "projects_other_name_regions_us-central1_subnetworks_shared_subnetwork_for_mig.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_operations_operation_regions_us_central1_addresses_lcaddress_delete( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_regions_us-central1_addresses_lcaddress_delete.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_operations_operation_global_addresses_post(self, method, url, body, headers): |
| body = self.fixtures.load("operations_operation_global_addresses_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_operations_operation_regions_us_central1_addresses_post( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load("operations_operation_regions_us-central1_addresses_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_operations_operation_regions_us_central1_subnetworks_post( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load("operations_operation_regions_us-central1_subnetworks_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_operations_operation_regions_us_central1_forwardingRules_post( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_regions_us-central1_forwardingRules_post.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_operations_operation_regions_us_central1_forwardingRules_lcforwardingrule_delete( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_regions_us-central1_forwardingRules_lcforwardingrule_delete.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instances_node_name_deleteAccessConfig( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_zones_us-central1-a_instances_node_name_deleteAccessConfig_post.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instances_node_name_serialPort(self, method, url, body, headers): |
| body = self.fixtures.load("zones_us-central1-a_instances_node_name_getSerialOutput.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instances_node_name_addAccessConfig(self, method, url, body, headers): |
| body = self.fixtures.load( |
| "operations_operation_zones_us-central1-a_instances_node_name_addAccessConfig_post.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_operations_operation_setMetadata_post( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_zones_us_central1_a_node_name_setMetadata_post.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_operations_operation_setLabels_post(self, method, url, body, headers): |
| body = self.fixtures.load( |
| "operations_operation_zones_us_central1_a_node_name_setLabels_post.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_operations_operation_setImageLabels_post(self, method, url, body, headers): |
| body = self.fixtures.load("global_operations_operation_setImageLabels_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_operations_operation_zones_us_central1_a_targetInstances_post( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_zones_us-central1-a_targetInstances_post.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_operations_operation_regions_us_central1_targetPools_post( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load("operations_operation_regions_us-central1_targetPools_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_operations_operation_zones_us_central1_a_instances_node_name_addAccessConfig_post( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_zones_us-central1-a_instances_node_name_addAccessConfig_done.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_operations_operation_zones_us_central1_a_instances_node_name_deleteAccessConfig_post( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_zones_us-central1-a_instances_node_name_deleteAccessConfig_done.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_operations_operation_zones_us_central1_a_targetInstances_lctargetinstance_delete( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_zones_us-central1-a_targetInstances_lctargetinstance_delete.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_operations_operation_regions_us_central1_targetPools_lctargetpool_delete( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_regions_us-central1_targetPools_lctargetpool_delete.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_operations_operation_regions_us_central1_targetPools_lctargetpool_removeHealthCheck_post( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_regions_us-central1_targetPools_lctargetpool_removeHealthCheck_post.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_operations_operation_regions_us_central1_targetPools_lctargetpool_addHealthCheck_post( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_regions_us-central1_targetPools_lctargetpool_addHealthCheck_post.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_operations_operation_regions_us_central1_targetPools_lctargetpool_removeInstance_post( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_regions_us-central1_targetPools_lctargetpool_removeInstance_post.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_operations_operation_regions_us_central1_targetPools_lb_pool_setBackup_post( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_regions_us-central1_targetPools_lb_pool_setBackup_post.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_operations_operation_regions_us_central1_targetPools_lctargetpool_addInstance_post( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_regions_us-central1_targetPools_lctargetpool_addInstance_post.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_operations_operation_zones_us_central1_a_disks_lcdisk_delete( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_zones_us-central1-a_disks_lcdisk_delete.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instances_node_name_setDiskAutoDelete( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load("zones_us_central1_a_instances_node_name_setDiskAutoDelete.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_operations_operation_volume_auto_delete( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "zones_us_central1_a_operations_operation_volume_auto_delete.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_operations_operation_zones_us_central1_a_disks_lcdisk_createSnapshot_post( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_zones_us-central1-a_disks_lcdisk_createSnapshot_post.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_operations_operation_zones_us_central1_a_disks_lcdisk_resize_post( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_zones_us-central1-a_disks_lcdisk_resize_post.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_operations_operation_zones_us_central1_a_disks_lcdisk_setLabels_post( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_zones_us-central1-a_disks_lcdisk_setLabels_post.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_operations_operation_zones_us_central1_a_disks_post( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load("operations_operation_zones_us-central1-a_disks_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_operations_operation_zones_us_central1_a_instances_lcnode_000_delete( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_zones_us-central1-a_instances_lcnode-000_delete.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_operations_operation_zones_us_central1_a_instances_lcnode_001_delete( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_zones_us-central1-a_instances_lcnode-001_delete.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_operations_operation_zones_us_central1_a_instances_node_name_delete( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_zones_us-central1-a_instances_node-name_delete.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_operations_operation_zones_us_central1_a_instances_node_name_attachDisk_post( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_zones_us-central1-a_instances_node-name_attachDisk_post.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_operations_operation_zones_us_central1_a_instances_node_name_detachDisk_post( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_zones_us-central1-a_instances_node-name_detachDisk_post.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_operations_operation_zones_us_central1_a_instances_node_name_setTags_post( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_zones_us-central1-a_instances_node-name_setTags_post.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_operations_operation_zones_us_central1_a_instances_node_name_reset_post( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "operations_operation_zones_us-central1-a_instances_node-name_reset_post.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_europe_west1_a_operations_operation_zones_europe_west1_a_instances_post( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load("operations_operation_zones_europe-west1-a_instances_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_operations_operation_zones_us_central1_a_instances_post( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load("operations_operation_zones_us-central1-a_instances_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _project(self, method, url, body, headers): |
| body = self.fixtures.load("project.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _projects_windows_cloud_global_licenses_windows_server_2008_r2_dc( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "projects_windows-cloud_global_licenses_windows_server_2008_r2_dc.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _projects_suse_cloud_global_licenses_sles_11(self, method, url, body, headers): |
| body = self.fixtures.load("projects_suse-cloud_global_licenses_sles_11.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _projects_rhel_cloud_global_licenses_rhel_7_server(self, method, url, body, headers): |
| body = self.fixtures.load("projects_rhel-cloud_global_licenses_rhel_server.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _projects_coreos_cloud_global_licenses_coreos_stable(self, method, url, body, headers): |
| body = self.fixtures.load("projects_coreos-cloud_global_licenses_coreos_stable.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _projects_suse_cloud_global_licenses_sles_12(self, method, url, body, headers): |
| body = self.fixtures.load("projects_suse-cloud_global_licenses_sles_12.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _projects_windows_cloud_global_images(self, method, url, body, header): |
| body = self.fixtures.load("projects_windows-cloud_global_images.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _projects_windows_sql_cloud_global_images(self, method, url, body, header): |
| body = self.fixtures.load("projects_windows-sql-cloud_global_images.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _projects_rhel_cloud_global_images(self, method, url, body, header): |
| body = self.fixtures.load("projects_rhel-cloud_global_images.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _projects_coreos_cloud_global_images(self, method, url, body, header): |
| body = self.fixtures.load("projects_coreos-cloud_global_images.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _projects_coreos_cloud_global_images_family_coreos_beta(self, method, url, body, header): |
| body = self.fixtures.load("projects_coreos-cloud_global_images_family_coreos_beta.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _projects_coreos_cloud_global_images_family_coreos_stable(self, method, url, body, header): |
| body = self.fixtures.load("projects_coreos-cloud_global_images_family_coreos_stable.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _projects_cos_cloud_global_images(self, method, url, body, header): |
| body = self.fixtures.load("projects_cos-cloud_global_images.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _projects_opensuse_cloud_global_images(self, method, url, body, header): |
| body = self.fixtures.load("projects_opensuse-cloud_global_images.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _projects_ubuntu_os_cloud_global_images(self, method, url, body, header): |
| body = self.fixtures.load("projects_ubuntu-os-cloud_global_images.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _projects_centos_cloud_global_images(self, method, url, body, header): |
| body = self.fixtures.load("projects_centos-cloud_global_images.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _projects_suse_cloud_global_images(self, method, url, body, headers): |
| body = self.fixtures.load("projects_suse-cloud_global_images.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _projects_suse_byos_cloud_global_images(self, method, url, body, headers): |
| body = self.fixtures.load("projects_suse-byos-cloud_global_images.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _projects_suse_sap_cloud_global_images(self, method, url, body, headers): |
| body = self.fixtures.load("projects_suse-sap-cloud_global_images.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _projects_debian_cloud_global_images(self, method, url, body, headers): |
| body = self.fixtures.load("projects_debian-cloud_global_images.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions(self, method, url, body, headers): |
| if "pageToken" in url or "filter" in url: |
| body = self.fixtures.load("regions-paged-2.json") |
| elif "maxResults" in url: |
| body = self.fixtures.load("regions-paged-1.json") |
| else: |
| body = self.fixtures.load("regions.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_addresses(self, method, url, body, headers): |
| if method == "POST": |
| body = self.fixtures.load("global_addresses_post.json") |
| else: |
| body = self.fixtures.load("global_addresses.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_europe_west1(self, method, url, body, headers): |
| body = self.fixtures.load("regions_europe-west1.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_asia_east1(self, method, url, body, headers): |
| body = self.fixtures.load("regions_asia-east1.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1(self, method, url, body, headers): |
| body = self.fixtures.load("regions_us-central1.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_east1(self, method, url, body, headers): |
| body = self.fixtures.load("regions_us-east1.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_subnetworks(self, method, url, body, headers): |
| if method == "POST": |
| body = self.fixtures.load("regions_us-central1_subnetworks_post.json") |
| else: |
| body = self.fixtures.load("regions_us-central1_subnetworks.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_addresses(self, method, url, body, headers): |
| if method == "POST": |
| body = self.fixtures.load("regions_us-central1_addresses_post.json") |
| else: |
| body = self.fixtures.load("regions_us-central1_addresses.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_addresses_lcaddressglobal(self, method, url, body, headers): |
| if method == "DELETE": |
| body = self.fixtures.load("global_addresses_lcaddressglobal_delete.json") |
| else: |
| body = self.fixtures.load("global_addresses_lcaddressglobal.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_addresses_lcaddress(self, method, url, body, headers): |
| if method == "DELETE": |
| body = self.fixtures.load("regions_us-central1_addresses_lcaddress_delete.json") |
| else: |
| body = self.fixtures.load("regions_us-central1_addresses_lcaddress.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_addresses_testaddress(self, method, url, body, headers): |
| body = self.fixtures.load("regions_us-central1_addresses_testaddress.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_subnetworks_subnet_1(self, method, url, body, headers): |
| body = self.fixtures.load("regions_us-central1_subnetworks_subnet_1.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_addresses_lcaddressinternal(self, method, url, body, headers): |
| body = self.fixtures.load("regions_us-central1_addresses_lcaddressinternal.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_forwardingRules(self, method, url, body, headers): |
| if method == "POST": |
| body = self.fixtures.load("regions_us-central1_forwardingRules_post.json") |
| else: |
| body = self.fixtures.load("regions_us-central1_forwardingRules.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_forwardingRules_libcloud_lb_demo_lb(self, method, url, body, headers): |
| body = self.fixtures.load("regions_us-central1_forwardingRules_libcloud-lb-demo-lb.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_forwardingRules_lcforwardingrule(self, method, url, body, headers): |
| if method == "DELETE": |
| body = self.fixtures.load( |
| "regions_us-central1_forwardingRules_lcforwardingrule_delete.json" |
| ) |
| else: |
| body = self.fixtures.load("regions_us-central1_forwardingRules_lcforwardingrule.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_targetInstances(self, method, url, body, headers): |
| if method == "POST": |
| body = self.fixtures.load("zones_us-central1-a_targetInstances_post.json") |
| else: |
| body = self.fixtures.load("zones_us-central1-a_targetInstances.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_targetPools(self, method, url, body, headers): |
| if method == "POST": |
| body = self.fixtures.load("regions_us-central1_targetPools_post.json") |
| else: |
| body = self.fixtures.load("regions_us-central1_targetPools.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_targetInstances_lctargetinstance(self, method, url, body, headers): |
| if method == "DELETE": |
| body = self.fixtures.load( |
| "zones_us-central1-a_targetInstances_lctargetinstance_delete.json" |
| ) |
| else: |
| body = self.fixtures.load("zones_us-central1-a_targetInstances_lctargetinstance.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_targetPools_lb_pool_getHealth(self, method, url, body, headers): |
| body = self.fixtures.load("regions_us-central1_targetPools_lb_pool_getHealth.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_targetPools_lb_pool(self, method, url, body, headers): |
| body = self.fixtures.load("regions_us-central1_targetPools_lb_pool.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_targetPools_lctargetpool(self, method, url, body, headers): |
| if method == "DELETE": |
| body = self.fixtures.load("regions_us-central1_targetPools_lctargetpool_delete.json") |
| else: |
| body = self.fixtures.load("regions_us-central1_targetPools_lctargetpool.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_targetPools_lctargetpool_sticky(self, method, url, body, headers): |
| body = self.fixtures.load("regions_us-central1_targetPools_lctargetpool_sticky.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_targetPools_backup_pool(self, method, url, body, headers): |
| body = self.fixtures.load("regions_us-central1_targetPools_backup_pool.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_targetPools_libcloud_lb_demo_lb_tp(self, method, url, body, headers): |
| body = self.fixtures.load("regions_us-central1_targetPools_libcloud-lb-demo-lb-tp.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_targetPools_lctargetpool_removeHealthCheck( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "regions_us-central1_targetPools_lctargetpool_removeHealthCheck_post.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_targetPools_lctargetpool_addHealthCheck( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "regions_us-central1_targetPools_lctargetpool_addHealthCheck_post.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_targetPools_lctargetpool_removeInstance( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "regions_us-central1_targetPools_lctargetpool_removeInstance_post.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_targetPools_lb_pool_setBackup(self, method, url, body, headers): |
| body = self.fixtures.load("regions_us-central1_targetPools_lb_pool_setBackup_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _regions_us_central1_targetPools_lctargetpool_addInstance(self, method, url, body, headers): |
| body = self.fixtures.load( |
| "regions_us-central1_targetPools_lctargetpool_addInstance_post.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones(self, method, url, body, headers): |
| body = self.fixtures.load("zones.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_asia_east_1a(self, method, url, body, headers): |
| body = self.fixtures.load("zones_asia-east1-a.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_asia_east1_b(self, method, url, body, headers): |
| body = self.fixtures.load("zones_asia-east1-b.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_east1_b(self, method, url, body, headers): |
| body = self.fixtures.load("zones_us-east1-b.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_diskTypes(self, method, url, body, headers): |
| body = self.fixtures.load("zones_us-central1-a_diskTypes.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_diskTypes_pd_standard(self, method, url, body, headers): |
| body = self.fixtures.load("zones_us-central1-a_diskTypes_pd_standard.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_diskTypes_pd_ssd(self, method, url, body, headers): |
| body = self.fixtures.load("zones_us-central1-a_diskTypes_pd_ssd.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_disks(self, method, url, body, headers): |
| if method == "POST": |
| body = self.fixtures.load("zones_us-central1-a_disks_post.json") |
| else: |
| body = self.fixtures.load("zones_us-central1-a_disks.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_disks_lcdisk(self, method, url, body, headers): |
| if method == "DELETE": |
| body = self.fixtures.load("zones_us-central1-a_disks_lcdisk_delete.json") |
| else: |
| body = self.fixtures.load("zones_us-central1-a_disks_lcdisk.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_disks_lcdisk_createSnapshot(self, method, url, body, headers): |
| body = self.fixtures.load("zones_us-central1-a_disks_lcdisk_createSnapshot_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_disks_lcdisk_resize(self, method, url, body, headers): |
| body = self.fixtures.load("zones_us-central1-a_disks_lcdisk_resize_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_disks_lcdisk_setLabels(self, method, url, body, header): |
| body = self.fixtures.load("zones_us-central1-a_disks_lcdisk_setLabel_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_disks_node_name(self, method, url, body, headers): |
| body = self.fixtures.load("generic_disk.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_disks_lcnode_000(self, method, url, body, headers): |
| body = self.fixtures.load("generic_disk.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_disks_lcnode_001(self, method, url, body, headers): |
| body = self.fixtures.load("generic_disk.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_b_disks_libcloud_lb_demo_www_000(self, method, url, body, headers): |
| body = self.fixtures.load("generic_disk.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_b_disks_libcloud_lb_demo_www_001(self, method, url, body, headers): |
| body = self.fixtures.load("generic_disk.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_b_disks_libcloud_lb_demo_www_002(self, method, url, body, headers): |
| body = self.fixtures.load("generic_disk.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central2_a_disks_libcloud_demo_boot_disk(self, method, url, body, headers): |
| body = self.fixtures.load("generic_disk.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central2_a_disks_libcloud_demo_np_node(self, method, url, body, headers): |
| body = self.fixtures.load("generic_disk.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central2_a_disks_libcloud_demo_multiple_nodes_000( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load("generic_disk.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central2_a_disks_libcloud_demo_multiple_nodes_001( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load("generic_disk.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_europe_west1_a_disks(self, method, url, body, headers): |
| if method == "POST": |
| body = self.fixtures.load("zones_us-central1-a_disks_post.json") |
| else: |
| body = self.fixtures.load("zones_us-central1-a_disks.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_europe_west1_a_disks_libcloud_demo_europe_np_node(self, method, url, body, headers): |
| body = self.fixtures.load("generic_disk.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_europe_west1_a_disks_libcloud_demo_europe_boot_disk( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load("generic_disk.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_europe_west1_a_disks_libcloud_demo_europe_multiple_nodes_000( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load("generic_disk.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_europe_west1_a_instances(self, method, url, body, headers): |
| if method == "POST": |
| body = self.fixtures.load("zones_europe-west1-a_instances_post.json") |
| else: |
| body = self.fixtures.load("zones_europe-west1-a_instances.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_b_instances(self, method, url, body, headers): |
| if method == "GET": |
| body = "{}" |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_europe_west1_a_diskTypes_pd_standard(self, method, url, body, headers): |
| body = self.fixtures.load("zones_europe-west1-a_diskTypes_pd_standard.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instances(self, method, url, body, headers): |
| if method == "POST": |
| body = self.fixtures.load("zones_us-central1-a_instances_post.json") |
| else: |
| body = self.fixtures.load("zones_us-central1-a_instances.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instances_sn_node_name(self, method, url, body, headers): |
| body = self.fixtures.load("zones_us-central1-a_instances_sn-node-name.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instances_node_name(self, method, url, body, headers): |
| if method == "DELETE": |
| body = self.fixtures.load("zones_us-central1-a_instances_node-name_delete.json") |
| else: |
| body = self.fixtures.load("zones_us-central1-a_instances_node-name.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instances_node_name_attachDisk(self, method, url, body, headers): |
| body = self.fixtures.load("zones_us-central1-a_instances_node-name_attachDisk_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instances_node_name_detachDisk(self, method, url, body, headers): |
| body = self.fixtures.load("zones_us-central1-a_instances_node-name_detachDisk_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instances_node_name_setTags(self, method, url, body, headers): |
| body = self.fixtures.load("zones_us-central1-a_instances_node-name_setTags_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instances_node_name_reset(self, method, url, body, headers): |
| body = self.fixtures.load("zones_us-central1-a_instances_node-name_reset_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instances_lcnode_000(self, method, url, body, headers): |
| if method == "DELETE": |
| body = self.fixtures.load("zones_us-central1-a_instances_lcnode-000_delete.json") |
| else: |
| body = self.fixtures.load("zones_us-central1-a_instances_lcnode-000.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instances_lcnode_001(self, method, url, body, headers): |
| if method == "DELETE": |
| body = self.fixtures.load("zones_us-central1-a_instances_lcnode-001_delete.json") |
| else: |
| body = self.fixtures.load("zones_us-central1-a_instances_lcnode-001.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_b_instances_libcloud_lb_nopubip_001(self, method, url, body, headers): |
| body = self.fixtures.load("zones_us-central1-b_instances_libcloud-lb-nopubip-001.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_b_instances_libcloud_lb_demo_www_000(self, method, url, body, headers): |
| body = self.fixtures.load("zones_us-central1-b_instances_libcloud-lb-demo-www-000.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_b_instances_libcloud_lb_demo_www_001(self, method, url, body, headers): |
| body = self.fixtures.load("zones_us-central1-b_instances_libcloud-lb-demo-www-001.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_b_instances_libcloud_lb_demo_www_002(self, method, url, body, headers): |
| body = self.fixtures.load("zones_us-central1-b_instances_libcloud-lb-demo-www-002.json") |
| return ( |
| httplib.NOT_FOUND, |
| body, |
| self.json_hdr, |
| httplib.responses[httplib.NOT_FOUND], |
| ) |
| |
| def _zones_us_central1_a(self, method, url, body, headers): |
| body = self.fixtures.load("zones_us-central1-a.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_machineTypes(self, method, url, body, headers): |
| body = self.fixtures.load("zones_us-central1-a_machineTypes.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_europe_west1_a_machineTypes_n1_standard_1(self, method, url, body, headers): |
| body = self.fixtures.load("zones_europe-west1-a_machineTypes_n1-standard-1.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_machineTypes_n1_standard_1(self, method, url, body, headers): |
| body = self.fixtures.load("zones_us-central1-a_machineTypes_n1-standard-1.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instanceGroups_myinstancegroup(self, method, url, body, headers): |
| body = self.fixtures.load("zones_us-central1-a_instanceGroup_myinstancegroup.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instanceGroups_myinstancegroup2(self, method, url, body, headers): |
| body = self.fixtures.load("zones_us-central1-a_instanceGroup_myinstancegroup2.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_b_instanceGroups_myinstancegroup(self, method, url, body, headers): |
| body = self.fixtures.load("zones_us-central1-b_instanceGroup_myinstancegroup.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_east1_b_instanceGroups_myinstancegroup(self, method, url, body, headers): |
| body = self.fixtures.load("zones_us-east1-b_instanceGroup_myinstancegroup.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instanceGroups_myinstancegroup_shared_network( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "zones_us-central1-a_instanceGroup_myinstancegroup_shared_network.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instanceGroupManagers_myinstancegroup( |
| self, method, url, body, headers |
| ): |
| if method == "PATCH": |
| # test_ex_instancegroupmanager_set_autohealing_policies |
| body = self.fixtures.load( |
| "zones_us-central1-a_operations_operation_zones_us-central1-a_instanceGroupManagers_insert_post.json" |
| ) |
| else: |
| body = self.fixtures.load( |
| "zones_us-central1-a_instanceGroupManagers_myinstancegroup.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instanceGroupManagers_myinstancegroup_shared_network( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "zones_us-central1-a_instanceGroupManagers_myinstancegroup_shared_network.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_b_instanceGroupManagers_myinstancegroup( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load("zones_us-central1-b_instanceGroupManagers_myinstancegroup.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instanceGroupManagers_myinstancegroup_listManagedInstances( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "_zones_us_central1_a_instanceGroupManagers_myinstancegroup_listManagedInstances.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_east1_b_instanceGroupManagers(self, method, url, body, headers): |
| body = self.fixtures.load("zones_us-east1-b_instanceGroupManagers.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instanceGroupManagers(self, method, url, body, headers): |
| # do an insert. Returns an operations link, which then |
| # returns the MIG URI. |
| if method == "POST": |
| body = self.fixtures.load("zones_us-central1-a_instanceGroupManagers_insert.json") |
| else: |
| body = self.fixtures.load("zones_us-central1-a_instanceGroupManagers.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_operations_operation_zones_us_central1_a_instanceGroupManagers_insert_post( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "zones_us-central1-a_operations_operation_zones_us-central1-a_instanceGroupManagers_insert_post.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_instanceTemplates(self, method, url, body, headers): |
| if method == "POST": |
| # insert |
| body = self.fixtures.load("global_instanceTemplates_insert.json") |
| else: |
| # get or list call |
| body = self.fixtures.load("global_instanceTemplates.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_operations_operation_global_instanceTemplates_my_instance_template1_insert( |
| self, method, url, body, headers |
| ): |
| """Redirects from _global_instanceTemplates""" |
| body = self.fixtures.load("operations_operation_global_instanceTemplates_insert.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_instanceTemplates_my_instance_template1(self, method, url, body, headers): |
| body = self.fixtures.load("global_instanceTemplates_my_instance_template1.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_instanceTemplates_my_instance_template_shared_network( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load( |
| "global_instanceTemplates_my_instance_template_shared_network.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _aggregated_autoscalers(self, method, url, body, headers): |
| body = self.fixtures.load("aggregated_autoscalers.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_sslCertificates(self, method, url, body, headers): |
| if method == "POST": |
| body = self.fixtures.load("global_sslcertificates_post.json") |
| else: |
| body = self.fixtures.load("global_sslcertificates.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_sslCertificates_example(self, method, url, body, headers): |
| body = self.fixtures.load("global_sslcertificates_example.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _global_operations_operation_global_sslcertificates_post(self, method, url, body, headers): |
| body = self.fixtures.load("operations_operation_global_sslcertificates_post.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instanceGroups_myname(self, method, url, body, headers): |
| if method == "DELETE": |
| # delete |
| body = self.fixtures.load("zones_us_central1_a_instanceGroups_myname_delete.json") |
| else: |
| # get or list call |
| body = self.fixtures.load("zones_us_central1_a_instanceGroups_myname.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_operations_operation_zones_us_central1_a_instanceGroups_myname_delete( |
| self, method, url, body, headers |
| ): |
| """Redirects from _zones_us_central1_a_instanceGroups_myname""" |
| body = self.fixtures.load( |
| "operations_operation_zones_us_central1_a_instanceGroups_myname_delete.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instanceGroups(self, method, url, body, headers): |
| if method == "POST": |
| # insert |
| body = self.fixtures.load("zones_us_central1_a_instanceGroups_insert.json") |
| else: |
| # get or list call |
| body = self.fixtures.load("zones_us_central1_a_instanceGroups.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instanceGroups_zone_attribute_not_present( |
| self, method, url, body, headers |
| ): |
| if method == "GET": |
| # get or list call |
| body = self.fixtures.load( |
| "zones_us_central1_a_instanceGroups_zone_attribute_not_present.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_operations_operation_zones_us_central1_a_instanceGroups_myname_insert( |
| self, method, url, body, headers |
| ): |
| """Redirects from _zones_us_central1_a_instanceGroups""" |
| body = self.fixtures.load( |
| "operations_operation_zones_us_central1_a_instanceGroups_insert.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instanceGroups_myname_listInstances(self, method, url, body, headers): |
| # POST |
| body = self.fixtures.load("zones_us_central1_a_instanceGroups_myname_listInstances.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instanceGroups_myname_addInstances(self, method, url, body, headers): |
| # POST |
| body = self.fixtures.load("zones_us_central1_a_instanceGroups_myname_addInstances.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_operations_operation_zones_us_central1_a_instanceGroups_myname_addInstances( |
| self, method, url, body, headers |
| ): |
| """Redirects from _zones_us_central1_a_instanceGroups_myname_addInstances""" |
| body = self.fixtures.load( |
| "operations_operation_zones_us_central1_a_instanceGroups_myname_addInstances.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instanceGroups_myname_removeInstances( |
| self, method, url, body, headers |
| ): |
| # POST |
| body = self.fixtures.load("zones_us_central1_a_instanceGroups_myname_removeInstances.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_operations_operation_zones_us_central1_a_instanceGroups_myname_removeInstances( |
| self, method, url, body, headers |
| ): |
| """Redirects from _zones_us_central1_a_instanceGroups_myname_removeInstances""" |
| body = self.fixtures.load( |
| "operations_operation_zones_us_central1_a_instanceGroups_myname_removeInstances.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_instanceGroups_myname_setNamedPorts(self, method, url, body, headers): |
| # POST |
| body = self.fixtures.load("zones_us_central1_a_instanceGroups_myname_setNamedPorts.json") |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| def _zones_us_central1_a_operations_operation_zones_us_central1_a_instanceGroups_myname_setNamedPorts( |
| self, method, url, body, headers |
| ): |
| """Redirects from _zones_us_central1_a_instanceGroups_myname_setNamedPorts""" |
| body = self.fixtures.load( |
| "operations_operation_zones_us_central1_a_instanceGroups_myname_setNamedPorts.json" |
| ) |
| return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) |
| |
| |
| if __name__ == "__main__": |
| sys.exit(unittest.main()) |
