diff --git a/neutron_fwaas/services/firewall/fwaas_plugin_v2.py b/neutron_fwaas/services/firewall/fwaas_plugin_v2.py index da1ef6db7..1477f212e 100644 --- a/neutron_fwaas/services/firewall/fwaas_plugin_v2.py +++ b/neutron_fwaas/services/firewall/fwaas_plugin_v2.py @@ -164,7 +164,7 @@ class FirewallPluginV2(Firewallv2PluginBase): raise f_exc.FirewallGroupPortInvalidProject( port_id=port_id, project_id=port['tenant_id']) device_owner = port.get('device_owner', '') - if (device_owner not in [nl_constants.DEVICE_OWNER_ROUTER_INTF] and + if (device_owner not in nl_constants.ROUTER_INTERFACE_OWNERS and not device_owner.startswith( nl_constants.DEVICE_OWNER_COMPUTE_PREFIX)): raise f_exc.FirewallGroupPortInvalid(port_id=port_id) diff --git a/neutron_fwaas/tests/unit/db/firewall/v2/test_firewall_db_v2.py b/neutron_fwaas/tests/unit/db/firewall/v2/test_firewall_db_v2.py index 913aca6e3..a0b40ea7f 100644 --- a/neutron_fwaas/tests/unit/db/firewall/v2/test_firewall_db_v2.py +++ b/neutron_fwaas/tests/unit/db/firewall/v2/test_firewall_db_v2.py @@ -760,11 +760,25 @@ class TestFirewallDBPluginV2(test_fwaas_plugin_v2.FirewallPluginV2TestCase): attrs = self._get_test_firewall_group_attrs("firewall1") self._test_create_firewall_group(attrs) - def test_create_firewall_group_with_ports(self): + def test_create_firewall_group_with_router_port(self): with self.port( - device_owner=nl_constants.DEVICE_OWNER_ROUTER_INTF) as dummy_port: + device_owner=nl_constants.DEVICE_OWNER_ROUTER_INTF) as port: attrs = self._get_test_firewall_group_attrs("fwg1") - attrs['ports'] = [dummy_port['port']['id']] + attrs['ports'] = [port['port']['id']] + self._test_create_firewall_group(attrs) + + def test_create_firewall_group_with_dvr_port(self): + with self.port( + device_owner=nl_constants.DEVICE_OWNER_DVR_INTERFACE) as port: + attrs = self._get_test_firewall_group_attrs("fwg1") + attrs['ports'] = [port['port']['id']] + self._test_create_firewall_group(attrs) + + def test_create_firewall_group_with_router_port_l3ha(self): + with self.port( + device_owner=nl_constants.DEVICE_OWNER_HA_REPLICATED_INT) as port: + attrs = self._get_test_firewall_group_attrs("fwg1") + attrs['ports'] = [port['port']['id']] self._test_create_firewall_group(attrs) def test_create_firewall_group_with_empty_ports(self): diff --git a/neutron_fwaas/tests/unit/services/firewall/test_fwaas_plugin_v2.py b/neutron_fwaas/tests/unit/services/firewall/test_fwaas_plugin_v2.py index 2bde6541d..6e759bd8d 100644 --- a/neutron_fwaas/tests/unit/services/firewall/test_fwaas_plugin_v2.py +++ b/neutron_fwaas/tests/unit/services/firewall/test_fwaas_plugin_v2.py @@ -418,6 +418,29 @@ class FirewallPluginV2TestCase(base.NeutronDbPluginV2TestCase): class TestFirewallPluginBasev2(FirewallPluginV2TestCase): + + def _test_fwg_with_port(self, device_owner): + with self.port(device_owner=device_owner) as port: + with self.firewall_rule() as fwr: + fwr_id = fwr['firewall_rule']['id'] + with self.firewall_policy(firewall_rules=[fwr_id]) as fwp: + fwp_id = fwp['firewall_policy']['id'] + self.firewall_group( + self.fmt, + "firewall_group", + self.DESCRIPTION, + ports=[port['port']['id']], + ingress_firewall_policy_id=fwp_id, + ) + + def test_create_fwg_with_l3_ports(self): + for device_owner_for_l3 in nl_constants.ROUTER_INTERFACE_OWNERS: + self._test_fwg_with_port(device_owner_for_l3) + + def test_create_fwg_with_l2_port(self): + device_owner_for_l2 = nl_constants.DEVICE_OWNER_COMPUTE_PREFIX + 'nova' + self._test_fwg_with_port(device_owner_for_l2) + def test_create_firewall_group_with_port_on_different_project(self): with self.port(tenant_id='fake_project_id_1') as port: admin_ctx = context.get_admin_context() @@ -473,7 +496,7 @@ class TestFirewallPluginBasev2(FirewallPluginV2TestCase): res = req.get_response(self.ext_api) self.assertEqual(webob.exc.HTTPConflict.code, res.status_int) - def test_create_firewall_group_with_port_already_in_use(self): + def test_create_firewall_group_with_router_port_already_in_use(self): with self.port( device_owner=nl_constants.DEVICE_OWNER_ROUTER_INTF) as port: with self.firewall_group(ports=[port['port']['id']]): @@ -485,6 +508,18 @@ class TestFirewallPluginBasev2(FirewallPluginV2TestCase): expected_res_status=webob.exc.HTTPConflict.code, ) + def test_create_firewall_group_with_dvr_port_already_in_use(self): + with self.port( + device_owner=nl_constants.DEVICE_OWNER_DVR_INTERFACE) as port: + with self.firewall_group(ports=[port['port']['id']]): + self._create_firewall_group( + self.fmt, + "firewall_group2", + self.DESCRIPTION, + ports=[port['port']['id']], + expected_res_status=webob.exc.HTTPConflict.code, + ) + def test_update_firewall_group_with_port_already_in_use(self): with self.port( device_owner=nl_constants.DEVICE_OWNER_ROUTER_INTF) as port: