diff options
| -rw-r--r-- | tempest_lib/services/compute/floating_ips_client.py | 103 | ||||
| -rw-r--r-- | tempest_lib/tests/services/compute/test_floating_ips_client.py | 113 |
2 files changed, 216 insertions, 0 deletions
diff --git a/tempest_lib/services/compute/floating_ips_client.py b/tempest_lib/services/compute/floating_ips_client.py new file mode 100644 index 0000000..6abf921 --- /dev/null +++ b/tempest_lib/services/compute/floating_ips_client.py @@ -0,0 +1,103 @@ +# Copyright 2012 OpenStack Foundation +# All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +from oslo_serialization import jsonutils as json +from six.moves.urllib import parse as urllib + +from tempest_lib.api_schema.response.compute.v2_1 import floating_ips as schema +from tempest_lib.common import rest_client +from tempest_lib import exceptions as lib_exc + + +class FloatingIPsClient(rest_client.RestClient): + + def list_floating_ips(self, **params): + """Returns a list of all floating IPs filtered by any parameters.""" + url = 'os-floating-ips' + if params: + url += '?%s' % urllib.urlencode(params) + + resp, body = self.get(url) + body = json.loads(body) + self.validate_response(schema.list_floating_ips, resp, body) + return rest_client.ResponseBody(resp, body) + + def show_floating_ip(self, floating_ip_id): + """Get the details of a floating IP.""" + url = "os-floating-ips/%s" % floating_ip_id + resp, body = self.get(url) + body = json.loads(body) + self.validate_response(schema.create_get_floating_ip, resp, body) + return rest_client.ResponseBody(resp, body) + + def create_floating_ip(self, **kwargs): + """Allocate a floating IP to the project. + + Available params: see http://developer.openstack.org/ + api-ref-compute-v2.1.html#createFloatingIP + """ + url = 'os-floating-ips' + post_body = json.dumps(kwargs) + resp, body = self.post(url, post_body) + body = json.loads(body) + self.validate_response(schema.create_get_floating_ip, resp, body) + return rest_client.ResponseBody(resp, body) + + def delete_floating_ip(self, floating_ip_id): + """Deletes the provided floating IP from the project.""" + url = "os-floating-ips/%s" % floating_ip_id + resp, body = self.delete(url) + self.validate_response(schema.add_remove_floating_ip, resp, body) + return rest_client.ResponseBody(resp, body) + + def associate_floating_ip_to_server(self, floating_ip, server_id): + """Associate the provided floating IP to a specific server.""" + url = "servers/%s/action" % server_id + post_body = { + 'addFloatingIp': { + 'address': floating_ip, + } + } + + post_body = json.dumps(post_body) + resp, body = self.post(url, post_body) + self.validate_response(schema.add_remove_floating_ip, resp, body) + return rest_client.ResponseBody(resp, body) + + def disassociate_floating_ip_from_server(self, floating_ip, server_id): + """Disassociate the provided floating IP from a specific server.""" + url = "servers/%s/action" % server_id + post_body = { + 'removeFloatingIp': { + 'address': floating_ip, + } + } + + post_body = json.dumps(post_body) + resp, body = self.post(url, post_body) + self.validate_response(schema.add_remove_floating_ip, resp, body) + return rest_client.ResponseBody(resp, body) + + def is_resource_deleted(self, id): + try: + self.show_floating_ip(id) + except lib_exc.NotFound: + return True + return False + + @property + def resource_type(self): + """Returns the primary type of resource this client works with.""" + return 'floating_ip' diff --git a/tempest_lib/tests/services/compute/test_floating_ips_client.py b/tempest_lib/tests/services/compute/test_floating_ips_client.py new file mode 100644 index 0000000..01796e5 --- /dev/null +++ b/tempest_lib/tests/services/compute/test_floating_ips_client.py @@ -0,0 +1,113 @@ +# Copyright 2015 IBM Corp. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +from oslotest import mockpatch + +from tempest_lib import exceptions as lib_exc +from tempest_lib.services.compute import floating_ips_client +from tempest_lib.tests import fake_auth_provider +from tempest_lib.tests.services.compute import base + + +class TestFloatingIpsClient(base.BaseComputeServiceTest): + + floating_ip = {"fixed_ip": None, + "id": "46d61064-13ba-4bf0-9557-69de824c3d6f", + "instance_id": "a1daa443-a6bb-463e-aea2-104b7d912eb8", + "ip": "10.10.10.1", + "pool": "nova"} + + def setUp(self): + super(TestFloatingIpsClient, self).setUp() + fake_auth = fake_auth_provider.FakeAuthProvider() + self.client = floating_ips_client.FloatingIPsClient( + fake_auth, 'compute', 'regionOne') + + def _test_list_floating_ips(self, bytes_body=False): + expected = {'floating_ips': [TestFloatingIpsClient.floating_ip]} + self.check_service_client_function( + self.client.list_floating_ips, + 'tempest_lib.common.rest_client.RestClient.get', + expected, + bytes_body) + + def test_list_floating_ips_str_body(self): + self._test_list_floating_ips(bytes_body=False) + + def test_list_floating_ips_byte_body(self): + self._test_list_floating_ips(bytes_body=True) + + def _test_show_floating_ip(self, bytes_body=False): + expected = {"floating_ip": TestFloatingIpsClient.floating_ip} + self.check_service_client_function( + self.client.show_floating_ip, + 'tempest_lib.common.rest_client.RestClient.get', + expected, + bytes_body, + floating_ip_id='a1daa443-a6bb-463e-aea2-104b7d912eb8') + + def test_show_floating_ip_str_body(self): + self._test_show_floating_ip(bytes_body=False) + + def test_show_floating_ip_byte_body(self): + self._test_show_floating_ip(bytes_body=True) + + def _test_create_floating_ip(self, bytes_body=False): + expected = {"floating_ip": TestFloatingIpsClient.floating_ip} + self.check_service_client_function( + self.client.create_floating_ip, + 'tempest_lib.common.rest_client.RestClient.post', + expected, + bytes_body, + pool_name='nova') + + def test_create_floating_ip_str_body(self): + self._test_create_floating_ip(bytes_body=False) + + def test_create_floating_ip_byte_body(self): + self._test_create_floating_ip(bytes_body=True) + + def test_delete_floating_ip(self): + self.check_service_client_function( + self.client.delete_floating_ip, + 'tempest_lib.common.rest_client.RestClient.delete', + {}, status=202, floating_ip_id='fake-id') + + def test_associate_floating_ip_to_server(self): + self.check_service_client_function( + self.client.associate_floating_ip_to_server, + 'tempest_lib.common.rest_client.RestClient.post', + {}, status=202, floating_ip='10.10.10.1', + server_id='c782b7a9-33cd-45f0-b795-7f87f456408b') + + def test_disassociate_floating_ip_from_server(self): + self.check_service_client_function( + self.client.disassociate_floating_ip_from_server, + 'tempest_lib.common.rest_client.RestClient.post', + {}, status=202, floating_ip='10.10.10.1', + server_id='c782b7a9-33cd-45f0-b795-7f87f456408b') + + def test_is_resource_deleted_true(self): + self.useFixture(mockpatch.Patch( + 'tempest_lib.services.compute.floating_ips_client.' + 'FloatingIPsClient.show_floating_ip', + side_effect=lib_exc.NotFound())) + self.assertTrue(self.client.is_resource_deleted('fake-id')) + + def test_is_resource_deleted_false(self): + self.useFixture(mockpatch.Patch( + 'tempest_lib.services.compute.floating_ips_client.' + 'FloatingIPsClient.show_floating_ip', + return_value={"floating_ip": TestFloatingIpsClient.floating_ip})) + self.assertFalse(self.client.is_resource_deleted('fake-id')) |