diff options
Diffstat (limited to 'networking-odl/networking_odl/tests/unit/journal')
4 files changed, 0 insertions, 289 deletions
diff --git a/networking-odl/networking_odl/tests/unit/journal/__init__.py b/networking-odl/networking_odl/tests/unit/journal/__init__.py deleted file mode 100644 index e69de29..0000000 --- a/networking-odl/networking_odl/tests/unit/journal/__init__.py +++ /dev/null diff --git a/networking-odl/networking_odl/tests/unit/journal/test_dependency_validations.py b/networking-odl/networking_odl/tests/unit/journal/test_dependency_validations.py deleted file mode 100644 index 39a4b98..0000000 --- a/networking-odl/networking_odl/tests/unit/journal/test_dependency_validations.py +++ /dev/null @@ -1,44 +0,0 @@ -# -# Copyright (C) 2016 Intel Corp. Isaku Yamahata <isaku.yamahata@gmail com> -# All Rights Reserved. -# -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. -# - -import mock - -from neutron.tests import base - -from networking_odl.journal import dependency_validations - - -class DependencyValidationsTestCase(base.DietTestCase): - _RESOURCE_DUMMY = 'test_type' - - def setUp(self): - super(DependencyValidationsTestCase, self).setUp() - mock_validation_map = mock.patch.dict( - dependency_validations._VALIDATION_MAP) - mock_validation_map.start() - self.addCleanup(mock_validation_map.stop) - - def test_register_validator(self): - mock_session = mock.Mock() - mock_validator = mock.Mock(return_value=False) - mock_row = mock.Mock() - mock_row.object_type = self._RESOURCE_DUMMY - dependency_validations.register_validator(self._RESOURCE_DUMMY, - mock_validator) - valid = dependency_validations.validate(mock_session, mock_row) - mock_validator.assert_called_once_with(mock_session, mock_row) - self.assertFalse(valid) diff --git a/networking-odl/networking_odl/tests/unit/journal/test_full_sync.py b/networking-odl/networking_odl/tests/unit/journal/test_full_sync.py deleted file mode 100644 index cedccbd..0000000 --- a/networking-odl/networking_odl/tests/unit/journal/test_full_sync.py +++ /dev/null @@ -1,152 +0,0 @@ -# -# Copyright (C) 2016 Red Hat, Inc. -# -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. -# - -import mock -import requests - -from neutron.db import api as neutron_db_api -from neutron import manager -from neutron.tests.unit.testlib_api import SqlTestCaseLight - -from networking_odl.common import constants as odl_const -from networking_odl.db import db -from networking_odl.db import models -from networking_odl.journal import full_sync - - -class FullSyncTestCase(SqlTestCaseLight): - def setUp(self): - super(FullSyncTestCase, self).setUp() - self.db_session = neutron_db_api.get_session() - - full_sync._CLIENT = mock.MagicMock() - self.plugin_mock = mock.patch.object(manager.NeutronManager, - 'get_plugin').start() - self.l3_plugin_mock = mock.patch.object(manager.NeutronManager, - 'get_service_plugins').start() - - self.addCleanup(self._db_cleanup) - - def _db_cleanup(self): - self.db_session.query(models.OpendaylightJournal).delete() - - def test_no_full_sync_when_canary_exists(self): - full_sync.full_sync(self.db_session) - self.assertEqual([], db.get_all_db_rows(self.db_session)) - - def _mock_l2_resources(self): - expected_journal = {odl_const.ODL_NETWORK: '1', - odl_const.ODL_SUBNET: '2', - odl_const.ODL_PORT: '3'} - plugin_instance = self.plugin_mock.return_value - plugin_instance.get_networks.return_value = [ - {'id': expected_journal[odl_const.ODL_NETWORK]}] - plugin_instance.get_subnets.return_value = [ - {'id': expected_journal[odl_const.ODL_SUBNET]}] - plugin_instance.get_ports.side_effect = ([ - {'id': expected_journal[odl_const.ODL_PORT]}], []) - return expected_journal - - def _filter_out_canary(self, rows): - return [row for row in rows if row['object_uuid'] != - full_sync._CANARY_NETWORK_ID] - - def _test_no_full_sync_when_canary_in_journal(self, state): - self._mock_canary_missing() - self._mock_l2_resources() - db.create_pending_row(self.db_session, odl_const.ODL_NETWORK, - full_sync._CANARY_NETWORK_ID, - odl_const.ODL_CREATE, {}) - row = db.get_all_db_rows(self.db_session)[0] - db.update_db_row_state(self.db_session, row, state) - - full_sync.full_sync(self.db_session) - - rows = db.get_all_db_rows(self.db_session) - self.assertEqual([], self._filter_out_canary(rows)) - - def test_no_full_sync_when_canary_pending_creation(self): - self._test_no_full_sync_when_canary_in_journal(odl_const.PENDING) - - def test_no_full_sync_when_canary_is_processing(self): - self._test_no_full_sync_when_canary_in_journal(odl_const.PROCESSING) - - def test_client_error_propagates(self): - class TestException(Exception): - def __init__(self): - pass - - full_sync._CLIENT.get.side_effect = TestException() - self.assertRaises(TestException, full_sync.full_sync, self.db_session) - - def _mock_canary_missing(self): - get_return = mock.MagicMock() - get_return.status_code = requests.codes.not_found - full_sync._CLIENT.get.return_value = get_return - - def _assert_canary_created(self): - rows = db.get_all_db_rows(self.db_session) - self.assertTrue(any(r['object_uuid'] == full_sync._CANARY_NETWORK_ID - for r in rows)) - return rows - - def _test_full_sync_resources(self, expected_journal): - self._mock_canary_missing() - - full_sync.full_sync(self.db_session) - - rows = self._assert_canary_created() - rows = self._filter_out_canary(rows) - self.assertItemsEqual(expected_journal.keys(), - [row['object_type'] for row in rows]) - for row in rows: - self.assertEqual(expected_journal[row['object_type']], - row['object_uuid']) - - def test_full_sync_removes_pending_rows(self): - db.create_pending_row(self.db_session, odl_const.ODL_NETWORK, "uuid", - odl_const.ODL_CREATE, {'foo': 'bar'}) - self._test_full_sync_resources({}) - - def test_full_sync_no_resources(self): - self._test_full_sync_resources({}) - - def test_full_sync_l2_resources(self): - self._test_full_sync_resources(self._mock_l2_resources()) - - def _mock_router_port(self, port_id): - router_port = {'id': port_id, - 'device_id': '1', - 'tenant_id': '1', - 'fixed_ips': [{'subnet_id': '1'}]} - plugin_instance = self.plugin_mock.return_value - plugin_instance.get_ports.side_effect = ([], [router_port]) - - def _mock_l3_resources(self): - expected_journal = {odl_const.ODL_ROUTER: '1', - odl_const.ODL_FLOATINGIP: '2', - odl_const.ODL_ROUTER_INTF: '3'} - plugin_instance = self.l3_plugin_mock.return_value.get.return_value - plugin_instance.get_routers.return_value = [ - {'id': expected_journal[odl_const.ODL_ROUTER]}] - plugin_instance.get_floatingips.return_value = [ - {'id': expected_journal[odl_const.ODL_FLOATINGIP]}] - self._mock_router_port(expected_journal[odl_const.ODL_ROUTER_INTF]) - - return expected_journal - - def test_full_sync_l3_resources(self): - self._test_full_sync_resources(self._mock_l3_resources()) diff --git a/networking-odl/networking_odl/tests/unit/journal/test_maintenance.py b/networking-odl/networking_odl/tests/unit/journal/test_maintenance.py deleted file mode 100644 index eb823cd..0000000 --- a/networking-odl/networking_odl/tests/unit/journal/test_maintenance.py +++ /dev/null @@ -1,93 +0,0 @@ -# -# Copyright (C) 2016 Red Hat, Inc. -# -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. -# - -import mock -import threading -from unittest2.case import TestCase - -from neutron.db import api as neutron_db_api -from neutron.tests.unit.testlib_api import SqlTestCaseLight - -from networking_odl.common import constants as odl_const -from networking_odl.db import models -from networking_odl.journal import maintenance - - -class MaintenanceThreadTestCase(SqlTestCaseLight, TestCase): - def setUp(self): - super(MaintenanceThreadTestCase, self).setUp() - self.db_session = neutron_db_api.get_session() - - row = models.OpendaylightMaintenance(state=odl_const.PENDING) - self.db_session.add(row) - self.db_session.flush() - - self.thread = maintenance.MaintenanceThread() - self.thread.maintenance_interval = 0.01 - - def test__execute_op_no_exception(self): - with mock.patch.object(maintenance, 'LOG') as mock_log: - operation = mock.MagicMock() - operation.__name__ = "test" - self.thread._execute_op(operation, self.db_session) - self.assertTrue(operation.called) - self.assertTrue(mock_log.info.called) - self.assertFalse(mock_log.exception.called) - - def test__execute_op_with_exception(self): - with mock.patch.object(maintenance, 'LOG') as mock_log: - operation = mock.MagicMock(side_effect=Exception()) - operation.__name__ = "test" - self.thread._execute_op(operation, self.db_session) - self.assertTrue(mock_log.exception.called) - - def test_thread_works(self): - callback_event = threading.Event() - count = [0] - - def callback_op(**kwargs): - count[0] += 1 - - # The following should be true on the second call, so we're making - # sure that the thread runs more than once. - if count[0] > 1: - callback_event.set() - - self.thread.register_operation(callback_op) - self.thread.start() - - # Make sure the callback event was called and not timed out - self.assertTrue(callback_event.wait(timeout=5)) - - def test_thread_continues_after_exception(self): - exception_event = threading.Event() - callback_event = threading.Event() - - def exception_op(**kwargs): - if not exception_event.is_set(): - exception_event.set() - raise Exception() - - def callback_op(**kwargs): - callback_event.set() - - for op in [exception_op, callback_op]: - self.thread.register_operation(op) - - self.thread.start() - - # Make sure the callback event was called and not timed out - self.assertTrue(callback_event.wait(timeout=5)) |