x/quark
Code Issues Proposed changes

Deleting a port without sg does not hit redis

Browse Source 
Introduced a check in security_groups.py which calls delete_vif
(which in turn hits redis) only if security_groups is present
in the kwargs passed to the delete_port function.

Introduced changes in test_unmanaged_driver.py which change
test_delete_port by explicitly passing a security_group kwarg &
test_delete_port_redis_is_dead by asserting that delete_vif
isn't  called as there is no security_group kwarg. Added a new
test case test_delete_port_no_security_group that verifies that
delete_vif isn't called if no security_group kwarg (or an empty
list) is passed.

Change-Id: I00648f6d490e883175d6592ae64e8c742d35a035
Closes-Bug: #1747417
This commit is contained in:
Sourav Banerjee 2017-11-02 19:52:13 +05:30
parent 025efa5558
commit 6e9ae2fc1b
2 changed files with 34 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
quark/drivers/security_groups.py
View File

@ -42,11 +42,12 @@ class SecurityGroupDriver(object):
def delete_port(self, **kwargs): def delete_port(self, **kwargs):
client = sg_client.SecurityGroupsClient() client = sg_client.SecurityGroupsClient()
try: try:
device_id = kwargs.get('device_id') if kwargs.get('security_groups'):
mac_address = kwargs.get('mac_address') device_id = kwargs.get('device_id')
if not device_id or not mac_address: mac_address = kwargs.get('mac_address')
LOG.warning('device_id or mac_address not given, ignored.') if not device_id or not mac_address:
return LOG.warning('device_id or mac_address not given, ignored.')
client.delete_vif(device_id, mac_address) return
client.delete_vif(device_id, mac_address)
except Exception: except Exception:
LOG.exception("Failed to reach the security groups backend") LOG.exception("Failed to reach the security groups backend")

31
quark/tests/test_unmanaged_driver.py
View File

@ -113,14 +113,38 @@ class TestUnmanagedDriver(test_base.TestBase):
@mock.patch("quark.cache.security_groups_client.SecurityGroupsClient") @mock.patch("quark.cache.security_groups_client.SecurityGroupsClient")
def test_delete_port(self, sg_cli): def test_delete_port(self, sg_cli):
device_id = str(uuid.uuid4())
mac_address = netaddr.EUI("AA:BB:CC:DD:EE:FF").value
security_groups = [str(uuid.uuid4())]
mock_client = mock.MagicMock()
sg_cli.return_value = mock_client
self.driver.delete_port(context=self.context, port_id=2,
mac_address=mac_address, device_id=device_id,
security_groups=security_groups)
mock_client.delete_vif.assert_called_once_with(
device_id, mac_address)
@mock.patch("quark.cache.security_groups_client.SecurityGroupsClient")
def test_delete_port_empty_security_group(self, sg_cli):
device_id = str(uuid.uuid4())
mac_address = netaddr.EUI("AA:BB:CC:DD:EE:FF").value
security_groups = []
mock_client = mock.MagicMock()
sg_cli.return_value = mock_client
self.driver.delete_port(context=self.context, port_id=2,
mac_address=mac_address, device_id=device_id,
security_groups=security_groups)
mock_client.delete_vif.assert_not_called()
@mock.patch("quark.cache.security_groups_client.SecurityGroupsClient")
def test_delete_port_no_security_group(self, sg_cli):
device_id = str(uuid.uuid4()) device_id = str(uuid.uuid4())
mac_address = netaddr.EUI("AA:BB:CC:DD:EE:FF").value mac_address = netaddr.EUI("AA:BB:CC:DD:EE:FF").value
mock_client = mock.MagicMock() mock_client = mock.MagicMock()
sg_cli.return_value = mock_client sg_cli.return_value = mock_client
self.driver.delete_port(context=self.context, port_id=2, self.driver.delete_port(context=self.context, port_id=2,
mac_address=mac_address, device_id=device_id) mac_address=mac_address, device_id=device_id)
mock_client.delete_vif.assert_called_once_with( mock_client.delete_vif.assert_not_called()
device_id, mac_address)
@mock.patch("quark.cache.security_groups_client.SecurityGroupsClient") @mock.patch("quark.cache.security_groups_client.SecurityGroupsClient")
def test_delete_port_redis_is_dead(self, sg_cli): def test_delete_port_redis_is_dead(self, sg_cli):
@ -134,8 +158,7 @@ class TestUnmanagedDriver(test_base.TestBase):
self.driver.delete_port(context=self.context, port_id=2, self.driver.delete_port(context=self.context, port_id=2,
mac_address=mac_address, mac_address=mac_address,
device_id=device_id) device_id=device_id)
mock_client.delete_vif.assert_called_once_with( mock_client.delete_vif.assert_not_called()
device_id, mac_address)
except Exception: except Exception:
# This test fails without the exception handling in # This test fails without the exception handling in
# _delete_port_security_groups # _delete_port_security_groups