Merge "Add FloatingIP object implementation"

openstack/nova

Code Issues Proposed changes

Browse Source

This commit is contained in:

Jenkins

2014年02月02日 00:11:02 +00:00

committed by Gerrit Code Review

parent f6cad6f33f e1c33e2ec7

commit 77ebcc052d

3 changed files with 350 additions and 0 deletions

Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL

Download Patch File Download Diff File Expand all files Collapse all files

1
nova/objects/init.py

View File

@@ -27,3 +27,4 @@ def register_all():

__import__('nova.objects.network')

__import__('nova.objects.block_device')

__import__('nova.objects.fixed_ip')

__import__('nova.objects.floating_ip')

158
nova/objects/floating_ip.py Normal file

View File

@@ -0,0 +1,158 @@

#

# 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 nova import db

from nova import exception

from nova.objects import base as obj_base

from nova.objects import fields

from nova.objects import fixed_ip

FLOATING_IP_OPTIONAL_ATTRS = ['fixed_ip']

class FloatingIP(obj_base.NovaPersistentObject, obj_base.NovaObject):

fields = {

'id': fields.IntegerField(),

'address': fields.IPAddressField(),

'fixed_ip_id': fields.IntegerField(nullable=True),

'project_id': fields.UUIDField(nullable=True),

'host': fields.StringField(nullable=True),

'auto_assigned': fields.BooleanField(),

'pool': fields.StringField(nullable=True),

'interface': fields.StringField(nullable=True),

'fixed_ip': fields.ObjectField('FixedIP', nullable=True),

}

@staticmethod

def _from_db_object(context, floatingip, db_floatingip,

expected_attrs=None):

if expected_attrs is None:

expected_attrs = []

for field in floatingip.fields:

if field not in FLOATING_IP_OPTIONAL_ATTRS:

floatingip[field] = db_floatingip[field]

if 'fixed_ip' in expected_attrs:

floatingip.fixed_ip = fixed_ip.FixedIP._from_db_object(

context, fixed_ip.FixedIP(), db_floatingip['fixed_ip'])

floatingip._context = context

floatingip.obj_reset_changes()

return floatingip

@obj_base.remotable_classmethod

def get_by_id(cls, context, id):

db_floatingip = db.floating_ip_get(context, id)

# XXX joins fixed.instance

return cls._from_db_object(context, cls(), db_floatingip, ['fixed_ip'])

@obj_base.remotable_classmethod

def get_by_address(cls, context, address):

db_floatingip = db.floating_ip_get_by_address(context, address)

return cls._from_db_object(context, cls(), db_floatingip)

@obj_base.remotable_classmethod

def get_pool_names(cls, context):

return [x['name'] for x in db.floating_ip_get_pools(context)]

@obj_base.remotable_classmethod

def allocate_address(cls, context, project_id, pool, auto_assigned=False):

return db.floating_ip_allocate_address(context, project_id, pool,

auto_assigned=auto_assigned)

@obj_base.remotable_classmethod

def associate(cls, context, floating_address, fixed_address, host):

db_fixed = db.floating_ip_fixed_ip_associate(context,

floating_address,

fixed_address,

host)

if db_fixed is None:

return None

floating = FloatingIP(

context=context, address=floating_address, host=host,

fixed_ip_id=db_fixed['id'],

fixed_ip=fixed_ip.FixedIP._from_db_object(

context, fixed_ip.FixedIP(), db_fixed,

expected_attrs=['network']))

return floating

@obj_base.remotable_classmethod

def deallocate(cls, context, address):

db.floating_ip_deallocate(context, address)

@obj_base.remotable_classmethod

def destroy(cls, context, address):

db.floating_ip_destroy(context, address)

@obj_base.remotable_classmethod

def disassociate(cls, context, address):

db_fixed = db.floating_ip_disassociate(context, address)

floating = FloatingIP(

context=context, address=address,

fixed_ip_id=db_fixed['id'],

fixed_ip=fixed_ip.FixedIP._from_db_object(

context, fixed_ip.FixedIP(), db_fixed,

expected_attrs=['network']))

return floating

@obj_base.remotable

def save(self, context):

updates = self.obj_get_changes()

if 'address' in updates:

raise exception.ObjectActionError(action='save',

reason='address is not mutable')

db_floatingip = db.floating_ip_update(context, str(self.address),

updates)

self._from_db_object(context, self, db_floatingip)

class FloatingIPList(obj_base.ObjectListBase, obj_base.NovaObject):

fields = {

'objects': fields.ListOfObjectsField('FloatingIP'),

}

child_versions = {

'1.0': '1.0',

}

@obj_base.remotable_classmethod

def get_all(cls, context):

db_floatingips = db.floating_ip_get_all(context)

return obj_base.obj_make_list(context, cls(), FloatingIP,

db_floatingips)

@obj_base.remotable_classmethod

def get_by_host(cls, context, host):

db_floatingips = db.floating_ip_get_all_by_host(context, host)

return obj_base.obj_make_list(context, cls(), FloatingIP,

db_floatingips)

@obj_base.remotable_classmethod

def get_by_project(cls, context, project_id):

db_floatingips = db.floating_ip_get_all_by_project(context, project_id)

return obj_base.obj_make_list(context, cls(), FloatingIP,

db_floatingips)

@obj_base.remotable_classmethod

def get_by_fixed_address(cls, context, fixed_address):

db_floatingips = db.floating_ip_get_by_fixed_address(context,

fixed_address)

return obj_base.obj_make_list(context, cls(), FloatingIP,

db_floatingips)

@obj_base.remotable_classmethod

def get_by_fixed_ip_id(cls, context, fixed_ip_id):

db_floatingips = db.floating_ip_get_by_fixed_ip_id(context,

fixed_ip_id)

return obj_base.obj_make_list(context, cls(), FloatingIP,

db_floatingips)

191
nova/tests/objects/test_floating_ip.py Normal file

View File

@@ -0,0 +1,191 @@

#

# 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 nova import exception

from nova.objects import floating_ip

from nova.tests.objects import test_fixed_ip

from nova.tests.objects import test_network

from nova.tests.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.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.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.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.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.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.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.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.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.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')

self.assertRaises(exception.ObjectActionError, floatingip.save)

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'})

@mock.patch('nova.db.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.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.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.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.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)

class TestFloatingIPObject(test_objects._LocalTest,

_TestFloatingIPObject):

pass

class TestRemoteFloatingIPObject(test_objects._RemoteTest,

_TestFloatingIPObject):

pass