openstack/nova
Code Issues Proposed changes

Merge "nova-net: Convert remaining API tests to use neutron"

Browse Source
This commit is contained in:
Zuul
2019-12-11 23:17:23 +00:00
committed by Gerrit Code Review
parent 17a8d3a5fc 3d4a468c38
commit 4a36ae9684
5 changed files with 52 additions and 880 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
nova/tests/unit/api/openstack/compute/test_access_ips.py
View File

@@ -12,6 +12,8 @@
# License for the specific language governing permissions and limitations
# under the License.
from oslo_serialization import jsonutils
from nova.api.openstack.compute import servers as servers_v21
from nova import exception
from nova import test
@@ -33,8 +35,7 @@ class AccessIPsAPIValidationTestV21(test.TestCase):
def fake_rebuild(*args, **kwargs):
pass
# Neutron security groups are tested in test_neutron_security_groups.py
self.flags(use_neutron=False)
fakes.stub_out_nw_api(self)
self._set_up_controller()
fake.stub_out_image_service(self)
@@ -44,8 +45,6 @@ class AccessIPsAPIValidationTestV21(test.TestCase):
self.stub_out('nova.objects.instance.Instance.save', fake_save)
self.stub_out('nova.compute.api.API.rebuild', fake_rebuild)
self.req = fakes.HTTPRequest.blank('')
def _set_up_controller(self):
self.controller = servers_v21.ServersController()
@@ -63,7 +62,11 @@ class AccessIPsAPIValidationTestV21(test.TestCase):
},
}
body['server'].update(params)
res_dict = self.controller.create(self.req, body=body).obj
req = fakes.HTTPRequest.blank('')
req.method = 'POST'
req.body = jsonutils.dump_as_bytes(body)
req.headers['content-type'] = 'application/json'
res_dict = self.controller.create(req, body=body).obj
return res_dict
def _test_update(self, params):
@@ -72,8 +75,9 @@ class AccessIPsAPIValidationTestV21(test.TestCase):
},
}
body['server'].update(params)
req = fakes.HTTPRequest.blank('')
res_dict = self.controller.update(self.req, fakes.FAKE_UUID, body=body)
res_dict = self.controller.update(req, fakes.FAKE_UUID, body=body)
self._verify_update_access_ip(res_dict, params)
def _test_rebuild(self, params):
@@ -83,7 +87,8 @@ class AccessIPsAPIValidationTestV21(test.TestCase):
},
}
body['rebuild'].update(params)
self.controller._action_rebuild(self.req, fakes.FAKE_UUID, body=body)
req = fakes.HTTPRequest.blank('')
self.controller._action_rebuild(req, fakes.FAKE_UUID, body=body)
def test_create_server_with_access_ipv4(self):
params = {v4_key: '192.168.0.10'}

27
nova/tests/unit/api/openstack/compute/test_availability_zone.py
View File

@@ -16,6 +16,7 @@ import datetime
import iso8601
import mock
from oslo_serialization import jsonutils
from oslo_utils.fixture import uuidsentinel
from nova.api.openstack.compute import availability_zone as az_v21
@@ -96,7 +97,6 @@ class AvailabilityZoneApiTestV21(test.NoDBTestCase):
self.controller.host_api, 'service_get_all',
side_effect=fake_service_get_all).start()
self.addCleanup(self.mock_service_get_all.stop)
self.req = fakes.HTTPRequest.blank('')
def test_filtered_availability_zones(self):
zones = ['zone1', 'internal']
@@ -114,7 +114,8 @@ class AvailabilityZoneApiTestV21(test.NoDBTestCase):
self.assertEqual(result, expected)
def test_availability_zone_index(self):
resp_dict = self.controller.index(self.req)
req = fakes.HTTPRequest.blank('')
resp_dict = self.controller.index(req)
self.assertIn('availabilityZoneInfo', resp_dict)
zones = resp_dict['availabilityZoneInfo']
@@ -127,7 +128,8 @@ class AvailabilityZoneApiTestV21(test.NoDBTestCase):
self.assertIsNone(zones[1]['hosts'])
def test_availability_zone_detail(self):
resp_dict = self.controller.detail(self.req)
req = fakes.HTTPRequest.blank('')
resp_dict = self.controller.detail(req)
self.assertIn('availabilityZoneInfo', resp_dict)
zones = resp_dict['availabilityZoneInfo']
@@ -184,7 +186,8 @@ class AvailabilityZoneApiTestV21(test.NoDBTestCase):
[{'zoneState': {'available': True},
'hosts': {},
'zoneName': 'nova'}]}
resp_dict = self.controller.detail(self.req)
req = fakes.HTTPRequest.blank('')
resp_dict = self.controller.detail(req)
self.assertThat(resp_dict,
matchers.DictMatches(expected_response))
@@ -198,8 +201,6 @@ class ServersControllerCreateTestV21(test.TestCase):
super(ServersControllerCreateTestV21, self).setUp()
self.instance_cache_num = 0
# Neutron security groups are tested in test_neutron_security_groups.py
self.flags(use_neutron=False)
fakes.stub_out_nw_api(self)
self._set_up_controller()
@@ -211,7 +212,6 @@ class ServersControllerCreateTestV21(test.TestCase):
fake.stub_out_image_service(self)
self.stub_out('nova.compute.api.API.create_db_entry_for_new_instance',
create_db_entry_for_new_instance)
self.req = fakes.HTTPRequest.blank('')
def _set_up_controller(self):
self.controller = servers_v21.ServersController()
@@ -239,6 +239,11 @@ class ServersControllerCreateTestV21(test.TestCase):
},
}
req = fakes.HTTPRequest.blank('')
req.method = 'POST'
req.body = jsonutils.dump_as_bytes(body)
req.headers['content-type'] = 'application/json'
admin_context = context.get_admin_context()
db.service_create(admin_context, {'host': 'host1_zones',
'binary': "nova-compute",
@@ -250,7 +255,7 @@ class ServersControllerCreateTestV21(test.TestCase):
metadata={'availability_zone': 'nova'})
agg.create()
agg.add_host('host1_zones')
return self.req, body
return req, body
def test_create_instance_with_availability_zone(self):
zone_name = 'nova'
@@ -291,7 +296,11 @@ class ServersControllerCreateTestV21(test.TestCase):
},
},
}
req = fakes.HTTPRequest.blank('')
req.method = 'POST'
req.body = jsonutils.dump_as_bytes(body)
req.headers['content-type'] = 'application/json'
res = self.controller.create(self.req, body=body).obj
res = self.controller.create(req, body=body).obj
server = res['server']
self.assertEqual(fakes.FAKE_UUID, server['id'])

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

