diff options
Diffstat (limited to 'networking-odl/networking_odl/tests/unit/common/test_callback.py')
| -rw-r--r-- | networking-odl/networking_odl/tests/unit/common/test_callback.py | 136 |
1 files changed, 95 insertions, 41 deletions
diff --git a/networking-odl/networking_odl/tests/unit/common/test_callback.py b/networking-odl/networking_odl/tests/unit/common/test_callback.py index f5e2ee6..b7720b3 100644 --- a/networking-odl/networking_odl/tests/unit/common/test_callback.py +++ b/networking-odl/networking_odl/tests/unit/common/test_callback.py @@ -18,6 +18,7 @@ from networking_odl.common import constants as odl_const from networking_odl.ml2.mech_driver import OpenDaylightDriver import mock +import testscenarios import testtools from neutron.callbacks import events @@ -27,57 +28,110 @@ from neutron.callbacks import resources FAKE_ID = 'fakeid' -class ODLCallbackTestCase(testtools.TestCase): - odl_driver = OpenDaylightDriver() - sgh = callback.OdlSecurityGroupsHandler(odl_driver) +class ODLCallbackTestCase(testscenarios.WithScenarios, testtools.TestCase): + odl_client = OpenDaylightDriver() + scenarios = [ + ('after', { + 'sgh': callback.OdlSecurityGroupsHandler(odl_client, + "AFTER")}), + ('precommit', { + 'sgh': callback.OdlSecurityGroupsHandler(odl_client, + "PRECOMMIT")}), + ] def setUp(self): super(ODLCallbackTestCase, self).setUp() @mock.patch.object(OpenDaylightDriver, 'sync_from_callback') - def _test_callback_for_sg(self, event, op, sg, sg_id, sfc): + def test_callback_sg_create(self, sfc): + context = mock.Mock() + sg = mock.Mock() + default_sg = mock.Mock() + kwargs = { + 'context': context, + 'security_group': sg, + 'security_groups': odl_const.ODL_SGS, + 'is_default': default_sg, + } self.sgh.sg_callback(resources.SECURITY_GROUP, - event, - None, - security_group=sg, - security_group_id=sg_id) + events.AFTER_CREATE, + "trigger", + **kwargs) - expected_dict = ({resources.SECURITY_GROUP: sg} - if sg is not None else None) - sfc.assert_called_with( - op, callback._RESOURCE_MAPPING[resources.SECURITY_GROUP], sg_id, - expected_dict) + sfc.assert_called_with(odl_const.ODL_CREATE, + 'security-groups', + None, {'security_group': sg}) - def test_callback_sg_create(self): - self._test_callback_for_sg(events.AFTER_CREATE, odl_const.ODL_CREATE, - mock.Mock(), None) + @mock.patch.object(OpenDaylightDriver, 'sync_from_callback') + def test_callback_sg_update(self, sfc): + context = mock.Mock() + sg = mock.Mock() + kwargs = { + 'context': context, + 'security_group_id': FAKE_ID, + 'security_group': sg, + 'security_groups': odl_const.ODL_SGS, + } + self.sgh.sg_callback(resources.SECURITY_GROUP, + events.AFTER_UPDATE, + "trigger", + **kwargs) + + sfc.assert_called_with(odl_const.ODL_UPDATE, + 'security-groups', + FAKE_ID, {'security_group': sg}) + + @mock.patch.object(OpenDaylightDriver, 'sync_from_callback') + def test_callback_sg_delete(self, sfc): + context = mock.Mock() + sg = mock.Mock() + kwargs = { + 'context': context, + 'security_group_id': FAKE_ID, + 'security_group': sg, + 'security_groups': odl_const.ODL_SGS, + } + self.sgh.sg_callback(resources.SECURITY_GROUP, + events.AFTER_DELETE, + "trigger", + **kwargs) - def test_callback_sg_update(self): - self._test_callback_for_sg(events.AFTER_UPDATE, odl_const.ODL_UPDATE, - mock.Mock(), FAKE_ID) + sfc.assert_called_with(odl_const.ODL_DELETE, + 'security-groups', + FAKE_ID, {'security_group': sg}) - def test_callback_sg_delete(self): - self._test_callback_for_sg(events.AFTER_DELETE, odl_const.ODL_DELETE, - None, FAKE_ID) + @mock.patch.object(OpenDaylightDriver, 'sync_from_callback') + def test_callback_sg_rules_create(self, sfc): + context = mock.Mock() + security_group_rule = mock.Mock() + kwargs = { + 'context': context, + 'security_group_rule': security_group_rule, + 'security_group_rules': odl_const.ODL_SG_RULES, + } + self.sgh.sg_callback(resources.SECURITY_GROUP_RULE, + events.AFTER_CREATE, + "trigger", + **kwargs) + + sfc.assert_called_with(odl_const.ODL_CREATE, + 'security-group-rules', + None, + {'security_group_rule': security_group_rule}) @mock.patch.object(OpenDaylightDriver, 'sync_from_callback') - def _test_callback_for_sg_rules(self, event, op, sg_rule, sg_rule_id, sfc): + def test_callback_sg_rules_delete(self, sfc): + context = mock.Mock() + kwargs = { + 'context': context, + 'security_group_rule_id': FAKE_ID, + 'security_group_rules': odl_const.ODL_SG_RULES, + } self.sgh.sg_callback(resources.SECURITY_GROUP_RULE, - event, - None, - security_group_rule=sg_rule, - security_group_rule_id=sg_rule_id) - - expected_dict = ({resources.SECURITY_GROUP_RULE: sg_rule} - if sg_rule is not None else None) - sfc.assert_called_with( - op, callback._RESOURCE_MAPPING[resources.SECURITY_GROUP_RULE], - sg_rule_id, expected_dict) - - def test_callback_sg_rules_create(self): - self._test_callback_for_sg_rules( - events.AFTER_CREATE, odl_const.ODL_CREATE, mock.Mock(), None) - - def test_callback_sg_rules_delete(self): - self._test_callback_for_sg_rules( - events.AFTER_DELETE, odl_const.ODL_DELETE, None, FAKE_ID) + events.AFTER_DELETE, + "trigger", + **kwargs) + + sfc.assert_called_with(odl_const.ODL_DELETE, + 'security-group-rules', + FAKE_ID, None) |