subnet_id=subnet_id).delete()
@staticmethod
- def _generate_ip(context, network_id, subnets):
+ def _generate_ip(context, subnets):
"""Generate an IP address.
The IP address will be generated from one of the subnets defined on
# increment the first free
range['first_ip'] = str(netaddr.IPAddress(ip_address) + 1)
return {'ip_address': ip_address, 'subnet_id': subnet['id']}
- raise q_exc.IpAddressGenerationFailure(net_id=network_id)
+ raise q_exc.IpAddressGenerationFailure(net_id=subnets[0]['network_id'])
@staticmethod
def _allocate_specific_ip(context, subnet_id, ip_address):
else:
subnets = [self._get_subnet(context, fixed['subnet_id'])]
# IP address allocation
- result = self._generate_ip(context, network, subnets)
+ result = self._generate_ip(context, subnets)
ips.append({'ip_address': result['ip_address'],
'subnet_id': result['subnet_id']})
return ips
version_subnets = [v4, v6]
for subnets in version_subnets:
if subnets:
- result = QuantumDbPluginV2._generate_ip(context, network,
- subnets)
+ result = QuantumDbPluginV2._generate_ip(context, subnets)
ips.append({'ip_address': result['ip_address'],
'subnet_id': result['subnet_id']})
return ips
self.assertEqual(res['port']['fixed_ips'],
data['port']['fixed_ips'])
+ def test_no_more_port_exception(self):
+ fmt = 'json'
+ with self.subnet(cidr='10.0.0.0/32') as subnet:
+ id = subnet['subnet']['network_id']
+ res = self._create_port(fmt, id)
+ data = self.deserialize(fmt, res)
+ msg = str(q_exc.IpAddressGenerationFailure(net_id=id))
+ self.assertEqual(data['QuantumError'], msg)
+ self.assertEqual(res.status_int, 409)
+
def test_update_port_update_ip(self):
"""Test update of port IP.