openstack/ovsdbapp
Code Issues Proposed changes

[NB] Add "if_exists" flag to ACL and Port_Group deletion commands

Browse Source 
Related-Bug: #2084977
Change-Id: I64fd656fa7ce48e44bcf2881e7a06697335fd09f
This commit is contained in:
Rodolfo Alonso Hernandez 2024-11-06 11:20:44 +00:00
parent d603441459
commit e2d8d39fa1
4 changed files with 68 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
ovsdbapp/schema/ovn_northbound/api.py
View File

@ -128,7 +128,8 @@ class API(api.API, metaclass=abc.ABCMeta):
"""
@abc.abstractmethod
def acl_del(self, switch, direction=None, priority=None, match=None):
def acl_del(self, switch, direction=None, priority=None, match=None,
if_exists=False):
"""Remove ACLs from 'switch'
If only switch is supplied, all the ACLs from the logical switch are
@ -144,6 +145,9 @@ class API(api.API, metaclass=abc.ABCMeta):
:type priority: int
:param match: The match rule
:type match: string
:param if_exists: If True, don't fail if the parent logical switch
containing the ACL doesn't exist
:type if_exists: boolean
:returns: :class:`Command` with no result
"""
@ -189,7 +193,7 @@ class API(api.API, metaclass=abc.ABCMeta):
@abc.abstractmethod
def pg_acl_del(self, port_group, direction=None, priority=None,
match=None):
match=None, if_exists=False):
"""Remove ACLs from 'port_group'
If only port_group is supplied, all the ACLs from the logical switch
@ -205,6 +209,9 @@ class API(api.API, metaclass=abc.ABCMeta):
:type priority: int
:param match: The match rule
:type match: string
:param if_exists: If True, don't fail if the parent port group
containing the ACL doesn't exist
:type if_exists: boolean
:returns: :class:`Command` with no result
"""

16
ovsdbapp/schema/ovn_northbound/commands.py
View File

@ -190,13 +190,14 @@ class PgAclAddCommand(_AclAddHelper):
class _AclDelHelper(cmd.BaseCommand):
def __init__(self, api, entity, direction=None,
priority=None, match=None):
priority=None, match=None, if_exists=False):
if (priority is None) != (match is None):
raise TypeError("Must specify priority and match together")
if priority is not None and not direction:
raise TypeError("Cannot specify priority/match without direction")
super().__init__(api)
self.entity = entity
self.if_exists = if_exists
self.conditions = []
if direction:
self.conditions.append(('direction', '=', direction))
@ -206,7 +207,14 @@ class _AclDelHelper(cmd.BaseCommand):
('match', '=', match)]
def run_idl(self, txn):
try:
entity = self.api.lookup(self.lookup_table, self.entity)
except idlutils.RowNotFound as e:
if self.if_exists:
return
msg = "%s %s does not exist" % (self.lookup_table, self.entity)
raise RuntimeError(msg) from e
for acl in [a for a in entity.acls
if idlutils.row_match(a, self.conditions)]:
entity.delvalue('acls', acl)
@ -216,12 +224,6 @@ class _AclDelHelper(cmd.BaseCommand):
class AclDelCommand(_AclDelHelper):
lookup_table = 'Logical_Switch'
def __init__(self, api, switch, direction=None,
priority=None, match=None):
# NOTE: we're overriding the constructor here to not break any
# existing callers before we introduced Port Groups.
super().__init__(api, switch, direction, priority, match)
class PgAclDelCommand(_AclDelHelper):
lookup_table = 'Port_Group'

10
ovsdbapp/schema/ovn_northbound/impl_idl.py
View File

@ -61,8 +61,10 @@ class OvnNbApiIdlImpl(ovs_idl.Backend, api.API):
return cmd.AclAddCommand(self, switch, direction, priority,
match, action, log, may_exist, **external_ids)
def acl_del(self, switch, direction=None, priority=None, match=None):
return cmd.AclDelCommand(self, switch, direction, priority, match)
def acl_del(self, switch, direction=None, priority=None, match=None,
if_exists=False):
return cmd.AclDelCommand(self, switch, direction, priority, match,
if_exists)
def acl_list(self, switch):
return cmd.AclListCommand(self, switch)
@ -76,9 +78,9 @@ class OvnNbApiIdlImpl(ovs_idl.Backend, api.API):
**external_ids)
def pg_acl_del(self, port_group, direction=None, priority=None,
match=None):
match=None, if_exists=False):
return cmd.PgAclDelCommand(self, port_group, direction, priority,
match)
match, if_exists)
def pg_acl_list(self, port_group):
return cmd.PgAclListCommand(self, port_group)

43
ovsdbapp/tests/functional/schema/ovn_northbound/test_impl_idl.py
View File

@ -217,6 +217,27 @@ class TestAclOps(OvnNorthboundTest):
self.assertRaises(TypeError, self.api.acl_del, self.switch.uuid,
priority=0)
def test_acl_del_acl_not_present(self):
r1 = self._acl_add('lswitch', 'from-lport', 0,
'output == "fake_port"', 'drop')
cmd = self.api.acl_del(self.switch.uuid,
'from-lport', 0, 'output == "fake_port"')
cmd.execute(check_error=True)
self.assertNotIn(r1, self.switch.acls)
# The acl_del command is idempotent.
cmd.execute(check_error=True)
self.assertNotIn(r1, self.switch.acls)
def test_acl_del_if_exists_false(self):
cmd = self.api.acl_del('lswitch2', 'from-lport', 0, 'match')
self.assertRaises(RuntimeError, cmd.execute, check_error=True)
def test_acl_del_if_exists_true(self):
self.assertIsNone(
self.api.acl_del('lswitch2', 'from-lport', 0, 'match',
if_exists=True).execute(check_error=True))
def test_acl_list(self):
r1 = self._acl_add('lswitch', 'from-lport', 0,
'output == "fake_port"', 'drop')
@ -255,6 +276,28 @@ class TestAclOps(OvnNorthboundTest):
self.assertNotIn(r1.uuid, self.api.tables['ACL'].rows)
self.assertEqual([], self.port_group.acls)
def test_pg_acl_del_acl_not_present(self):
r1 = self._acl_add('port_group', 'from-lport', 0,
'output == "fake_port"', 'drop')
cmd = self.api.pg_acl_del(self.port_group.uuid)
cmd.execute(check_error=True)
self.assertNotIn(r1.uuid, self.api.tables['ACL'].rows)
self.assertEqual([], self.port_group.acls)
# The pg_acl_del command is idempotent.
cmd.execute(check_error=True)
self.assertNotIn(r1.uuid, self.api.tables['ACL'].rows)
self.assertEqual([], self.port_group.acls)
def test_pg_acl_del_if_exists_false(self):
cmd = self.api.pg_acl_del('port_group2')
self.assertRaises(RuntimeError, cmd.execute, check_error=True)
def test_pg_acl_del_if_exists_true(self):
self.assertIsNone(
self.api.pg_acl_del('port_group2',
if_exists=True).execute(check_error=True))
def test_pg_acl_list(self):
r1 = self._acl_add('port_group', 'from-lport', 0,
'output == "fake_port"', 'drop')