Fix associating firewall group with DVR/L3HA port

This commit enables to specify DVR/L3HA port for firewall group. We can
select a port with following device_owner in creating/updating firewall
group.

    * DVR:  'network:router_interface_distributed'
    * L3HA: 'network:ha_router_replicated_interface'

Co-Authored-By: Nguyen Phuong An <annp@vn.fujitsu.com>
Change-Id: I05f0f652f3e43d5c1ce5ae7933991cf92a418920
Closes-Bug: #1762454

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)

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):

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: