diff --git a/neutron/common/utils.py b/neutron/common/utils.py index 48fcd8d6754..71dfbab31dc 100644 --- a/neutron/common/utils.py +++ b/neutron/common/utils.py @@ -204,6 +204,15 @@ def is_dvr_serviced(device_owner): device_owner in get_other_dvr_serviced_device_owners()) +def is_fip_serviced(device_owner): + """Check if the port can be assigned a floating IP + + Helper function to check the device owner of a + port can be assigned a floating IP. + """ + return device_owner != n_const.DEVICE_OWNER_DHCP + + def ip_to_cidr(ip, prefix=None): """Convert an ip with no prefix to cidr notation diff --git a/neutron/db/l3_db.py b/neutron/db/l3_db.py index 6957dfbaf87..3d8b38451d4 100644 --- a/neutron/db/l3_db.py +++ b/neutron/db/l3_db.py @@ -1143,6 +1143,9 @@ class L3_NAT_dbonly_mixin(l3.RouterPluginBase, raise n_exc.BadRequest(resource='floatingip', msg=msg) internal_subnet_id = None + if not utils.is_fip_serviced(internal_port.get('device_owner')): + msg = _('Port %(id)s is unable to be assigned a floating IP') + raise n_exc.BadRequest(resource='floatingip', msg=msg) if fip.get('fixed_ip_address'): internal_ip_address = fip['fixed_ip_address'] if netaddr.IPAddress(internal_ip_address).version != 4: diff --git a/neutron/tests/unit/common/test_utils.py b/neutron/tests/unit/common/test_utils.py index cb93ac277e0..f5616e737f6 100644 --- a/neutron/tests/unit/common/test_utils.py +++ b/neutron/tests/unit/common/test_utils.py @@ -494,6 +494,24 @@ class TestDvrServices(base.BaseTestCase): self._test_is_dvr_serviced(constants.DEVICE_OWNER_COMPUTE_PREFIX, True) +class TestFipServices(base.BaseTestCase): + + def _test_is_fip_serviced(self, device_owner, expected): + self.assertEqual(expected, utils.is_fip_serviced(device_owner)) + + def test_is_fip_serviced_with_lb_port(self): + self._test_is_fip_serviced(constants.DEVICE_OWNER_LOADBALANCER, True) + + def test_is_fip_serviced_with_lbv2_port(self): + self._test_is_fip_serviced(constants.DEVICE_OWNER_LOADBALANCERV2, True) + + def test_is_fip_serviced_with_dhcp_port(self): + self._test_is_fip_serviced(constants.DEVICE_OWNER_DHCP, False) + + def test_is_fip_serviced_with_vm_port(self): + self._test_is_fip_serviced(constants.DEVICE_OWNER_COMPUTE_PREFIX, True) + + class TestIpToCidr(base.BaseTestCase): def test_ip_to_cidr_ipv4_default(self): self.assertEqual('15.1.2.3/32', utils.ip_to_cidr('15.1.2.3')) diff --git a/neutron/tests/unit/extensions/test_l3.py b/neutron/tests/unit/extensions/test_l3.py index e2ad178132b..a055c4cce75 100644 --- a/neutron/tests/unit/extensions/test_l3.py +++ b/neutron/tests/unit/extensions/test_l3.py @@ -3456,6 +3456,16 @@ class L3NatTestCaseBase(L3NatTestCaseMixin): self._delete('routers', router['router']['id'], exc.HTTPForbidden.code) + def test_associate_to_dhcp_port_fails(self): + with self.subnet(cidr="10.0.0.0/24", ip_version=4) as sub: + with self.port(subnet=sub, + device_owner=lib_constants.DEVICE_OWNER_DHCP) as p: + res = self._create_floatingip( + self.fmt, + sub['subnet']['network_id'], + port_id=p['port']['id']) + self.assertEqual(exc.HTTPBadRequest.code, res.status_int) + class L3AgentDbTestCaseBase(L3NatTestCaseMixin):