|
55 | 55 | from neutron.db import provisioning_blocks
|
56 | 56 | from neutron.db import securitygroups_db as sg_db
|
57 | 57 | from neutron.db import segments_db
|
| 58 | +from neutron.ipam import driver |
58 | 59 | from neutron.objects import base as base_obj
|
59 | 60 | from neutron.objects import ports as port_obj
|
60 | 61 | from neutron.objects import router as l3_obj
|
@@ -1737,6 +1738,39 @@ def test_create_ports_bulk_with_extra_dhcp_opts(self):
|
1737 | 1738 | ports_out = self.plugin.create_port_bulk(ctx, ports_in)
|
1738 | 1739 | self.assertEqual(edo, ports_out[0]['extra_dhcp_opts'])
|
1739 | 1740 |
|
| 1741 | + def test_create_ports_bulk_with_wrong_fixed_ips(self): |
| 1742 | + cidr = '10.0.10.0/24' |
| 1743 | + with self.network() as net: |
| 1744 | + with self.subnet(net, cidr=cidr) as snet: |
| 1745 | + net_id = net['network']['id'] |
| 1746 | + data = [{'network_id': net_id, |
| 1747 | + 'fixed_ips': [{'subnet_id': snet['subnet']['id'], |
| 1748 | + 'ip_address': '10.0.10.100'}], |
| 1749 | + 'tenant_id': snet['subnet']['tenant_id'] |
| 1750 | + }, |
| 1751 | + {'network_id': net_id, |
| 1752 | + 'fixed_ips': [{'subnet_id': snet['subnet']['id'], |
| 1753 | + 'ip_address': '10.0.20.101'}], |
| 1754 | + 'tenant_id': snet['subnet']['tenant_id'] |
| 1755 | + }] |
| 1756 | + res = self._create_bulk_from_list(self.fmt, 'port', |
| 1757 | + data, as_admin=True) |
| 1758 | + self.assertEqual(webob.exc.HTTPBadRequest.code, res.status_int) |
| 1759 | + self.assertIn('IP address 10.0.20.101 is not a valid IP for ' |
| 1760 | + 'the specified subnet.', |
| 1761 | + res.json['NeutronError']['message']) |
| 1762 | + |
| 1763 | + ipam_driver = driver.Pool.get_instance(None, self.context) |
| 1764 | + ipam_allocator = ipam_driver.get_allocator([cidr]) |
| 1765 | + with db_api.CONTEXT_READER.using(self.context): |
| 1766 | + ipam_subnet = ipam_allocator._driver.get_subnet( |
| 1767 | + snet['subnet']['id']) |
| 1768 | + allocations = ipam_subnet.subnet_manager.list_allocations( |
| 1769 | + self.context) |
| 1770 | + # There are no leftovers (e.g.: 10.0.10.100) in the |
| 1771 | + # "IpamAllocation" registers |
| 1772 | + self.assertEqual([], allocations) |
| 1773 | + |
1740 | 1774 | def test_delete_port_no_notify_in_disassociate_floatingips(self):
|
1741 | 1775 | ctx = context.get_admin_context()
|
1742 | 1776 | plugin = directory.get_plugin()
|
|
0 commit comments