@@ -16,7 +16,6 @@
import mock
from oslo_utils.fixture import uuidsentinel as uuids
import six
import webob
from nova.api.openstack.compute import floating_ips as fips_v21
@@ -26,7 +25,6 @@ from nova.db import api as db
from nova import exception
from nova import network
from nova import objects
from nova.objects import base as obj_base
from nova import test
from nova.tests.unit.api.openstack import fakes
from nova.tests.unit import fake_network
@@ -107,12 +105,11 @@ def get_instance_by_floating_ip_addr(self, context, address):
return None
class FloatingIpTestNeutronV21(test.NoDBTestCase):
class FloatingIpTestV21(test.NoDBTestCase):
floating_ips = fips_v21
def setUp(self):
super(FloatingIpTestNeutronV21, self).setUp()
self.flags(use_neutron=True)
super(FloatingIpTestV21, self).setUp()
self.controller = self.floating_ips.FloatingIPController()
def test_floatingip_delete(self):
@@ -198,715 +195,6 @@ class FloatingIpTestNeutronV21(test.NoDBTestCase):
expected_exc)
class FloatingIpTestV21(test.TestCase):
floating_ip = "10.10.10.10"
floating_ip_2 = "10.10.10.11"
floating_ips = fips_v21
validation_error = exception.ValidationError
def _create_floating_ips(self, floating_ips=None):
"""Create a floating IP object."""
if floating_ips is None:
floating_ips = [self.floating_ip]
elif not isinstance(floating_ips, (list, tuple)):
floating_ips = [floating_ips]
dict_ = {'pool': 'nova', 'host': 'fake_host'}
return db.floating_ip_bulk_create(
self.context, [dict(address=ip, **dict_) for ip in floating_ips],
)
def _delete_floating_ip(self):
db.floating_ip_destroy(self.context, self.floating_ip)
def setUp(self):
super(FloatingIpTestV21, self).setUp()
self.flags(use_neutron=False)
self.stubs.Set(compute.api.API, "get",
compute_api_get)
self.stubs.Set(network.api.API, "get_floating_ip",
network_api_get_floating_ip)
self.stubs.Set(network.api.API, "get_floating_ip_by_address",
network_api_get_floating_ip_by_address)
self.stubs.Set(network.api.API, "get_floating_ips_by_project",
network_api_get_floating_ips_by_project)
self.stubs.Set(network.api.API, "release_floating_ip",
network_api_release)
self.stubs.Set(network.api.API, "disassociate_floating_ip",
network_api_disassociate)
self.stubs.Set(network.api.API, "get_instance_id_by_floating_address",
get_instance_by_floating_ip_addr)
self.stubs.Set(objects.Instance, "get_network_info",
stub_nw_info(self))
fake_network.stub_out_nw_api_get_instance_nw_info(self)
self.stub_out('nova.db.api.instance_get',
fake_instance_get)
self.context = context.get_admin_context()
self._create_floating_ips()
self.controller = self.floating_ips.FloatingIPController()
self.manager = self.floating_ips.\
FloatingIPActionController()
self.fake_req = fakes.HTTPRequest.blank('')
def tearDown(self):
self._delete_floating_ip()
super(FloatingIpTestV21, self).tearDown()
def test_floatingip_delete(self):
fip_val = {'address': '1.1.1.1', 'fixed_ip_id': '192.168.1.2'}
with test.nested(
mock.patch.object(self.controller.network_api,
'disassociate_floating_ip'),
mock.patch.object(self.controller.network_api,
'release_floating_ip'),
mock.patch.object(self.controller.network_api,
'get_instance_id_by_floating_address',
return_value=None),
mock.patch.object(self.controller.network_api,
'get_floating_ip',
return_value=fip_val)) as (
disoc_fip, rel_fip, _, _):
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):
with mock.patch.object(self.controller.network_api,
'get_floating_ip', side_effect=ex):
self.assertRaises(expect_ex,
self.controller.delete, self.fake_req, 1)
def test_floatingip_delete_not_found_ip(self):
ex = exception.FloatingIpNotFound(id=1)
self._test_floatingip_delete_not_found(ex)
def test_floatingip_delete_not_found(self):
ex = exception.NotFound
self._test_floatingip_delete_not_found(ex)
def test_floatingip_delete_invalid_id(self):
ex = exception.InvalidID(id=1)
self._test_floatingip_delete_not_found(ex, webob.exc.HTTPBadRequest)
def test_translate_floating_ip_view(self):
floating_ip_address = self.floating_ip
floating_ip = db.floating_ip_get_by_address(self.context,
floating_ip_address)
# NOTE(vish): network_get uses the id not the address
floating_ip = db.floating_ip_get(self.context, floating_ip['id'])
floating_obj = objects.FloatingIP()
objects.FloatingIP._from_db_object(self.context, floating_obj,
floating_ip)
view = self.floating_ips._translate_floating_ip_view(floating_obj)
self.assertIn('floating_ip', view)
self.assertTrue(view['floating_ip']['id'])
self.assertEqual(view['floating_ip']['ip'], floating_obj.address)
self.assertIsNone(view['floating_ip']['fixed_ip'])
self.assertIsNone(view['floating_ip']['instance_id'])
def test_translate_floating_ip_view_neutronesque(self):
uuid = 'ca469a10-fa76-11e5-86aa-5e5517507c66'
fixed_id = 'ae900cf4-fb73-11e5-86aa-5e5517507c66'
floating_ip = objects.floating_ip.NeutronFloatingIP(id=uuid,
address='1.2.3.4', pool='pool', context='ctxt',
fixed_ip_id=fixed_id)
view = self.floating_ips._translate_floating_ip_view(floating_ip)
self.assertEqual(uuid, view['floating_ip']['id'])
def test_translate_floating_ip_view_dict(self):
floating_ip = {'id': 0, 'address': '10.0.0.10', 'pool': 'nova',
'fixed_ip': None}
view = self.floating_ips._translate_floating_ip_view(floating_ip)
self.assertIn('floating_ip', view)
def test_translate_floating_ip_view_obj(self):
fip = objects.FixedIP(address='192.168.1.2', instance_uuid=FAKE_UUID)
floater = self._build_floating_ip('10.0.0.2', fip)
result = self.floating_ips._translate_floating_ip_view(floater)
expected = self._build_expected(floater, fip.address,
fip.instance_uuid)
self._test_result(expected, result)
def test_translate_floating_ip_bad_address(self):
fip = objects.FixedIP(instance_uuid=FAKE_UUID)
floater = self._build_floating_ip('10.0.0.2', fip)
result = self.floating_ips._translate_floating_ip_view(floater)
expected = self._build_expected(floater, None, fip.instance_uuid)
self._test_result(expected, result)
def test_translate_floating_ip_bad_instance_id(self):
fip = objects.FixedIP(address='192.168.1.2')
floater = self._build_floating_ip('10.0.0.2', fip)
result = self.floating_ips._translate_floating_ip_view(floater)
expected = self._build_expected(floater, fip.address, None)
self._test_result(expected, result)
def test_translate_floating_ip_bad_instance_and_address(self):
fip = objects.FixedIP()
floater = self._build_floating_ip('10.0.0.2', fip)
result = self.floating_ips._translate_floating_ip_view(floater)
expected = self._build_expected(floater, None, None)
self._test_result(expected, result)
def test_translate_floating_ip_null_fixed(self):
floater = self._build_floating_ip('10.0.0.2', None)
result = self.floating_ips._translate_floating_ip_view(floater)
expected = self._build_expected(floater, None, None)
self._test_result(expected, result)
def test_translate_floating_ip_unset_fixed(self):
floater = objects.FloatingIP(id=1, address='10.0.0.2', pool='foo')
result = self.floating_ips._translate_floating_ip_view(floater)
expected = self._build_expected(floater, None, None)
self._test_result(expected, result)
def test_translate_floating_ips_view(self):
mock_trans = mock.Mock()
mock_trans.return_value = {'floating_ip': 'foo'}
self.floating_ips._translate_floating_ip_view = mock_trans
fip1 = objects.FixedIP(address='192.168.1.2', instance_uuid=FAKE_UUID)
fip2 = objects.FixedIP(address='192.168.1.3', instance_uuid=FAKE_UUID)
floaters = [self._build_floating_ip('10.0.0.2', fip1),
self._build_floating_ip('10.0.0.3', fip2)]
result = self.floating_ips._translate_floating_ips_view(floaters)
called_floaters = [call[0][0] for call in mock_trans.call_args_list]
self.assertTrue(any(obj_base.obj_equal_prims(floaters[0], f)
for f in called_floaters),
"_translate_floating_ip_view was not called with all "
"floating ips")
self.assertTrue(any(obj_base.obj_equal_prims(floaters[1], f)
for f in called_floaters),
"_translate_floating_ip_view was not called with all "
"floating ips")
expected_result = {'floating_ips': ['foo', 'foo']}
self.assertEqual(expected_result, result)
def test_floating_ips_list(self):
res_dict = self.controller.index(self.fake_req)
response = {'floating_ips': [{'instance_id': FAKE_UUID,
'ip': '10.10.10.10',
'pool': 'nova',
'fixed_ip': '10.0.0.1',
'id': 1},
{'instance_id': None,
'ip': '10.10.10.11',
'pool': 'nova',
'fixed_ip': None,
'id': 2}]}
self.assertEqual(res_dict, response)
def test_floating_ip_release_nonexisting(self):
def fake_get_floating_ip(*args, **kwargs):
raise exception.FloatingIpNotFound(id=id)
self.stubs.Set(network.api.API, "get_floating_ip",
fake_get_floating_ip)
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):
return {'fixed_ip_id': 1, 'address': self.floating_ip}
def fake_get_instance_by_floating_ip_addr(*args, **kwargs):
return 'test-inst'
def fake_disassociate_floating_ip(*args, **kwargs):
raise exception.FloatingIpNotAssociated(args[3])
self.stubs.Set(network.api.API, "get_floating_ip",
fake_get_floating_ip)
self.stubs.Set(self.floating_ips, "get_instance_by_floating_ip_addr",
fake_get_instance_by_floating_ip_addr)
self.stubs.Set(self.floating_ips, "disassociate_floating_ip",
fake_disassociate_floating_ip)
delete = self.controller.delete
res = 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 = delete.wsgi_code
else:
status_int = res.status_int
self.assertEqual(status_int, 202)
def test_floating_ip_show(self):
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')
self.assertIsNone(res_dict['floating_ip']['instance_id'])
def test_floating_ip_show_not_found(self):
def fake_get_floating_ip(*args, **kwargs):
raise exception.FloatingIpNotFound(id='fake')
self.stubs.Set(network.api.API, "get_floating_ip",
fake_get_floating_ip)
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):
return {'id': 1, 'address': '10.10.10.10', 'pool': 'nova',
'fixed_ip': {'address': '10.0.0.1',
'instance_uuid': FAKE_UUID,
'instance': {'uuid': FAKE_UUID}}}
self.stubs.Set(network.api.API, "get_floating_ip", get_floating_ip)
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')
self.assertEqual(res_dict['floating_ip']['fixed_ip'], '10.0.0.1')
self.assertEqual(res_dict['floating_ip']['instance_id'], FAKE_UUID)
def test_recreation_of_floating_ip(self):
self._delete_floating_ip()
self._create_floating_ips()
def test_floating_ip_in_bulk_creation(self):
self._delete_floating_ip()
self._create_floating_ips([self.floating_ip, self.floating_ip_2])
all_ips = db.floating_ip_get_all(self.context)
ip_list = [ip['address'] for ip in all_ips]
self.assertIn(self.floating_ip, ip_list)
self.assertIn(self.floating_ip_2, ip_list)
def test_fail_floating_ip_in_bulk_creation(self):
self.assertRaises(exception.FloatingIpExists,
self._create_floating_ips,
[self.floating_ip, self.floating_ip_2])
all_ips = db.floating_ip_get_all(self.context)
ip_list = [ip['address'] for ip in all_ips]
self.assertIn(self.floating_ip, ip_list)
self.assertNotIn(self.floating_ip_2, ip_list)
def test_floating_ip_allocate_no_free_ips(self):
def fake_allocate(*args, **kwargs):
raise exception.NoMoreFloatingIps()
self.stubs.Set(network.api.API, "allocate_floating_ip", fake_allocate)
ex = self.assertRaises(webob.exc.HTTPNotFound,
self.controller.create, self.fake_req)
self.assertIn('No more floating IPs', ex.explanation)
def test_floating_ip_allocate_no_free_ips_pool(self):
def fake_allocate(*args, **kwargs):
raise exception.NoMoreFloatingIps()
self.stubs.Set(network.api.API, "allocate_floating_ip", fake_allocate)
ex = self.assertRaises(webob.exc.HTTPNotFound,
self.controller.create, self.fake_req,
{'pool': 'non_existent_pool'})
self.assertIn('No more floating IPs in pool non_existent_pool',
ex.explanation)
@mock.patch.object(network.api.API, 'allocate_floating_ip',
side_effect=exception.FloatingIpBadRequest(
'Bad floatingip request: Network '
'c8f0e88f-ae41-47cb-be6c-d8256ba80576 does not contain any '
'IPv4 subnet'))
def test_floating_ip_allocate_no_ipv4_subnet(self, allocate_mock):
ex = self.assertRaises(webob.exc.HTTPBadRequest,
self.controller.create, self.fake_req,
{'pool': 'non_existent_pool'})
self.assertIn("does not contain any IPv4 subnet",
six.text_type(ex))
@mock.patch('nova.network.api.API.allocate_floating_ip',
side_effect=exception.FloatingIpLimitExceeded())
def test_floating_ip_allocate_over_quota(self, allocate_mock):
ex = self.assertRaises(webob.exc.HTTPForbidden,
self.controller.create, self.fake_req)
self.assertIn('IP allocation over quota', ex.explanation)
@mock.patch('nova.objects.FloatingIP.deallocate')
@mock.patch('nova.objects.FloatingIP.allocate_address')
@mock.patch('nova.objects.quotas.Quotas.check_deltas')
def test_floating_ip_allocate_over_quota_during_recheck(self, check_mock,
alloc_mock,
dealloc_mock):
ctxt = self.fake_req.environ['nova.context']
# Simulate a race where the first check passes and the recheck fails.
check_mock.side_effect = [None,
exception.OverQuota(overs='floating_ips')]
self.assertRaises(webob.exc.HTTPForbidden,
self.controller.create, self.fake_req)
self.assertEqual(2, check_mock.call_count)
call1 = mock.call(ctxt, {'floating_ips': 1}, ctxt.project_id)
call2 = mock.call(ctxt, {'floating_ips': 0}, ctxt.project_id)
check_mock.assert_has_calls([call1, call2])
# Verify we removed the floating IP that was added after the first
# quota check passed.
dealloc_mock.assert_called_once_with(ctxt,
alloc_mock.return_value.address)
@mock.patch('nova.objects.FloatingIP.allocate_address')
@mock.patch('nova.objects.quotas.Quotas.check_deltas')
def test_floating_ip_allocate_no_quota_recheck(self, check_mock,
alloc_mock):
# Disable recheck_quota.
self.flags(recheck_quota=False, group='quota')
ctxt = self.fake_req.environ['nova.context']
self.controller.create(self.fake_req)
# check_deltas should have been called only once.
check_mock.assert_called_once_with(ctxt, {'floating_ips': 1},
ctxt.project_id)
@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):
ex = self.assertRaises(webob.exc.HTTPForbidden,
self.controller.create, self.fake_req,
{'pool': 'non_existent_pool'})
self.assertIn('IP allocation over quota in pool non_existent_pool.',
ex.explanation)
@mock.patch('nova.network.api.API.allocate_floating_ip',
side_effect=exception.FloatingIpPoolNotFound())
def test_floating_ip_create_with_unknown_pool(self, allocate_mock):
ex = self.assertRaises(webob.exc.HTTPNotFound,
self.controller.create, self.fake_req,
{'pool': 'non_existent_pool'})
self.assertIn('Floating IP pool not found.', ex.explanation)
def test_floating_ip_allocate(self):
def fake1(*args, **kwargs):
pass
def fake2(*args, **kwargs):
return {'id': 1, 'address': '10.10.10.10', 'pool': 'nova'}
self.stubs.Set(network.api.API, "allocate_floating_ip",
fake1)
self.stubs.Set(network.api.API, "get_floating_ip_by_address",
fake2)
res_dict = self.controller.create(self.fake_req)
ip = res_dict['floating_ip']
expected = {
"id": 1,
"instance_id": None,
"ip": "10.10.10.10",
"fixed_ip": None,
"pool": 'nova'}
self.assertEqual(ip, expected)
def test_floating_ip_release(self):
self.controller.delete(self.fake_req, 1)
def _test_floating_ip_associate(self, fixed_address):
def fake_associate_floating_ip(*args, **kwargs):
self.assertEqual(fixed_address, kwargs['fixed_address'])
self.stubs.Set(network.api.API, "associate_floating_ip",
fake_associate_floating_ip)
body = dict(addFloatingIp=dict(address=self.floating_ip))
rsp = self.manager._add_floating_ip(self.fake_req, TEST_INST,
body=body)
self.assertEqual(202, rsp.status_int)
def test_floating_ip_associate(self):
self._test_floating_ip_associate(fixed_address='192.168.1.100')
@mock.patch.object(network.model.NetworkInfo, 'fixed_ips')
def test_associate_floating_ip_v4v6_fixed_ip(self, fixed_ips_mock):
fixed_address = '192.168.1.100'
fixed_ips_mock.return_value = [{'address': 'fc00:2001:db8::100'},
{'address': ''},
{'address': fixed_address}]
self._test_floating_ip_associate(fixed_address=fixed_address)
@mock.patch.object(network.model.NetworkInfo, 'fixed_ips',
return_value=[{'address': 'fc00:2001:db8::100'}])
def test_associate_floating_ip_v6_fixed_ip(self, fixed_ips_mock):
body = dict(addFloatingIp=dict(address=self.floating_ip))
self.assertRaises(webob.exc.HTTPBadRequest,
self.manager._add_floating_ip, self.fake_req,
TEST_INST, body=body)
def test_floating_ip_associate_invalid_instance(self):
def fake_get(self, context, id, expected_attrs=None,
cell_down_support=False):
raise exception.InstanceNotFound(instance_id=id)
self.stubs.Set(compute.api.API, "get", fake_get)
body = dict(addFloatingIp=dict(address=self.floating_ip))
self.assertRaises(webob.exc.HTTPNotFound,
self.manager._add_floating_ip, self.fake_req,
'test_inst', body=body)
def test_associate_not_allocated_floating_ip_to_instance(self):
def fake_associate_floating_ip(self, context, instance,
floating_address, fixed_address,
affect_auto_assigned=False):
raise exception.FloatingIpNotFoundForAddress(
address=floating_address)
self.stubs.Set(network.api.API, "associate_floating_ip",
fake_associate_floating_ip)
floating_ip = '10.10.10.11'
body = dict(addFloatingIp=dict(address=floating_ip))
ex = self.assertRaises(webob.exc.HTTPNotFound,
self.manager._add_floating_ip,
self.fake_req, TEST_INST, body=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'))
self.assertRaises(webob.exc.HTTPForbidden,
self.manager._add_floating_ip, self.fake_req,
TEST_INST, body=body)
@mock.patch.object(network.api.API, 'associate_floating_ip',
side_effect=exception.FloatingIpAssociateFailed(
address='10.10.10.11'))
def test_associate_floating_ip_failed(self, associate_mock):
body = dict(addFloatingIp=dict(address='10.10.10.11'))
self.assertRaises(webob.exc.HTTPBadRequest,
self.manager._add_floating_ip, self.fake_req,
TEST_INST, body=body)
def test_associate_floating_ip_bad_address_key(self):
body = dict(addFloatingIp=dict(bad_address='10.10.10.11'))
req = fakes.HTTPRequest.blank(
'/v2/%s/servers/test_inst/action' % fakes.FAKE_PROJECT_ID)
self.assertRaises(self.validation_error,
self.manager._add_floating_ip, req, 'test_inst',
body=body)
def test_associate_floating_ip_bad_addfloatingip_key(self):
body = dict(bad_addFloatingIp=dict(address='10.10.10.11'))
req = fakes.HTTPRequest.blank(
'/v2/%s/servers/test_inst/action' % fakes.FAKE_PROJECT_ID)
self.assertRaises(self.validation_error,
self.manager._add_floating_ip, req, 'test_inst',
body=body)
def test_floating_ip_disassociate(self):
def get_instance_by_floating_ip_addr(self, context, address):
if address == '10.10.10.10':
return TEST_INST
self.stubs.Set(network.api.API, "get_instance_id_by_floating_address",
get_instance_by_floating_ip_addr)
body = dict(removeFloatingIp=dict(address='10.10.10.10'))
rsp = self.manager._remove_floating_ip(self.fake_req, TEST_INST,
body=body)
self.assertEqual(202, rsp.status_int)
def test_floating_ip_disassociate_missing(self):
body = dict(removeFloatingIp=dict(address='10.10.10.10'))
self.assertRaises(webob.exc.HTTPConflict,
self.manager._remove_floating_ip,
self.fake_req, 'test_inst', body=body)
def test_floating_ip_associate_non_existent_ip(self):
def fake_network_api_associate(self, context, instance,
floating_address=None,
fixed_address=None):
floating_ips = ["10.10.10.10", "10.10.10.11"]
if floating_address not in floating_ips:
raise exception.FloatingIpNotFoundForAddress(
address=floating_address)
self.stubs.Set(network.api.API, "associate_floating_ip",
fake_network_api_associate)
body = dict(addFloatingIp=dict(address='1.1.1.1'))
self.assertRaises(webob.exc.HTTPNotFound,
self.manager._add_floating_ip,
self.fake_req, TEST_INST, body=body)
def test_floating_ip_disassociate_non_existent_ip(self):
def network_api_get_floating_ip_by_address(self, context,
floating_address):
floating_ips = ["10.10.10.10", "10.10.10.11"]
if floating_address not in floating_ips:
raise exception.FloatingIpNotFoundForAddress(
address=floating_address)
self.stubs.Set(network.api.API, "get_floating_ip_by_address",
network_api_get_floating_ip_by_address)
body = dict(removeFloatingIp=dict(address='1.1.1.1'))
self.assertRaises(webob.exc.HTTPNotFound,
self.manager._remove_floating_ip,
self.fake_req, TEST_INST, body=body)
def test_floating_ip_disassociate_wrong_instance_uuid(self):
def get_instance_by_floating_ip_addr(self, context, address):
if address == '10.10.10.10':
return TEST_INST
self.stubs.Set(network.api.API, "get_instance_id_by_floating_address",
get_instance_by_floating_ip_addr)
wrong_uuid = 'aaaaaaaa-ffff-ffff-ffff-aaaaaaaaaaaa'
body = dict(removeFloatingIp=dict(address='10.10.10.10'))
self.assertRaises(webob.exc.HTTPConflict,
self.manager._remove_floating_ip,
self.fake_req, wrong_uuid, body=body)
def test_floating_ip_disassociate_wrong_instance_id(self):
def get_instance_by_floating_ip_addr(self, context, address):
if address == '10.10.10.10':
return WRONG_INST
self.stubs.Set(network.api.API, "get_instance_id_by_floating_address",
get_instance_by_floating_ip_addr)
body = dict(removeFloatingIp=dict(address='10.10.10.10'))
self.assertRaises(webob.exc.HTTPConflict,
self.manager._remove_floating_ip,
self.fake_req, TEST_INST, body=body)
def test_floating_ip_disassociate_auto_assigned(self):
def fake_get_floating_ip_addr_auto_assigned(self, context, address):
return {'id': 1, 'address': '10.10.10.10', 'pool': 'nova',
'fixed_ip_id': 10, 'auto_assigned': 1}
def get_instance_by_floating_ip_addr(self, context, address):
if address == '10.10.10.10':
return TEST_INST
def network_api_disassociate(self, context, instance,
floating_address):
raise exception.CannotDisassociateAutoAssignedFloatingIP()
self.stubs.Set(network.api.API, "get_floating_ip_by_address",
fake_get_floating_ip_addr_auto_assigned)
self.stubs.Set(network.api.API, "get_instance_id_by_floating_address",
get_instance_by_floating_ip_addr)
self.stubs.Set(network.api.API, "disassociate_floating_ip",
network_api_disassociate)
body = dict(removeFloatingIp=dict(address='10.10.10.10'))
self.assertRaises(webob.exc.HTTPForbidden,
self.manager._remove_floating_ip,
self.fake_req, TEST_INST, body=body)
def test_floating_ip_disassociate_map_authorization_exc(self):
def fake_get_floating_ip_addr_auto_assigned(self, context, address):
return {'id': 1, 'address': '10.10.10.10', 'pool': 'nova',
'fixed_ip_id': 10, 'auto_assigned': 1}
def get_instance_by_floating_ip_addr(self, context, address):
if address == '10.10.10.10':
return TEST_INST
def network_api_disassociate(self, context, instance, address):
raise exception.Forbidden()
self.stubs.Set(network.api.API, "get_floating_ip_by_address",
fake_get_floating_ip_addr_auto_assigned)
self.stubs.Set(network.api.API, "get_instance_id_by_floating_address",
get_instance_by_floating_ip_addr)
self.stubs.Set(network.api.API, "disassociate_floating_ip",
network_api_disassociate)
body = dict(removeFloatingIp=dict(address='10.10.10.10'))
self.assertRaises(webob.exc.HTTPForbidden,
self.manager._remove_floating_ip,
self.fake_req, TEST_INST, body=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'))
self.assertRaises(self.validation_error,
self.manager._remove_floating_ip, self.fake_req,
TEST_INST, body=body)
def test_missing_dict_param_in_remove_floating_ip(self):
body = dict(removeFloatingIp='11.0.0.1')
self.assertRaises(self.validation_error,
self.manager._remove_floating_ip, self.fake_req,
TEST_INST, body=body)
def test_missing_dict_param_in_add_floating_ip(self):
body = dict(addFloatingIp='11.0.0.1')
self.assertRaises(self.validation_error,
self.manager._add_floating_ip, self.fake_req,
TEST_INST, body=body)
def _build_floating_ip(self, address, fixed_ip):
floating = objects.FloatingIP(id=1, address=address, pool='foo',
fixed_ip=fixed_ip)
return floating
def _build_expected(self, floating_ip, fixed_ip, instance_id):
return {'floating_ip': {'id': floating_ip.id,
'ip': floating_ip.address,
'pool': floating_ip.pool,
'fixed_ip': fixed_ip,
'instance_id': instance_id}}
def _test_result(self, expected, actual):
expected_fl = expected['floating_ip']
actual_fl = actual['floating_ip']
self.assertEqual(expected_fl, actual_fl)
class ExtendedFloatingIpTestV21(test.TestCase):
floating_ip = "10.10.10.10"
floating_ip_2 = "10.10.10.11"

166
nova/tests/unit/api/openstack/compute/test_serversV21.py
View File

@@ -228,22 +228,14 @@ class ControllerTest(test.TestCase):
def setUp(self):
super(ControllerTest, self).setUp()
self.flags(use_ipv6=False)
# Neutron security groups are tested in test_neutron_security_groups.py
self.flags(use_neutron=False)
fakes.stub_out_nw_api(self)
fakes.stub_out_key_pair_funcs(self)
fake.stub_out_image_service(self)
security_groups = [
{'name': 'fake-0-0', 'id': 1, 'description': 'foo',
'user_id': 'bar', 'project_id': 'baz', 'deleted': False,
'deleted_at': None, 'updated_at': None, 'created_at': None},
{'name': 'fake-0-1', 'id': 1, 'description': 'foo',
'user_id': 'bar', 'project_id': 'baz', 'deleted': False,
'deleted_at': None, 'updated_at': None, 'created_at': None}]
fakes.stub_out_secgroup_api(
self, security_groups=[{'name': 'default'}])
return_server = fakes.fake_compute_get(id=2, availability_zone='nova',
launched_at=None,
terminated_at=None,
security_groups=security_groups,
task_state=None,
vm_state=vm_states.ACTIVE,
power_state=1)
@@ -305,7 +297,6 @@ class ServersControllerTest(ControllerTest):
"""Tests that we no longer support the legacy br-<uuid> format for
a network id.
"""
self.flags(use_neutron=True)
uuid = 'br-00000000-0000-0000-0000-000000000000'
requested_networks = [{'uuid': uuid}]
ex = self.assertRaises(webob.exc.HTTPBadRequest,
@@ -315,40 +306,25 @@ class ServersControllerTest(ControllerTest):
'format', six.text_type(ex))
def test_requested_networks_neutronv2_enabled_with_port(self):
self.flags(use_neutron=True)
port = 'eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee'
requested_networks = [{'port': port}]
res = self.controller._get_requested_networks(requested_networks)
self.assertEqual([(None, None, port, None)], res.as_tuples())
def test_requested_networks_neutronv2_enabled_with_network(self):
self.flags(use_neutron=True)
network = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
requested_networks = [{'uuid': network}]
res = self.controller._get_requested_networks(requested_networks)
self.assertEqual([(network, None, None, None)], res.as_tuples())
def test_requested_networks_neutronv2_enabled_with_network_and_port(self):
self.flags(use_neutron=True)
network = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
port = 'eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee'
requested_networks = [{'uuid': network, 'port': port}]
res = self.controller._get_requested_networks(requested_networks)
self.assertEqual([(None, None, port, None)], res.as_tuples())
def test_requested_networks_with_duplicate_networks_nova_net(self):
# duplicate networks are allowed only for nova neutron v2.0
self.flags(use_neutron=False)
network = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
requested_networks = [{'uuid': network}, {'uuid': network}]
self.assertRaises(
webob.exc.HTTPBadRequest,
self.controller._get_requested_networks,
requested_networks)
def test_requested_networks_with_neutronv2_and_duplicate_networks(self):
# duplicate networks are allowed only for nova neutron v2.0
self.flags(use_neutron=True)
network = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
requested_networks = [{'uuid': network}, {'uuid': network}]
res = self.controller._get_requested_networks(requested_networks)
@@ -356,7 +332,6 @@ class ServersControllerTest(ControllerTest):
(network, None, None, None)], res.as_tuples())
def test_requested_networks_neutronv2_enabled_conflict_on_fixed_ip(self):
self.flags(use_neutron=True)
network = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
port = 'eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee'
addr = '10.0.0.1'
@@ -368,17 +343,7 @@ class ServersControllerTest(ControllerTest):
self.controller._get_requested_networks,
requested_networks)
def test_requested_networks_neutronv2_disabled_with_port(self):
self.flags(use_neutron=False)
port = 'eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee'
requested_networks = [{'port': port}]
self.assertRaises(
webob.exc.HTTPBadRequest,
self.controller._get_requested_networks,
requested_networks)
def test_requested_networks_api_enabled_with_v2_subclass(self):
self.flags(use_neutron=True)
network = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
port = 'eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee'
requested_networks = [{'uuid': network, 'port': port}]
@@ -487,8 +452,7 @@ class ServersControllerTest(ControllerTest):
"key_name": '',
"OS-SRV-USG:launched_at": None,
"OS-SRV-USG:terminated_at": None,
"security_groups": [{'name': 'fake-0-0'},
{'name': 'fake-0-1'}],
"security_groups": [{'name': 'default'}],
"OS-EXT-STS:task_state": None,
"OS-EXT-STS:vm_state": vm_states.ACTIVE,
"OS-EXT-STS:power_state": 1,
@@ -1677,7 +1641,6 @@ class ServersControllerTest(ControllerTest):
DATE2 = datetime.datetime(year=2013, month=4, day=5, hour=13)
self.mock_get.side_effect = fakes.fake_compute_get(
id=1, uuid=FAKE_UUID, launched_at=DATE1, terminated_at=DATE2)
fakes.stub_out_secgroup_api(self)
req = self.req(self.path_with_id % FAKE_UUID)
req.accept = 'application/json'
req.method = 'GET'
@@ -1704,7 +1667,6 @@ class ServersControllerTest(ControllerTest):
]
return objects.InstanceList(objects=db_list)
self.mock_get_all.side_effect = fake_compute_get_all
fakes.stub_out_secgroup_api(self)
req = self.req(self.path_detail)
req.accept = 'application/json'
servers = req.get_response(compute.APIRouterV21())
@@ -1824,13 +1786,6 @@ class ServersControllerTestV23(ServersControllerTest):
availability_zone='nova',
launched_at=None,
terminated_at=None,
security_groups=[
{'name': 'fake-0-0', 'id': 1, 'description': 'foo',
'user_id': 'bar', 'project_id': 'baz', 'deleted': False,
'deleted_at': None, 'updated_at': None, 'created_at': None},
{'name': 'fake-0-1', 'id': 1, 'description': 'foo',
'user_id': 'bar', 'project_id': 'baz', 'deleted': False,
'deleted_at': None, 'updated_at': None, 'created_at': None}],
task_state=None,
vm_state=vm_states.ACTIVE,
power_state=1)
@@ -1891,17 +1846,6 @@ class ServersControllerTestV23(ServersControllerTest):
availability_zone='nova',
launched_at=None,
terminated_at=None,
security_groups=[
{'name': 'fake-0-0', 'id': 1,
'description': 'foo',
'user_id': 'bar', 'project_id': 'baz',
'deleted': False, 'deleted_at': None,
'updated_at': None, 'created_at': None},
{'name': 'fake-0-1', 'id': 1,
'description': 'foo',
'user_id': 'bar', 'project_id': 'baz',
'deleted': False, 'deleted_at': None,
'updated_at': None, 'created_at': None}],
task_state=None,
vm_state=vm_states.ACTIVE,
power_state=1)
@@ -1940,13 +1884,6 @@ class ServersControllerTestV29(ServersControllerTest):
availability_zone='nova',
launched_at=None,
terminated_at=None,
security_groups=[
{'name': 'fake-0-0', 'id': 1, 'description': 'foo',
'user_id': 'bar', 'project_id': 'baz', 'deleted': False,
'deleted_at': None, 'updated_at': None, 'created_at': None},
{'name': 'fake-0-1', 'id': 1, 'description': 'foo',
'user_id': 'bar', 'project_id': 'baz', 'deleted': False,
'deleted_at': None, 'updated_at': None, 'created_at': None}],
task_state=None,
vm_state=vm_states.ACTIVE,
power_state=1)
@@ -1992,13 +1929,6 @@ class ServersControllerTestV29(ServersControllerTest):
availability_zone='nova',
launched_at=None,
terminated_at=None,
security_groups=[
{'name': 'fake-0-0', 'id': 1, 'description': 'foo',
'user_id': 'bar', 'project_id': 'baz', 'deleted': False,
'deleted_at': None, 'updated_at': None, 'created_at': None},
{'name': 'fake-0-1', 'id': 1, 'description': 'foo',
'user_id': 'bar', 'project_id': 'baz', 'deleted': False,
'deleted_at': None, 'updated_at': None, 'created_at': None}],
task_state=None,
vm_state=vm_states.ACTIVE,
power_state=1)
@@ -2042,13 +1972,6 @@ class ServersControllerTestV29(ServersControllerTest):
availability_zone='nova',
launched_at=None,
terminated_at=None,
security_groups=[
{'name': 'fake-0-0', 'id': 1, 'description': 'foo',
'user_id': 'bar', 'project_id': 'baz', 'deleted': False,
'deleted_at': None, 'updated_at': None, 'created_at': None},
{'name': 'fake-0-1', 'id': 1, 'description': 'foo',
'user_id': 'bar', 'project_id': 'baz', 'deleted': False,
'deleted_at': None, 'updated_at': None, 'created_at': None}],
task_state=None,
vm_state=vm_states.ACTIVE,
power_state=1)
@@ -2111,13 +2034,6 @@ class ServersControllerTestV216(ServersControllerTest):
availability_zone='nova',
launched_at=None,
terminated_at=None,
security_groups=[
{'name': 'fake-0-0', 'id': 1, 'description': 'foo',
'user_id': 'bar', 'project_id': 'baz', 'deleted': False,
'deleted_at': None, 'updated_at': None, 'created_at': None},
{'name': 'fake-0-1', 'id': 1, 'description': 'foo',
'user_id': 'bar', 'project_id': 'baz', 'deleted': False,
'deleted_at': None, 'updated_at': None, 'created_at': None}],
task_state=None,
vm_state=vm_states.ACTIVE,
power_state=1)
@@ -2204,17 +2120,6 @@ class ServersControllerTestV216(ServersControllerTest):
availability_zone='nova',
launched_at=None,
terminated_at=None,
security_groups=[
{'name': 'fake-0-0', 'id': 1,
'description': 'foo',
'user_id': 'bar', 'project_id': 'baz',
'deleted': False, 'deleted_at': None,
'updated_at': None, 'created_at': None},
{'name': 'fake-0-1', 'id': 1,
'description': 'foo',
'user_id': 'bar', 'project_id': 'baz',
'deleted': False, 'deleted_at': None,
'updated_at': None, 'created_at': None}],
task_state=None,
vm_state=vm_states.ACTIVE,
power_state=1)
@@ -2260,13 +2165,6 @@ class ServersControllerTestV219(ServersControllerTest):
availability_zone='nova',
launched_at=None,
terminated_at=None,
security_groups=[
{'name': 'fake-0-0', 'id': 1, 'description': 'foo',
'user_id': 'bar', 'project_id': 'baz', 'deleted': False,
'deleted_at': None, 'updated_at': None, 'created_at': None},
{'name': 'fake-0-1', 'id': 1, 'description': 'foo',
'user_id': 'bar', 'project_id': 'baz', 'deleted': False,
'deleted_at': None, 'updated_at': None, 'created_at': None}],
task_state=None,
vm_state=vm_states.ACTIVE,
power_state=1)
@@ -2318,13 +2216,6 @@ class ServersControllerTestV219(ServersControllerTest):
availability_zone='nova',
launched_at=None,
terminated_at=None,
security_groups=[
{'name': 'fake-0-0', 'id': 1, 'description': 'foo',
'user_id': 'bar', 'project_id': 'baz', 'deleted': False,
'deleted_at': None, 'updated_at': None, 'created_at': None},
{'name': 'fake-0-1', 'id': 1, 'description': 'foo',
'user_id': 'bar', 'project_id': 'baz', 'deleted': False,
'deleted_at': None, 'updated_at': None, 'created_at': None}],
task_state=None,
vm_state=vm_states.ACTIVE,
power_state=1)
@@ -4082,8 +3973,9 @@ class ServerStatusTest(test.TestCase):
def setUp(self):
super(ServerStatusTest, self).setUp()
self.flags(use_neutron=False)
fakes.stub_out_nw_api(self)
fakes.stub_out_secgroup_api(
self, security_groups=[{'name': 'default'}])
self.controller = servers.ServersController()
@@ -4206,15 +4098,6 @@ class ServersControllerCreateTest(test.TestCase):
"task_state": "",
"vm_state": "",
"root_device_name": inst.get('root_device_name', 'vda'),
"security_groups": [
{'name': 'fake-0-0', 'id': 1, 'description': 'foo',
'user_id': 'bar', 'project_id': 'baz', 'deleted': False,
'deleted_at': None, 'updated_at': None,
'created_at': None},
{'name': 'fake-0-1', 'id': 1, 'description': 'foo',
'user_id': 'bar', 'project_id': 'baz', 'deleted': False,
'deleted_at': None, 'updated_at': None,
'created_at': None}]
})
self.instance_cache_by_id[instance['id']] = instance
@@ -4472,7 +4355,6 @@ class ServersControllerCreateTest(test.TestCase):
self._test_create_extra(params)
def test_create_instance_with_networks_disabled_neutronv2(self):
self.flags(use_neutron=True)
net_uuid = '76fa36fc-c930-4bf3-8c8a-ea2a2420deb6'
requested_networks = [{'uuid': net_uuid}]
params = {'networks': requested_networks}
@@ -6400,7 +6282,6 @@ class ServersControllerCreateTestV219(ServersControllerCreateTest):
class ServersControllerCreateTestV232(test.NoDBTestCase):
def setUp(self):
super(ServersControllerCreateTestV232, self).setUp()
self.flags(use_neutron=True)
self.controller = servers.ServersController()
@@ -6478,8 +6359,6 @@ class ServersControllerCreateTestV237(test.NoDBTestCase):
"""
def setUp(self):
super(ServersControllerCreateTestV237, self).setUp()
# Set the use_neutron flag to process requested networks.
self.flags(use_neutron=True)
# Create the server controller.
self.controller = servers.ServersController()
# Define a basic server create request body which tests can customize.
@@ -7099,7 +6978,6 @@ class ServersControllerCreateTestWithMock(test.TestCase):
@mock.patch.object(compute_api.API, 'create')
def test_create_instance_with_neutronv2_invalid_fixed_ip(self,
create_mock):
self.flags(use_neutron=True)
network = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
address = '999.0.2.3'
requested_networks = [{'uuid': network, 'fixed_ip': address}]
@@ -7122,8 +7000,6 @@ class ServersViewBuilderTest(test.TestCase):
def setUp(self):
super(ServersViewBuilderTest, self).setUp()
self.flags(use_ipv6=True)
# Neutron security groups are tested in test_neutron_security_groups.py
self.flags(use_neutron=False)
fakes.stub_out_nw_api(self)
self.flags(group='glance', api_servers=['http://localhost:9292'])
nw_cache_info = self._generate_nw_cache_info()
@@ -7137,16 +7013,9 @@ class ServersViewBuilderTest(test.TestCase):
nw_cache=nw_cache_info,
launched_at=None,
terminated_at=None,
security_groups=[
{'name': 'fake-0-0', 'id': 1, 'description': 'foo',
'user_id': 'bar', 'project_id': 'baz', 'deleted': False,
'deleted_at': None, 'updated_at': None, 'created_at': None},
{'name': 'fake-0-1', 'id': 1, 'description': 'foo',
'user_id': 'bar', 'project_id': 'baz', 'deleted': False,
'deleted_at': None, 'updated_at': None, 'created_at': None}],
task_state=None,
vm_state=vm_states.ACTIVE,
power_state=1)
task_state=None,
vm_state=vm_states.ACTIVE,
power_state=1)
privates = ['172.19.0.1']
publics = ['192.168.0.3']
@@ -7160,6 +7029,10 @@ class ServersViewBuilderTest(test.TestCase):
'ips': [dict(ip=ip) for ip in privates]})]
fakes.stub_out_nw_api_get_instance_nw_info(self, nw_info)
fakes.stub_out_secgroup_api(
self, security_groups=[{'name': 'default'}])
self.stub_out('nova.db.api.'
'block_device_mapping_get_all_by_instance_uuids',
fake_bdms_get_all_by_instance_uuids)
@@ -7351,8 +7224,7 @@ class ServersViewBuilderTest(test.TestCase):
"key_name": '',
"OS-SRV-USG:launched_at": None,
"OS-SRV-USG:terminated_at": None,
"security_groups": [{'name': 'fake-0-0'},
{'name': 'fake-0-1'}],
"security_groups": [{'name': 'default'}],
"OS-EXT-STS:task_state": None,
"OS-EXT-STS:vm_state": vm_states.ACTIVE,
"OS-EXT-STS:power_state": 1,
@@ -7447,8 +7319,7 @@ class ServersViewBuilderTest(test.TestCase):
"key_name": '',
"OS-SRV-USG:launched_at": None,
"OS-SRV-USG:terminated_at": None,
"security_groups": [{'name': 'fake-0-0'},
{'name': 'fake-0-1'}],
"security_groups": [{'name': 'default'}],
"OS-EXT-STS:task_state": None,
"OS-EXT-STS:vm_state": vm_states.ERROR,
"OS-EXT-STS:power_state": 1,
@@ -7642,8 +7513,7 @@ class ServersViewBuilderTest(test.TestCase):
"key_name": '',
"OS-SRV-USG:launched_at": None,
"OS-SRV-USG:terminated_at": None,
"security_groups": [{'name': 'fake-0-0'},
{'name': 'fake-0-1'}],
"security_groups": [{'name': 'default'}],
"OS-EXT-STS:task_state": None,
"OS-EXT-STS:vm_state": vm_states.ACTIVE,
"OS-EXT-STS:power_state": 1,
@@ -7735,8 +7605,7 @@ class ServersViewBuilderTest(test.TestCase):
"key_name": '',
"OS-SRV-USG:launched_at": None,
"OS-SRV-USG:terminated_at": None,
"security_groups": [{'name': 'fake-0-0'},
{'name': 'fake-0-1'}],
"security_groups": [{'name': 'default'}],
"OS-EXT-STS:task_state": None,
"OS-EXT-STS:vm_state": vm_states.ACTIVE,
"OS-EXT-STS:power_state": 1,
@@ -7873,8 +7742,7 @@ class ServersViewBuilderTestV269(ServersViewBuilderTest):
"description": None,
"OS-SRV-USG:launched_at": None,
"OS-SRV-USG:terminated_at": None,
"security_groups": [{'name': 'fake-0-0'},
{'name': 'fake-0-1'}],
"security_groups": [{'name': 'default'}],
"OS-EXT-STS:task_state": None,
"OS-EXT-STS:vm_state": vm_states.ACTIVE,
"OS-EXT-STS:power_state": 1,

4
nova/tests/unit/api/openstack/fakes.py
View File

@@ -227,8 +227,10 @@ def stub_out_secgroup_api(test, security_groups=None):
raise Exception('Invalid security group API call for nova-net')
instances_security_group_bindings = {}
if servers:
# we don't get security group information for down cells
instances_security_group_bindings = {
server['id']: [] for server in servers
server['id']: security_groups or [] for server in servers
if server['status'] != 'UNKNOWN'
}
return instances_security_group_bindings