openstack/deb-nova
Code Issues Proposed changes

Merge "Use controller methods in test_floating_ips"

Browse Source
This commit is contained in:
Jenkins
2015-01-26 03:28:30 +00:00
committed by Gerrit Code Review
parent 0855b017b3 27831421c6
commit f3fadb5a2b
1 changed files with 67 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files

204
nova/tests/unit/api/openstack/compute/contrib/test_floating_ips.py
View File

@@ -18,7 +18,6 @@ import contextlib
import uuid
import mock
from oslo.serialization import jsonutils
import webob
from nova.api.openstack.compute.contrib import floating_ips as fips_v2
@@ -114,7 +113,7 @@ class FloatingIpTestNeutronV21(test.NoDBTestCase):
self.controller = self.floating_ips.FloatingIPController()
def test_floatingip_delete(self):
req = fakes.HTTPRequest.blank('/v2/fake/os-floating-ips/1')
req = fakes.HTTPRequest.blank('')
fip_val = {'address': '1.1.1.1', 'fixed_ip_id': '192.168.1.2'}
with contextlib.nested(
mock.patch.object(self.controller.network_api,
@@ -139,7 +138,7 @@ class FloatingIpTestNeutronV21(test.NoDBTestCase):
def _test_floatingip_delete_not_found(self, ex,
expect_ex=webob.exc.HTTPNotFound):
req = fakes.HTTPRequest.blank('/v2/fake/os-floating-ips/1')
req = fakes.HTTPRequest.blank('')
with contextlib.nested(
mock.patch.object(self.controller.network_api,
'get_floating_ip',
@@ -173,7 +172,6 @@ class FloatingIpTestV21(test.TestCase):
floating_ip = "10.10.10.10"
floating_ip_2 = "10.10.10.11"
floating_ips = fips_v21
url = '/v2/fake/servers/test_inst/action'
def _create_floating_ips(self, floating_ips=None):
"""Create a floating ip object."""
@@ -194,15 +192,6 @@ class FloatingIpTestV21(test.TestCase):
def _delete_floating_ip(self):
db.floating_ip_destroy(self.context, self.floating_ip)
def _get_fake_fip_request(self, act=''):
return fakes.HTTPRequest.blank('/v2/fake/os-floating-ips/%s' % act)
def _get_fake_server_request(self):
return fakes.HTTPRequest.blank('/v2/fake/servers/test_inst/action')
def _get_fake_app(self):
return fakes.wsgi_app_v21(init_only=('servers', 'os-floating-ips'))
def setUp(self):
super(FloatingIpTestV21, self).setUp()
self.stubs.Set(compute.api.API, "get",
@@ -234,14 +223,13 @@ class FloatingIpTestV21(test.TestCase):
self.controller = self.floating_ips.FloatingIPController()
self.manager = self.floating_ips.\
FloatingIPActionController(self.ext_mgr)
self.app = self._get_fake_app()
self.fake_req = fakes.HTTPRequest.blank('')
def tearDown(self):
self._delete_floating_ip()
super(FloatingIpTestV21, self).tearDown()
def test_floatingip_delete(self):
req = self._get_fake_fip_request('1')
fip_val = {'address': '1.1.1.1', 'fixed_ip_id': '192.168.1.2'}
with contextlib.nested(
mock.patch.object(self.controller.network_api,
@@ -255,20 +243,19 @@ class FloatingIpTestV21(test.TestCase):
'get_floating_ip',
return_value=fip_val)) as (
disoc_fip, rel_fip, _, _):
self.controller.delete(req, 1)
self.controller.delete(self.fake_req, 1)
self.assertTrue(disoc_fip.called)
self.assertTrue(rel_fip.called)
def _test_floatingip_delete_not_found(self, ex,
expect_ex=webob.exc.HTTPNotFound):
req = self._get_fake_fip_request('1')
with contextlib.nested(
mock.patch.object(self.controller.network_api,
'get_floating_ip',
side_effect=ex)
):
self.assertRaises(expect_ex,
self.controller.delete, req, 1)
self.controller.delete, self.fake_req, 1)
def test_floatingip_delete_not_found_ip(self):
ex = exception.FloatingIpNotFound(id=1)
@@ -302,8 +289,7 @@ class FloatingIpTestV21(test.TestCase):
self.assertIn('floating_ip', view)
def test_floating_ips_list(self):
req = self._get_fake_fip_request()
res_dict = self.controller.index(req)
res_dict = self.controller.index(self.fake_req)
response = {'floating_ips': [{'instance_id': FAKE_UUID,
'ip': '10.10.10.10',
@@ -324,13 +310,9 @@ class FloatingIpTestV21(test.TestCase):
self.stubs.Set(network.api.API, "get_floating_ip",
fake_get_floating_ip)
req = self._get_fake_fip_request('9876')
req.method = 'DELETE'
res = req.get_response(self.app)
self.assertEqual(res.status_int, 404)
expected_msg = ('{"itemNotFound": {"message": "Floating ip not found '
'for id 9876", "code": 404}}')
self.assertEqual(res.body, expected_msg)
ex = self.assertRaises(webob.exc.HTTPNotFound,
self.controller.delete, self.fake_req, '9876')
self.assertIn("Floating ip not found for id 9876", ex.explanation)
def test_floating_ip_release_race_cond(self):
def fake_get_floating_ip(*args, **kwargs):
@@ -349,14 +331,18 @@ class FloatingIpTestV21(test.TestCase):
self.stubs.Set(self.floating_ips, "disassociate_floating_ip",
fake_disassociate_floating_ip)
req = self._get_fake_fip_request('1')
req.method = 'DELETE'
res = req.get_response(self.app)
self.assertEqual(res.status_int, 202)
res = self.controller.delete(self.fake_req, '9876')
# NOTE: on v2.1, http status code is set as wsgi_code of API
# method instead of status_int in a response object.
if isinstance(self.controller,
fips_v21.FloatingIPController):
status_int = self.controller.delete.wsgi_code
else:
status_int = res.status_int
self.assertEqual(status_int, 202)
def test_floating_ip_show(self):
req = self._get_fake_fip_request('1')
res_dict = self.controller.show(req, 1)
res_dict = self.controller.show(self.fake_req, 1)
self.assertEqual(res_dict['floating_ip']['id'], 1)
self.assertEqual(res_dict['floating_ip']['ip'], '10.10.10.10')
@@ -369,12 +355,9 @@ class FloatingIpTestV21(test.TestCase):
self.stubs.Set(network.api.API, "get_floating_ip",
fake_get_floating_ip)
req = self._get_fake_fip_request('9876')
res = req.get_response(self.app)
self.assertEqual(res.status_int, 404)
expected_msg = ('{"itemNotFound": {"message": "Floating ip not found '
'for id 9876", "code": 404}}')
self.assertEqual(res.body, expected_msg)
ex = self.assertRaises(webob.exc.HTTPNotFound,
self.controller.show, self.fake_req, '9876')
self.assertIn("Floating ip not found for id 9876", ex.explanation)
def test_show_associated_floating_ip(self):
def get_floating_ip(self, context, id):
@@ -385,8 +368,7 @@ class FloatingIpTestV21(test.TestCase):
self.stubs.Set(network.api.API, "get_floating_ip", get_floating_ip)
req = self._get_fake_fip_request('1')
res_dict = self.controller.show(req, 1)
res_dict = self.controller.show(self.fake_req, 1)
self.assertEqual(res_dict['floating_ip']['id'], 1)
self.assertEqual(res_dict['floating_ip']['ip'], '10.10.10.10')
@@ -421,9 +403,8 @@ class FloatingIpTestV21(test.TestCase):
self.stubs.Set(network.api.API, "allocate_floating_ip", fake_allocate)
req = self._get_fake_fip_request()
ex = self.assertRaises(webob.exc.HTTPNotFound,
self.controller.create, req)
self.controller.create, self.fake_req)
self.assertIn('No more floating ips', ex.explanation)
@@ -433,9 +414,9 @@ class FloatingIpTestV21(test.TestCase):
self.stubs.Set(network.api.API, "allocate_floating_ip", fake_allocate)
req = self._get_fake_fip_request()
ex = self.assertRaises(webob.exc.HTTPNotFound,
self.controller.create, req, {'pool': 'non_existent_pool'})
self.controller.create, self.fake_req,
{'pool': 'non_existent_pool'})
self.assertIn('No more floating ips in pool non_existent_pool',
ex.explanation)
@@ -443,18 +424,17 @@ class FloatingIpTestV21(test.TestCase):
@mock.patch('nova.network.api.API.allocate_floating_ip',
side_effect=exception.FloatingIpLimitExceeded())
def test_floating_ip_allocate_over_quota(self, allocate_mock):
req = self._get_fake_fip_request()
ex = self.assertRaises(webob.exc.HTTPForbidden,
self.controller.create, req)
self.controller.create, self.fake_req)
self.assertIn('IP allocation over quota', ex.explanation)
@mock.patch('nova.network.api.API.allocate_floating_ip',
side_effect=exception.FloatingIpLimitExceeded())
def test_floating_ip_allocate_quota_exceed_in_pool(self, allocate_mock):
req = self._get_fake_fip_request()
ex = self.assertRaises(webob.exc.HTTPForbidden,
self.controller.create, req, {'pool': 'non_existent_pool'})
self.controller.create, self.fake_req,
{'pool': 'non_existent_pool'})
self.assertIn('IP allocation over quota in pool non_existent_pool.',
ex.explanation)
@@ -462,9 +442,9 @@ class FloatingIpTestV21(test.TestCase):
@mock.patch('nova.network.api.API.allocate_floating_ip',
side_effect=exception.FloatingIpPoolNotFound())
def test_floating_ip_create_with_unknown_pool(self, allocate_mock):
req = self._get_fake_fip_request()
ex = self.assertRaises(webob.exc.HTTPNotFound,
self.controller.create, req, {'pool': 'non_existent_pool'})
self.controller.create, self.fake_req,
{'pool': 'non_existent_pool'})
self.assertIn('Floating ip pool not found.', ex.explanation)
@@ -480,8 +460,7 @@ class FloatingIpTestV21(test.TestCase):
self.stubs.Set(network.api.API, "get_floating_ip_by_address",
fake2)
req = self._get_fake_fip_request()
res_dict = self.controller.create(req)
res_dict = self.controller.create(self.fake_req)
ip = res_dict['floating_ip']
@@ -494,8 +473,7 @@ class FloatingIpTestV21(test.TestCase):
self.assertEqual(ip, expected)
def test_floating_ip_release(self):
req = self._get_fake_fip_request('1')
self.controller.delete(req, 1)
self.controller.delete(self.fake_req, 1)
def test_floating_ip_associate(self):
fixed_address = '192.168.1.100'
@@ -507,8 +485,7 @@ class FloatingIpTestV21(test.TestCase):
fake_associate_floating_ip)
body = dict(addFloatingIp=dict(address=self.floating_ip))
req = self._get_fake_server_request()
rsp = self.manager._add_floating_ip(req, 'test_inst', body)
rsp = self.manager._add_floating_ip(self.fake_req, 'test_inst', body)
self.assertEqual(202, rsp.status_int)
def test_floating_ip_associate_invalid_instance(self):
@@ -521,10 +498,9 @@ class FloatingIpTestV21(test.TestCase):
body = dict(addFloatingIp=dict(address=self.floating_ip))
req = self._get_fake_server_request()
self.assertRaises(webob.exc.HTTPNotFound,
self.manager._add_floating_ip, req, 'test_inst',
body)
self.manager._add_floating_ip, self.fake_req,
'test_inst', body)
def test_associate_not_allocated_floating_ip_to_instance(self):
def fake_associate_floating_ip(self, context, instance,
@@ -536,24 +512,19 @@ class FloatingIpTestV21(test.TestCase):
fake_associate_floating_ip)
floating_ip = '10.10.10.11'
body = dict(addFloatingIp=dict(address=floating_ip))
req = webob.Request.blank(self.url)
req.method = "POST"
req.body = jsonutils.dumps(body)
req.headers["content-type"] = "application/json"
resp = req.get_response(self.app)
res_dict = jsonutils.loads(resp.body)
self.assertEqual(resp.status_int, 404)
self.assertEqual(res_dict['itemNotFound']['message'],
"floating ip not found")
ex = self.assertRaises(webob.exc.HTTPNotFound,
self.manager._add_floating_ip,
self.fake_req, 'test_inst', body)
self.assertIn("floating ip not found", ex.explanation)
@mock.patch.object(network.api.API, 'associate_floating_ip',
side_effect=exception.Forbidden)
def test_associate_floating_ip_forbidden(self, associate_mock):
body = dict(addFloatingIp=dict(address='10.10.10.11'))
req = self._get_fake_server_request()
self.assertRaises(webob.exc.HTTPForbidden,
self.manager._add_floating_ip, req, 'test_inst',
body)
self.manager._add_floating_ip, self.fake_req,
'test_inst', body)
def test_associate_floating_ip_bad_address_key(self):
body = dict(addFloatingIp=dict(bad_address='10.10.10.11'))
@@ -579,17 +550,16 @@ class FloatingIpTestV21(test.TestCase):
body = dict(removeFloatingIp=dict(address='10.10.10.10'))
req = self._get_fake_server_request()
rsp = self.manager._remove_floating_ip(req, 'test_inst', body)
rsp = self.manager._remove_floating_ip(self.fake_req, 'test_inst',
body)
self.assertEqual(202, rsp.status_int)
def test_floating_ip_disassociate_missing(self):
body = dict(removeFloatingIp=dict(address='10.10.10.10'))
req = self._get_fake_server_request()
self.assertRaises(webob.exc.HTTPConflict,
self.manager._remove_floating_ip,
req, 'test_inst', body)
self.fake_req, 'test_inst', body)
def test_floating_ip_associate_non_existent_ip(self):
def fake_network_api_associate(self, context, instance,
@@ -604,10 +574,9 @@ class FloatingIpTestV21(test.TestCase):
fake_network_api_associate)
body = dict(addFloatingIp=dict(address='1.1.1.1'))
req = self._get_fake_server_request()
self.assertRaises(webob.exc.HTTPNotFound,
self.manager._add_floating_ip,
req, 'test_inst', body)
self.fake_req, 'test_inst', body)
def test_floating_ip_disassociate_non_existent_ip(self):
def network_api_get_floating_ip_by_address(self, context,
@@ -621,10 +590,9 @@ class FloatingIpTestV21(test.TestCase):
network_api_get_floating_ip_by_address)
body = dict(removeFloatingIp=dict(address='1.1.1.1'))
req = self._get_fake_server_request()
self.assertRaises(webob.exc.HTTPNotFound,
self.manager._remove_floating_ip,
req, 'test_inst', body)
self.fake_req, 'test_inst', body)
def test_floating_ip_disassociate_wrong_instance_uuid(self):
def get_instance_by_floating_ip_addr(self, context, address):
@@ -637,10 +605,9 @@ class FloatingIpTestV21(test.TestCase):
wrong_uuid = 'aaaaaaaa-ffff-ffff-ffff-aaaaaaaaaaaa'
body = dict(removeFloatingIp=dict(address='10.10.10.10'))
req = self._get_fake_server_request()
self.assertRaises(webob.exc.HTTPConflict,
self.manager._remove_floating_ip,
req, wrong_uuid, body)
self.fake_req, wrong_uuid, body)
def test_floating_ip_disassociate_wrong_instance_id(self):
def get_instance_by_floating_ip_addr(self, context, address):
@@ -652,10 +619,9 @@ class FloatingIpTestV21(test.TestCase):
body = dict(removeFloatingIp=dict(address='10.10.10.10'))
req = self._get_fake_server_request()
self.assertRaises(webob.exc.HTTPConflict,
self.manager._remove_floating_ip,
req, 'test_inst', body)
self.fake_req, 'test_inst', body)
def test_floating_ip_disassociate_auto_assigned(self):
def fake_get_floating_ip_addr_auto_assigned(self, context, address):
@@ -677,10 +643,9 @@ class FloatingIpTestV21(test.TestCase):
self.stubs.Set(network.api.API, "disassociate_floating_ip",
network_api_disassociate)
body = dict(removeFloatingIp=dict(address='10.10.10.10'))
req = self._get_fake_server_request()
self.assertRaises(webob.exc.HTTPForbidden,
self.manager._remove_floating_ip,
req, 'test_inst', body)
self.fake_req, 'test_inst', body)
def test_floating_ip_disassociate_map_authorization_exc(self):
def fake_get_floating_ip_addr_auto_assigned(self, context, address):
@@ -701,51 +666,37 @@ class FloatingIpTestV21(test.TestCase):
self.stubs.Set(network.api.API, "disassociate_floating_ip",
network_api_disassociate)
body = dict(removeFloatingIp=dict(address='10.10.10.10'))
req = self._get_fake_server_request()
self.assertRaises(webob.exc.HTTPForbidden,
self.manager._remove_floating_ip,
req, 'test_inst', body)
self.fake_req, 'test_inst', body)
# these are a few bad param tests
def test_bad_address_param_in_remove_floating_ip(self):
body = dict(removeFloatingIp=dict(badparam='11.0.0.1'))
req = self._get_fake_server_request()
self.assertRaises(webob.exc.HTTPBadRequest,
self.manager._remove_floating_ip, req, 'test_inst',
body)
self.manager._remove_floating_ip, self.fake_req,
'test_inst', body)
def test_missing_dict_param_in_remove_floating_ip(self):
body = dict(removeFloatingIp='11.0.0.1')
req = self._get_fake_server_request()
self.assertRaises(webob.exc.HTTPBadRequest,
self.manager._remove_floating_ip, req, 'test_inst',
body)
self.manager._remove_floating_ip, self.fake_req,
'test_inst', body)
def test_missing_dict_param_in_add_floating_ip(self):
body = dict(addFloatingIp='11.0.0.1')
req = self._get_fake_server_request()
self.assertRaises(webob.exc.HTTPBadRequest,
self.manager._add_floating_ip, req, 'test_inst',
body)
self.manager._add_floating_ip, self.fake_req,
'test_inst', body)
class FloatingIpTestV2(FloatingIpTestV21):
floating_ips = fips_v2
def _get_fake_app(self):
return fakes.wsgi_app(init_only=('servers', 'os-floating-ips'))
def setUp(self):
super(FloatingIpTestV2, self).setUp()
self.flags(
osapi_compute_extension=[
'nova.api.openstack.compute.contrib.select_extensions'],
osapi_compute_ext_list=['Floating_ips'])
def test_not_extended_floating_ip_associate_fixed(self):
# Check that fixed_address is ignored if os-extended-floating-ips
# is not loaded
@@ -761,8 +712,7 @@ class FloatingIpTestV2(FloatingIpTestV21):
body = dict(addFloatingIp=dict(address=self.floating_ip,
fixed_address=fixed_address_requested))
req = self._get_fake_server_request()
rsp = self.manager._add_floating_ip(req, 'test_inst', body)
rsp = self.manager._add_floating_ip(self.fake_req, 'test_inst', body)
self.assertEqual(202, rsp.status_int)
def test_floatingip_delete_invalid_id(self):
@@ -794,12 +744,6 @@ class ExtendedFloatingIpTestV21(test.TestCase):
def _delete_floating_ip(self):
db.floating_ip_destroy(self.context, self.floating_ip)
def _get_fake_request(self):
return fakes.HTTPRequest.blank('/v2/fake/servers/test_inst/action')
def _get_fake_app(self):
return fakes.wsgi_app_v21(init_only=('servers', 'os-floating-ips'))
def setUp(self):
super(ExtendedFloatingIpTestV21, self).setUp()
self.stubs.Set(compute.api.API, "get",
@@ -833,7 +777,7 @@ class ExtendedFloatingIpTestV21(test.TestCase):
self.controller = self.floating_ips.FloatingIPController()
self.manager = self.floating_ips.\
FloatingIPActionController(self.ext_mgr)
self.app = self._get_fake_app()
self.fake_req = fakes.HTTPRequest.blank('')
def tearDown(self):
self._delete_floating_ip()
@@ -850,8 +794,7 @@ class ExtendedFloatingIpTestV21(test.TestCase):
body = dict(addFloatingIp=dict(address=self.floating_ip,
fixed_address=fixed_address))
req = self._get_fake_request()
rsp = self.manager._add_floating_ip(req, 'test_inst', body)
rsp = self.manager._add_floating_ip(self.fake_req, 'test_inst', body)
self.assertEqual(202, rsp.status_int)
def test_extended_floating_ip_associate_fixed_not_allocated(self):
@@ -863,26 +806,13 @@ class ExtendedFloatingIpTestV21(test.TestCase):
body = dict(addFloatingIp=dict(address=self.floating_ip,
fixed_address='11.11.11.11'))
req = self._get_fake_request()
req.method = "POST"
req.body = jsonutils.dumps(body)
req.headers["content-type"] = "application/json"
resp = req.get_response(self.app)
res_dict = jsonutils.loads(resp.body)
self.assertEqual(resp.status_int, 400)
self.assertEqual(res_dict['badRequest']['message'],
"Specified fixed address not assigned to instance")
ex = self.assertRaises(webob.exc.HTTPBadRequest,
self.manager._add_floating_ip,
self.fake_req, 'test_inst', body)
self.assertIn("Specified fixed address not assigned to instance",
ex.explanation)
class ExtendedFloatingIpTestV2(ExtendedFloatingIpTestV21):
floating_ips = fips_v2
def _get_fake_app(self):
return fakes.wsgi_app(init_only=('servers', 'os-floating-ips'))
def setUp(self):
super(ExtendedFloatingIpTestV2, self).setUp()
self.flags(
osapi_compute_extension=[
'nova.api.openstack.compute.contrib.select_extensions'],
osapi_compute_ext_list=['Floating_ips', 'Extended_floating_ips'])