openstack-attic/melange
Code Issues Proposed changes

Deepak / Vinkesh | Sending ip_block details along with ip_address while allocating ips for a network

Browse Source
This commit is contained in:
vinkesh banka
2011-08-05 11:37:30 +05:30
committed by Rajaram Mallya
parent e67c68ddd3
commit ea870d0fa9
4 changed files with 56 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
melange/ipam/models.py
View File

@@ -209,7 +209,7 @@ class ModelBase(object):
def to_dict(self):
return self.__dict__()
def data(self):
def data(self, **options):
data_fields = self._data_fields + self._auto_generated_attrs
return dict([(field, self[field])
for field in data_fields])
@@ -257,8 +257,8 @@ class IpAddressIterator(object):
class IpBlock(ModelBase):
_allowed_types = ["private", "public"]
_data_fields = ['cidr', 'network_id', 'policy_id', 'tenant_id',
'parent_id', 'type']
_data_fields = ['cidr', 'network_id', 'policy_id', 'tenant_id', 'gateway',
'parent_id', 'type', 'dns', 'broadcast', 'netmask']
@classmethod
def find_or_allocate_ip(cls, ip_block_id, address):
@@ -283,7 +283,6 @@ class IpBlock(ModelBase):
for block in db_api.find_all_blocks_with_deallocated_ips():
block.update(is_full=False)
block.delete_deallocated_ips()
@property
def broadcast(self):
return str(IPNetwork(self.cidr).broadcast)
@@ -292,6 +291,10 @@ class IpBlock(ModelBase):
def netmask(self):
return str(IPNetwork(self.cidr).netmask)
@property
def dns(self):
return Config.get('nameserver', None)
def is_ipv6(self):
return IPNetwork(self.cidr).version == 6
@@ -570,14 +573,11 @@ class IpAddress(ModelBase):
def locked(self):
return self.marked_for_deallocation
def data_with_network_info(self):
ip_block = self.ip_block()
network_info = {'broadcast': ip_block.broadcast,
'gateway': ip_block.gateway,
'netmask': ip_block.netmask,
'dns': Config.get('nameserver', None)}
return dict(self.data().items() + network_info.items())
def data(self, **options):
data = super(IpAddress, self).data(**options)
if options.get('with_ip_block', False):
data['ip_block'] = self.ip_block().data()
return data
class Policy(ModelBase):

2
melange/ipam/service.py
View File

