summaryrefslogtreecommitdiffstats
path: root/networking-odl/networking_odl/tests/unit/journal/test_full_sync.py
diff options
context:
space:
mode:
Diffstat (limited to 'networking-odl/networking_odl/tests/unit/journal/test_full_sync.py')
-rw-r--r--networking-odl/networking_odl/tests/unit/journal/test_full_sync.py152
1 files changed, 152 insertions, 0 deletions
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
new file mode 100644
index 0000000..cedccbd
--- /dev/null
+++ b/networking-odl/networking_odl/tests/unit/journal/test_full_sync.py
@@ -0,0 +1,152 @@
+#
+# 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())