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, 0 insertions, 152 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
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())