@@ -284,7 +284,7 @@ class NetworksController(BaseController):
'addresses', 'mac_address')
ips = network.allocate_ips(addresses=addresses, port_id=port_id,
mac_address=mac_address)
return Result(dict(ip_addresses=[ip.data_with_network_info()
return Result(dict(ip_addresses=[ip.data(with_ip_block=True)
for ip in ips]), 201)

52
melange/tests/unit/test_ipam_models.py
View File

@@ -608,14 +608,15 @@ class TestIpBlock(BaseTest):
self.assertRaises(models.DuplicateAddressError, block.allocate_ip,
address=ip.address)
def test_ip_block_data(self):
def test_ipv4_block_data(self):
policy = PolicyFactory()
parent_block = PrivateIpBlockFactory(cidr="10.0.0.0/24")
ip_block = PrivateIpBlockFactory(cidr="10.0.0.0/29",
policy_id=policy.id,
parent_id=parent_block.id)
data = ip_block.data()
with StubConfig(nameserver="ns.example.com"):
data = ip_block.data()
self.assertEqual(data['id'], ip_block.id)
self.assertEqual(data['cidr'], ip_block.cidr)
@@ -625,6 +626,31 @@ class TestIpBlock(BaseTest):
self.assertEqual(data['parent_id'], ip_block.parent_id)
self.assertEqual(data['created_at'], ip_block.created_at)
self.assertEqual(data['updated_at'], ip_block.updated_at)
self.assertEqual(data['broadcast'], "10.0.0.7")
self.assertEqual(data['gateway'], "10.0.0.1")
self.assertEqual(data['netmask'], "255.255.255.248")
self.assertEqual(data['dns'], "ns.example.com")
def test_ipv6_block_data(self):
policy = PolicyFactory()
ip_block = IpV6IpBlockFactory(cidr="fe::/96",
policy_id=policy.id)
with StubConfig(nameserver="ns.example.com"):
data = ip_block.data()
self.assertEqual(data['id'], ip_block.id)
self.assertEqual(data['cidr'], ip_block.cidr)
self.assertEqual(data['network_id'], ip_block.network_id)
self.assertEqual(data['tenant_id'], ip_block.tenant_id)
self.assertEqual(data['policy_id'], ip_block.policy_id)
self.assertEqual(data['parent_id'], ip_block.parent_id)
self.assertEqual(data['created_at'], ip_block.created_at)
self.assertEqual(data['updated_at'], ip_block.updated_at)
self.assertEqual(data['broadcast'], "fe::ffff:ffff")
self.assertEqual(data['gateway'], "fe::1")
self.assertEqual(data['netmask'], "ffff:ffff:ffff:ffff:ffff:ffff::")
self.assertEqual(data['dns'], "ns.example.com")
def test_find_all_ip_blocks(self):
PrivateIpBlockFactory(cidr="10.2.0.0/28")
@@ -929,6 +955,15 @@ class TestIpAddress(BaseTest):
self.assertEqual(data['created_at'], ip.created_at)
self.assertEqual(data['updated_at'], ip.updated_at)
def test_ip_address_data_with_ip_block(self):
ip_block = PrivateIpBlockFactory(cidr="10.0.0.1/8")
ip = IpAddressFactory(ip_block_id=ip_block.id)
data = ip.data(with_ip_block=True)
self.assertEqual(data['id'], ip.id)
self.assertEqual(data['ip_block']['id'], ip_block.id)
def test_deallocate(self):
ip_block = PrivateIpBlockFactory(cidr="10.0.0.1/8")
ip_address = ip_block.allocate_ip()
@@ -974,19 +1009,6 @@ class TestIpAddress(BaseTest):
self.assertEqual([actual_ip], found_ips)
def test_data_with_network_info_should_have_extra_network_info(self):
block = IpBlockFactory(cidr="10.0.0.0/24", gateway="10.0.0.1")
ip_address = block.allocate_ip()
with StubConfig(nameserver="ns.example.com"):
data = ip_address.data_with_network_info()
self.assertEqual(data['broadcast'], "10.0.0.255")
self.assertEqual(data['gateway'], "10.0.0.1")
self.assertEqual(data['netmask'], "255.255.255.0")
self.assertEqual(data['dns'], "ns.example.com")
class TestPolicy(BaseTest):

14
melange/tests/unit/test_ipam_service.py
View File

@@ -1455,7 +1455,7 @@ class NetworksControllerBase(object):
ip_address = IpAddress.find_by(ip_block_id=ip_block.id)
self.assertEqual(response.status_int, 201)
self.assertEqual([_data(ip_address, self.data_method)],
self.assertEqual([_data(ip_address, with_ip_block=True)],
response.json['ip_addresses'])
self.assertEqual(ip_address.port_id, "123")
@@ -1467,7 +1467,7 @@ class NetworksControllerBase(object):
ip_address = IpAddress.find_by(ip_block_id=ip_block.id, port_id=123)
self.assertEqual(response.status_int, 201)
self.assertEqual([_data(ip_address, self.data_method)],
self.assertEqual([_data(ip_address, with_ip_block=True)],
response.json['ip_addresses'])
def test_allocate_ip_with_given_address(self):
@@ -1480,7 +1480,7 @@ class NetworksControllerBase(object):
ip_address = IpAddress.find_by(ip_block_id=ip_block.id,
address="10.0.0.1")
self.assertEqual(response.status_int, 201)
self.assertEqual([_data(ip_address, self.data_method)],
self.assertEqual([_data(ip_address, with_ip_block=True)],
response.json['ip_addresses'])
def test_allocate_ip_allocates_v6_address_with_given_mac_address(self):
@@ -1497,7 +1497,7 @@ class NetworksControllerBase(object):
{'network': {'mac_address': mac_address}})
ipv6_address = IpAddress.find_by(ip_block_id=ipv6_block.id)
self.assertEqual([_data(ipv6_address, self.data_method)],
self.assertEqual([_data(ipv6_address, with_ip_block=True)],
response.json['ip_addresses'])
@@ -1506,7 +1506,6 @@ class TestNetworksController(BaseTestController,
def setUp(self):
self.network_path = "/ipam"
self.data_method = lambda res: res.data_with_network_info()
super(TestNetworksController, self).setUp()
def _ip_block_factory(self, **kwargs):
@@ -1529,7 +1528,6 @@ class TestTenantNetworksController(NetworksControllerBase,
def setUp(self):
self.network_path = "/ipam/tenants/123"
self.data_method = lambda res: res.data_with_network_info()
super(TestTenantNetworksController, self).setUp()
def _ip_block_factory(self, **kwargs):
@@ -1557,7 +1555,7 @@ def _create_blocks(*args):
return [PrivateIpBlockFactory(cidr=cidr) for cidr in args]
def _data(resource, data_lambda=lambda res: res.data()):
def _data(resource, **options):
if isinstance(resource, models.ModelBase):
return sanitize(data_lambda(resource))
return sanitize(resource.data(**options))
return [_data(model) for model in resource]