Add new function ls_get_localnet_ports

This is used to detect if a given logical_switch has localnet
ports associated to it, and return them

Change-Id: Iabfc1bd0a1b2c6998a013d19c47cda10a3879c82

ovsdbapp/constants.py

@@ -39,3 +39,5 @@ PROTO_TCP = 'tcp'
 PROTO_UDP = 'udp'
 
 ROUTE_DISCARD = "discard"
+
+LOCALNET = 'localnet'

ovsdbapp/schema/ovn_northbound/api.py

@@ -40,11 +40,11 @@ class API(api.API, metaclass=abc.ABCMeta):
     def ls_del(self, switch, if_exists=False):
         """Delete logical switch 'switch' and all its ports
 
-        :param switch:   The name or uuid of the switch
-        :type switch:    string or uuid.UUID
-        :type if_exists: If True, don't fail if the switch doesn't exist
-        :type if_exists: boolean
-        :returns:        :class:`Command` with no result
+        :param switch:    The name or uuid of the switch
+        :type switch:     string or uuid.UUID
+        :param if_exists: If True, don't fail if the switch doesn't exist
+        :type if_exists:  boolean
+        :returns:         :class:`Command` with no result
         """
 
     @abc.abstractmethod
@@ -96,6 +96,17 @@ class API(api.API, metaclass=abc.ABCMeta):
         :returns: :class:`Command` with RowView result
         """
 
+    @abc.abstractmethod
+    def ls_get_localnet_ports(self, switch, if_exists=False):
+        """Get localnet ports on logical switch 'switch'
+
+        :param switch:    The name or uuid of the switch
+        :type switch:     string or uuid.UUID
+        :param if_exists: If True, don't fail if the switch doesn't exist
+        :type if_exists:  boolean
+        :returns: :class:`Command` with the RowView list result
+        """
+
     @abc.abstractmethod
     def acl_add(self, switch, direction, priority, match, action, log=False):
         """Add an ACL to 'switch'

ovsdbapp/schema/ovn_northbound/commands.py

@@ -80,6 +80,28 @@ class LsGetCommand(cmd.BaseGetRowCommand):
     table = 'Logical_Switch'
 
 
+class LSGetLocalnetPortsCommand(cmd.ReadOnlyCommand):
+    def __init__(self, api, switch, if_exists=False):
+        super().__init__(api)
+        self.switch = switch
+        self.if_exists = if_exists
+
+    def localnet_port(self, row):
+        return row.type == const.LOCALNET
+
+    def run_idl(self, txn):
+        try:
+            lswitch = self.api.lookup('Logical_Switch', self.switch)
+            self.result = [rowview.RowView(p) for p in lswitch.ports
+                           if self.localnet_port(p)]
+        except idlutils.RowNotFound as e:
+            if self.if_exists:
+                self.result = []
+                return
+            msg = "Logical Switch %s does not exist" % self.switch
+            raise RuntimeError(msg) from e
+
+
 class _AclAddHelper(cmd.AddCommand):
     table_name = 'ACL'
 

ovsdbapp/schema/ovn_northbound/impl_idl.py

@@ -53,6 +53,9 @@ class OvnNbApiIdlImpl(ovs_idl.Backend, api.API):
         return self.db_remove('Logical_Switch', switch_uuid, 'dns_records',
                               dns_uuid, if_exists=if_exists)
 
+    def ls_get_localnet_ports(self, switch, if_exists=False):
+        return cmd.LSGetLocalnetPortsCommand(self, switch, if_exists)
+
     def acl_add(self, switch, direction, priority, match, action, log=False,
                 may_exist=False, **external_ids):
         return cmd.AclAddCommand(self, switch, direction, priority,

ovsdbapp/tests/functional/schema/ovn_northbound/test_impl_idl.py

@@ -41,6 +41,11 @@ class TestLogicalSwitchOps(OvnNorthboundTest):
         self.assertIn(fix.obj.uuid, self.table.rows)
         return fix.obj
 
+    def _lsp_add(self, switch, name, *args, **kwargs):
+        name = utils.get_rand_device_name() if name is None else name
+        self.api.lsp_add(switch.uuid, name, *args, **kwargs).execute(
+            check_error=True)
+
     def _test_ls_get(self, col):
         ls = self._ls_add(switch=utils.get_rand_device_name())
         val = getattr(ls, col)
@@ -102,6 +107,30 @@ class TestLogicalSwitchOps(OvnNorthboundTest):
         switch_set = set(self.api.ls_list().execute(check_error=True))
         self.assertTrue(switches.issubset(switch_set))
 
+    def test_ls_get_localnet_ports(self):
+        ls = self._ls_add()
+        self._lsp_add(ls, None, type=const.LOCALNET)
+        localnet_ports = self.api.ls_get_localnet_ports(ls.uuid).execute(
+            check_error=True)
+        self.assertEqual(ls.ports, localnet_ports)
+
+    def test_ls_get_localnet_ports_no_ports(self):
+        ls = self._ls_add()
+        localnet_ports = self.api.ls_get_localnet_ports(ls.uuid).execute(
+            check_error=True)
+        self.assertEqual([], localnet_ports)
+
+    def test_ls_get_localnet_ports_no_exist(self):
+        name = utils.get_rand_device_name()
+        cmd = self.api.ls_get_localnet_ports(name)
+        self.assertRaises(RuntimeError, cmd.execute, check_error=True)
+
+    def test_ls_get_localnet_ports_if_exists(self):
+        name = utils.get_rand_device_name()
+        localnet_ports = self.api.ls_get_localnet_ports(
+            name, if_exists=True).execute(check_error=True)
+        self.assertEqual([], localnet_ports)
+
 
 class TestAclOps(OvnNorthboundTest):
     def setUp(self):