290 lines
12 KiB
Python
290 lines
12 KiB
Python
# Copyright 2014 Red Hat, Inc.
|
|
#
|
|
# 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.
|
|
|
|
import mock
|
|
import netaddr
|
|
from oslo_versionedobjects import base as ovo_base
|
|
|
|
from nova import exception
|
|
from nova import objects
|
|
from nova.objects import floating_ip
|
|
from nova import test
|
|
from nova.tests.unit.api.openstack import fakes
|
|
from nova.tests.unit.objects import test_fixed_ip
|
|
from nova.tests.unit.objects import test_network
|
|
from nova.tests.unit.objects import test_objects
|
|
|
|
fake_floating_ip = {
|
|
'created_at': None,
|
|
'updated_at': None,
|
|
'deleted_at': None,
|
|
'deleted': False,
|
|
'id': 123,
|
|
'address': '172.17.0.1',
|
|
'fixed_ip_id': None,
|
|
'project_id': None,
|
|
'host': None,
|
|
'auto_assigned': False,
|
|
'pool': None,
|
|
'interface': None,
|
|
'fixed_ip': None,
|
|
}
|
|
|
|
|
|
class _TestFloatingIPObject(object):
|
|
def _compare(self, obj, db_obj):
|
|
for field in obj.fields:
|
|
if field in floating_ip.FLOATING_IP_OPTIONAL_ATTRS:
|
|
if obj.obj_attr_is_set(field):
|
|
obj_val = obj[field].id
|
|
db_val = db_obj[field]['id']
|
|
else:
|
|
continue
|
|
else:
|
|
obj_val = obj[field]
|
|
db_val = db_obj[field]
|
|
if isinstance(obj_val, netaddr.IPAddress):
|
|
obj_val = str(obj_val)
|
|
self.assertEqual(db_val, obj_val)
|
|
|
|
@mock.patch('nova.db.api.floating_ip_get')
|
|
def test_get_by_id(self, get):
|
|
db_floatingip = dict(fake_floating_ip,
|
|
fixed_ip=test_fixed_ip.fake_fixed_ip)
|
|
get.return_value = db_floatingip
|
|
floatingip = floating_ip.FloatingIP.get_by_id(self.context, 123)
|
|
get.assert_called_once_with(self.context, 123)
|
|
self._compare(floatingip, db_floatingip)
|
|
|
|
@mock.patch('nova.db.api.floating_ip_get_by_address')
|
|
def test_get_by_address(self, get):
|
|
get.return_value = fake_floating_ip
|
|
floatingip = floating_ip.FloatingIP.get_by_address(self.context,
|
|
'1.2.3.4')
|
|
get.assert_called_once_with(self.context, '1.2.3.4')
|
|
self._compare(floatingip, fake_floating_ip)
|
|
|
|
@mock.patch('nova.db.api.floating_ip_get_pools')
|
|
def test_get_pool_names(self, get):
|
|
get.return_value = [{'name': 'a'}, {'name': 'b'}]
|
|
self.assertEqual(['a', 'b'],
|
|
floating_ip.FloatingIP.get_pool_names(self.context))
|
|
|
|
@mock.patch('nova.db.api.floating_ip_allocate_address')
|
|
def test_allocate_address(self, allocate):
|
|
allocate.return_value = '1.2.3.4'
|
|
self.assertEqual('1.2.3.4',
|
|
floating_ip.FloatingIP.allocate_address(self.context,
|
|
'project',
|
|
'pool'))
|
|
allocate.assert_called_with(self.context, 'project', 'pool',
|
|
auto_assigned=False)
|
|
|
|
@mock.patch('nova.db.api.floating_ip_fixed_ip_associate')
|
|
def test_associate(self, associate):
|
|
db_fixed = dict(test_fixed_ip.fake_fixed_ip,
|
|
network=test_network.fake_network)
|
|
associate.return_value = db_fixed
|
|
floatingip = floating_ip.FloatingIP.associate(self.context,
|
|
'172.17.0.1',
|
|
'192.168.1.1',
|
|
'host')
|
|
associate.assert_called_with(self.context, '172.17.0.1',
|
|
'192.168.1.1', 'host')
|
|
self.assertEqual(db_fixed['id'], floatingip.fixed_ip.id)
|
|
self.assertEqual('172.17.0.1', str(floatingip.address))
|
|
self.assertEqual('host', floatingip.host)
|
|
|
|
@mock.patch('nova.db.api.floating_ip_deallocate')
|
|
def test_deallocate(self, deallocate):
|
|
floating_ip.FloatingIP.deallocate(self.context, '1.2.3.4')
|
|
deallocate.assert_called_with(self.context, '1.2.3.4')
|
|
|
|
@mock.patch('nova.db.api.floating_ip_destroy')
|
|
def test_destroy(self, destroy):
|
|
floating_ip.FloatingIP.destroy(self.context, '1.2.3.4')
|
|
destroy.assert_called_with(self.context, '1.2.3.4')
|
|
|
|
@mock.patch('nova.db.api.floating_ip_disassociate')
|
|
def test_disassociate(self, disassociate):
|
|
db_fixed = dict(test_fixed_ip.fake_fixed_ip,
|
|
network=test_network.fake_network)
|
|
disassociate.return_value = db_fixed
|
|
floatingip = floating_ip.FloatingIP.disassociate(self.context,
|
|
'1.2.3.4')
|
|
disassociate.assert_called_with(self.context, '1.2.3.4')
|
|
self.assertEqual(db_fixed['id'], floatingip.fixed_ip.id)
|
|
self.assertEqual('1.2.3.4', str(floatingip.address))
|
|
|
|
@mock.patch('nova.db.api.floating_ip_update')
|
|
def test_save(self, update):
|
|
update.return_value = fake_floating_ip
|
|
floatingip = floating_ip.FloatingIP(context=self.context,
|
|
id=123, address='1.2.3.4',
|
|
host='foo')
|
|
floatingip.obj_reset_changes(['address', 'id'])
|
|
floatingip.save()
|
|
self.assertEqual(set(), floatingip.obj_what_changed())
|
|
update.assert_called_with(self.context, '1.2.3.4',
|
|
{'host': 'foo'})
|
|
|
|
def test_save_errors(self):
|
|
floatingip = floating_ip.FloatingIP(context=self.context,
|
|
id=123, host='foo')
|
|
floatingip.obj_reset_changes()
|
|
floating_ip.address = '1.2.3.4'
|
|
self.assertRaises(exception.ObjectActionError, floatingip.save)
|
|
|
|
floatingip.obj_reset_changes()
|
|
floatingip.fixed_ip_id = 1
|
|
self.assertRaises(exception.ObjectActionError, floatingip.save)
|
|
|
|
@mock.patch('nova.db.api.floating_ip_update')
|
|
def test_save_no_fixedip(self, update):
|
|
update.return_value = fake_floating_ip
|
|
floatingip = floating_ip.FloatingIP(context=self.context,
|
|
id=123)
|
|
floatingip.fixed_ip = objects.FixedIP(context=self.context,
|
|
id=456)
|
|
self.assertNotIn('fixed_ip', update.calls[1])
|
|
|
|
@mock.patch('nova.db.api.floating_ip_get_all')
|
|
def test_get_all(self, get):
|
|
get.return_value = [fake_floating_ip]
|
|
floatingips = floating_ip.FloatingIPList.get_all(self.context)
|
|
self.assertEqual(1, len(floatingips))
|
|
self._compare(floatingips[0], fake_floating_ip)
|
|
get.assert_called_with(self.context)
|
|
|
|
@mock.patch('nova.db.api.floating_ip_get_all_by_host')
|
|
def test_get_by_host(self, get):
|
|
get.return_value = [fake_floating_ip]
|
|
floatingips = floating_ip.FloatingIPList.get_by_host(self.context,
|
|
'host')
|
|
self.assertEqual(1, len(floatingips))
|
|
self._compare(floatingips[0], fake_floating_ip)
|
|
get.assert_called_with(self.context, 'host')
|
|
|
|
@mock.patch('nova.db.api.floating_ip_get_all_by_project')
|
|
def test_get_by_project(self, get):
|
|
get.return_value = [fake_floating_ip]
|
|
floatingips = floating_ip.FloatingIPList.get_by_project(self.context,
|
|
'project')
|
|
self.assertEqual(1, len(floatingips))
|
|
self._compare(floatingips[0], fake_floating_ip)
|
|
get.assert_called_with(self.context, 'project')
|
|
|
|
@mock.patch('nova.db.api.floating_ip_get_by_fixed_address')
|
|
def test_get_by_fixed_address(self, get):
|
|
get.return_value = [fake_floating_ip]
|
|
floatingips = floating_ip.FloatingIPList.get_by_fixed_address(
|
|
self.context, '1.2.3.4')
|
|
self.assertEqual(1, len(floatingips))
|
|
self._compare(floatingips[0], fake_floating_ip)
|
|
get.assert_called_with(self.context, '1.2.3.4')
|
|
|
|
@mock.patch('nova.db.api.floating_ip_get_by_fixed_ip_id')
|
|
def test_get_by_fixed_ip_id(self, get):
|
|
get.return_value = [fake_floating_ip]
|
|
floatingips = floating_ip.FloatingIPList.get_by_fixed_ip_id(
|
|
self.context, 123)
|
|
self.assertEqual(1, len(floatingips))
|
|
self._compare(floatingips[0], fake_floating_ip)
|
|
get.assert_called_with(self.context, 123)
|
|
|
|
@mock.patch('nova.db.api.instance_floating_address_get_all')
|
|
def test_get_addresses_by_instance(self, get_all):
|
|
expected = ['1.2.3.4', '4.5.6.7']
|
|
get_all.return_value = list(expected)
|
|
ips = floating_ip.FloatingIP.get_addresses_by_instance(
|
|
self.context, {'uuid': '1234'})
|
|
self.assertEqual(expected, ips)
|
|
get_all.assert_called_once_with(self.context, '1234')
|
|
|
|
def test_make_ip_info(self):
|
|
result = objects.FloatingIPList.make_ip_info('1.2.3.4', 'pool', 'eth0')
|
|
self.assertEqual({'address': '1.2.3.4', 'pool': 'pool',
|
|
'interface': 'eth0'},
|
|
result)
|
|
|
|
@mock.patch('nova.db.api.floating_ip_bulk_create')
|
|
def test_bulk_create(self, create_mock):
|
|
def fake_create(ctxt, ip_info, want_result=False):
|
|
return [{'id': 1, 'address': ip['address'], 'fixed_ip_id': 1,
|
|
'project_id': fakes.FAKE_PROJECT_ID, 'host': 'host',
|
|
'auto_assigned': False, 'pool': ip['pool'],
|
|
'interface': ip['interface'], 'fixed_ip': None,
|
|
'created_at': None, 'updated_at': None,
|
|
'deleted_at': None, 'deleted': False}
|
|
for ip in ip_info]
|
|
|
|
create_mock.side_effect = fake_create
|
|
ips = [objects.FloatingIPList.make_ip_info('1.1.1.1', 'pool', 'eth0'),
|
|
objects.FloatingIPList.make_ip_info('1.1.1.2', 'loop', 'eth1')]
|
|
result = objects.FloatingIPList.create(None, ips)
|
|
self.assertIsNone(result)
|
|
result = objects.FloatingIPList.create(None, ips, want_result=True)
|
|
self.assertEqual('1.1.1.1', str(result[0].address))
|
|
self.assertEqual('1.1.1.2', str(result[1].address))
|
|
|
|
@mock.patch('nova.db.api.floating_ip_bulk_destroy')
|
|
def test_bulk_destroy(self, destroy_mock):
|
|
ips = [{'address': '1.2.3.4'}, {'address': '4.5.6.7'}]
|
|
objects.FloatingIPList.destroy(None, ips)
|
|
destroy_mock.assert_called_once_with(None, ips)
|
|
|
|
def test_backport_fixedip_1_1(self):
|
|
floating = objects.FloatingIP()
|
|
fixed = objects.FixedIP()
|
|
floating.fixed_ip = fixed
|
|
versions = ovo_base.obj_tree_get_versions('FloatingIP')
|
|
versions['FixedIP'] = '1.1'
|
|
primitive = floating.obj_to_primitive(target_version='1.1',
|
|
version_manifest=versions)
|
|
self.assertEqual('1.1',
|
|
primitive['nova_object.data']['fixed_ip']['nova_object.version'])
|
|
|
|
def test_get_count_by_project(self):
|
|
ips = [objects.FloatingIPList.make_ip_info('1.1.1.1', 'pool', 'eth0')]
|
|
objects.FloatingIPList.create(self.context, ips)
|
|
floating_ip.FloatingIP.allocate_address(self.context,
|
|
self.context.project_id,
|
|
'pool')
|
|
self.assertEqual(1, floating_ip.FloatingIPList.get_count_by_project(
|
|
self.context, self.context.project_id))
|
|
|
|
|
|
class TestFloatingIPObject(test_objects._LocalTest,
|
|
_TestFloatingIPObject):
|
|
pass
|
|
|
|
|
|
class TestRemoteFloatingIPObject(test_objects._RemoteTest,
|
|
_TestFloatingIPObject):
|
|
pass
|
|
|
|
|
|
class TestNeutronFloatingIPObject(test.NoDBTestCase):
|
|
def test_create_with_uuid_id(self):
|
|
uuid = 'fc9b4956-fd97-11e5-86aa-5e5517507c66'
|
|
fip = objects.floating_ip.NeutronFloatingIP(id=uuid)
|
|
|
|
self.assertEqual(uuid, fip.id)
|
|
|
|
def test_create_with_uuid_fixed_id(self):
|
|
uuid = 'fc9b4c3a-fd97-11e5-86aa-5e5517507c66'
|
|
fip = objects.floating_ip.NeutronFloatingIP(fixed_ip_id=uuid)
|
|
|
|
self.assertEqual(uuid, fip.fixed_ip_id)
|