Add is_vm_port method to LogicalPort

Add is_vm_port method to LogicalPort, returning true if there is no
device owner (compute is the default) or owner begins with "compute:".

This should also solve current gate issues.

Change-Id: Ica5c5b7ddfd6233f553b5a3ecb39b638cfdec354
Co-Authored-By: Dima Kuznetsov <dima.kuznetsov@toganetworks.com>

dragonflow/controller/dhcp_app.py

@@ -321,18 +321,12 @@ class DHCPApp(df_base_app.DFlowApp):
             priority=const.PRIORITY_MEDIUM,
             match=match)
 
-    def _is_vm_port(self, lport):
-        owner = lport.get_device_owner()
-        if not owner or "compute" in owner:
-            return True
-        return False
-
     def add_local_port(self, lport):
         if not netaddr.valid_ipv4(lport.get_ip()):
             LOG.warning(_LW("No support for non IPv4 protocol"))
             return
 
-        if not self._is_vm_port(lport):
+        if not lport.is_vm_port():
             return
 
         subnet_id = lport.get_subnets()[0]

dragonflow/db/models/legacy.py

@@ -205,6 +205,15 @@ class LogicalPort(NbDbObject, UniqueKeyMixin):
     def get_extra_dhcp_opts(self):
         return self.inner_obj.get('extra_dhcp_opts', [])
 
+    def is_vm_port(self):
+        """
+        Return True if the device owner starts with 'compute:' (or is None)
+        """
+        owner = self.get_device_owner()
+        if not owner or owner.startswith("compute:"):
+            return True
+        return False
+
     def __str__(self):
         lport_with_exteral_dict = dict(self.inner_obj)
         lport_with_exteral_dict['external_dict'] = self.external_dict

dragonflow/tests/fullstack/test_l2_flows.py

@@ -43,7 +43,7 @@ class TestL2FLows(test_base.DFTestBase):
     def _get_vm_port(self, ip, mac):
         ports = self.nb_api.get_all_logical_ports()
         for port in ports:
-            if port.get_device_owner() == 'compute:None':
+            if port.is_vm_port():
                 if port.get_ip() == ip and port.get_mac() == mac:
                     return port
         return None

dragonflow/tests/fullstack/test_portsec_flows.py

@@ -43,7 +43,7 @@ class TestOVSFlowsForPortSecurity(test_base.DFTestBase):
     def _get_vm_port(self, ip, mac):
         ports = self.nb_api.get_all_logical_ports()
         for port in ports:
-            if port.get_device_owner() == 'compute:None':
+            if port.is_vm_port():
                 if port.get_ip() == ip and port.get_mac() == mac:
                     return port
         return None

dragonflow/tests/fullstack/test_sg_flows.py

@@ -190,7 +190,7 @@ class TestOVSFlowsForSecurityGroup(test_base.DFTestBase):
     def _get_vm_port(self, ip, mac):
         ports = self.nb_api.get_all_logical_ports()
         for port in ports:
-            if port.get_device_owner() == 'compute:None':
+            if port.is_vm_port():
                 if port.get_ip() == ip and port.get_mac() == mac:
                     return port
         return None

dragonflow/tests/fullstack/test_snat_flows.py

@@ -25,7 +25,7 @@ class TestSnatFlows(test_base.DFTestBase):
     def _get_vm_port(self, ip, mac):
         ports = self.nb_api.get_all_logical_ports()
         for port in ports:
-            if port.get_device_owner() == 'compute:None':
+            if port.is_vm_port():
                 if port.get_ip() == ip and port.get_mac() == mac:
                     return port
         return None