[OVN] Fix db-sync-util Traceback when port security not enabled

Change-Id: Id1ed2922a908148b2b271bd28cc974ef424530d5
Closes-Bug: #1882061
(cherry picked from commit 58d1d0dbdd)
(add newline in neutron/tests/unit/common/ovn/test_utils.py to resolve merge
 conflict with https://review.opendev.org/#/c/738214/)

neutron/common/ovn/constants.py

@@ -264,6 +264,11 @@ UNKNOWN_ADDR = 'unknown'
 
 PORT_CAP_SWITCHDEV = 'switchdev'
 
+# The name of the port security group attribute is currently not in neutron nor
+# neutron-lib api definitions or constants. To avoid importing the extension
+# code directly we keep a copy here.
+PORT_SECURITYGROUPS = 'security_groups'
+
 # TODO(lucasagomes): Create constants for other LSP types
 LSP_TYPE_LOCALNET = 'localnet'
 LSP_TYPE_VIRTUAL = 'virtual'

neutron/common/ovn/utils.py

@@ -210,6 +210,10 @@ def is_port_security_enabled(port):
     return port.get(psec.PORTSECURITY)
 
 
+def is_security_groups_enabled(port):
+    return port.get(constants.PORT_SECURITYGROUPS)
+
+
 def validate_and_get_data_from_binding_profile(port):
     if (constants.OVN_PORT_BINDING_PROFILE not in port or
             not validators.is_attr_set(

neutron/plugins/ml2/drivers/ovn/mech_driver/ovsdb/ovn_db_sync.py

@@ -220,8 +220,8 @@ class OvnNbSynchronizer(OvnDbSynchronizer):
                     # already exists in OVN. The rest will be added during the
                     # ports sync operation later.
                     for n_port in db_ports:
-                        if ((n_port['security_groups'] or
-                                n_port['port_security_enabled']) and
+                        if ((utils.is_security_groups_enabled(n_port) or
+                                utils.is_port_security_enabled(n_port)) and
                                 n_port['id'] in ovn_ports):
                             txn.add(self.ovn_api.pg_add_ports(
                                 pg, n_port['id']))

neutron/tests/unit/common/ovn/test_utils.py

@@ -88,6 +88,12 @@ class TestUtils(base.BaseTestCase):
         self.assertEqual(
             [], utils.get_chassis_availability_zones(chassis))
 
+    def test_is_security_groups_enabled(self):
+        self.assertTrue(utils.is_security_groups_enabled(
+            {constants.PORT_SECURITYGROUPS: ['fake']}))
+        self.assertFalse(utils.is_security_groups_enabled(
+            {}))
+
 
 class TestGateWayChassisValidity(base.BaseTestCase):