--- /dev/null
+# -*- coding: utf-8 -*-
+##
+# Copyright 2018 University of Bristol - High Performance Networks Research
+# Group
+# All Rights Reserved.
+#
+# Contributors: Anderson Bravalheri, Dimitrios Gkounis, Abubakar Siddique
+# Muqaddas, Navdeep Uniyal, Reza Nejabati and Dimitra Simeonidou
+#
+# 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.
+#
+# For those usages not covered by the Apache License, Version 2.0 please
+# contact with: <highperformance-networks@bristol.ac.uk>
+#
+# Neither the name of the University of Bristol nor the names of its
+# contributors may be used to endorse or promote products derived from
+# this software without specific prior written permission.
+#
+# This work has been performed in the context of DCMS UK 5G Testbeds
+# & Trials Programme and in the framework of the Metro-Haul project -
+# funded by the European Commission under Grant number 761727 through the
+# Horizon 2020 and 5G-PPP programmes.
+##
+
+import unittest
+from difflib import unified_diff
+from operator import itemgetter
+from time import time
+
+import json
+
+from unittest.mock import MagicMock, patch
+
+from . import fixtures as eg
+from ...tests.db_helpers import (
+ TestCaseWithDatabasePerTest,
+ disable_foreign_keys,
+ uuid
+)
+from ..engine import WimEngine
+from ..persistence import WimPersistence
+from ..wim_thread import WimThread
+
+
+ignore_connector = patch('osm_ro.wim.wim_thread.CONNECTORS', MagicMock())
+
+
+def _repr(value):
+ return json.dumps(value, indent=4, sort_keys=True)
+
+
+@ignore_connector
+class TestWimThreadWithDb(TestCaseWithDatabasePerTest):
+ def setUp(self):
+ super(TestWimThreadWithDb, self).setUp()
+ self.persist = WimPersistence(self.db)
+ wim = eg.wim(0)
+ account = eg.wim_account(0, 0)
+ account['wim'] = wim
+ self.thread = WimThread(self.persist, {}, account)
+ self.thread.connector = MagicMock()
+
+ def assertTasksEqual(self, left, right):
+ fields = itemgetter('item', 'item_id', 'action', 'status')
+ left_ = (t.as_dict() for t in left)
+ left_ = [fields(t) for t in left_]
+ right_ = [fields(t) for t in right]
+
+ try:
+ self.assertItemsEqual(left_, right_)
+ except AssertionError:
+ print('left', _repr(left))
+ print('left', len(left_), 'items')
+ print('right', len(right_), 'items')
+ result = list(unified_diff(_repr(sorted(left_)).split('\n'),
+ _repr(sorted(right_)).split('\n'),
+ 'left', 'right'))
+ print('diff:\n', '\n'.join(result))
+ raise
+
+ def test_reload_actions__all_create(self):
+ # Given we have 3 CREATE actions stored in the database
+ actions = eg.wim_actions('CREATE',
+ action_id=uuid('action0'), num_links=3)
+ self.populate([
+ {'nfvo_tenants': eg.tenant()}
+ ] + eg.wim_set() + [
+ {'instance_actions':
+ eg.instance_action(action_id=uuid('action0'))},
+ {'vim_wim_actions': actions}
+ ])
+
+ # When we reload the tasks
+ self.thread.reload_actions()
+ # All of them should be inserted as pending
+ self.assertTasksEqual(self.thread.pending_tasks, actions)
+
+ def test_reload_actions__all_refresh(self):
+ # Given just DONE tasks are in the database
+ actions = eg.wim_actions(status='DONE',
+ action_id=uuid('action0'), num_links=3)
+ self.populate([
+ {'nfvo_tenants': eg.tenant()}
+ ] + eg.wim_set() + [
+ {'instance_actions':
+ eg.instance_action(action_id=uuid('action0'))},
+ {'vim_wim_actions': actions}
+ ])
+
+ # When we reload the tasks
+ self.thread.reload_actions()
+ # All of them should be inserted as refresh
+ self.assertTasksEqual(self.thread.refresh_tasks, actions)
+
+ def test_reload_actions__grouped(self):
+ # Given we have 2 tasks for the same item in the database
+ kwargs = {'action_id': uuid('action0')}
+ actions = (eg.wim_actions('CREATE', **kwargs) +
+ eg.wim_actions('FIND', **kwargs))
+ for i, action in enumerate(actions):
+ action['task_index'] = i
+
+ self.populate([
+ {'nfvo_tenants': eg.tenant()}
+ ] + eg.wim_set() + [
+ {'instance_actions': eg.instance_action(**kwargs)},
+ {'vim_wim_actions': actions}
+ ])
+
+ # When we reload the tasks
+ self.thread.reload_actions()
+ # Just one group should be created
+ self.assertEqual(len(self.thread.grouped_tasks.values()), 1)
+
+ def test_reload_actions__delete_scheduled(self):
+ # Given we have 3 tasks for the same item in the database, but one of
+ # them is a DELETE task and it is SCHEDULED
+ kwargs = {'action_id': uuid('action0')}
+ actions = (eg.wim_actions('CREATE', **kwargs) +
+ eg.wim_actions('FIND', **kwargs) +
+ eg.wim_actions('DELETE', status='SCHEDULED', **kwargs))
+ for i, action in enumerate(actions):
+ action['task_index'] = i
+
+ self.populate([
+ {'nfvo_tenants': eg.tenant()}
+ ] + eg.wim_set() + [
+ {'instance_actions': eg.instance_action(**kwargs)},
+ {'vim_wim_actions': actions}
+ ])
+
+ # When we reload the tasks
+ self.thread.reload_actions()
+ # Just one group should be created
+ self.assertEqual(len(self.thread.grouped_tasks.values()), 1)
+
+ def test_reload_actions__delete_done(self):
+ # Given we have 3 tasks for the same item in the database, but one of
+ # them is a DELETE task and it is not SCHEDULED
+ kwargs = {'action_id': uuid('action0')}
+ actions = (eg.wim_actions('CREATE', **kwargs) +
+ eg.wim_actions('FIND', **kwargs) +
+ eg.wim_actions('DELETE', status='DONE', **kwargs))
+ for i, action in enumerate(actions):
+ action['task_index'] = i
+
+ self.populate([
+ {'nfvo_tenants': eg.tenant()}
+ ] + eg.wim_set() + [
+ {'instance_actions': eg.instance_action(**kwargs)},
+ {'vim_wim_actions': actions}
+ ])
+
+ # When we reload the tasks
+ self.thread.reload_actions()
+ # No pending task should be found
+ self.assertEqual(self.thread.pending_tasks, [])
+
+ def test_reload_actions__batch(self):
+ # Given the group_limit is 10, and we have 24
+ group_limit = 10
+ kwargs = {'action_id': uuid('action0')}
+ actions = (eg.wim_actions('CREATE', num_links=8, **kwargs) +
+ eg.wim_actions('FIND', num_links=8, **kwargs) +
+ eg.wim_actions('FIND', num_links=8, **kwargs))
+ for i, action in enumerate(actions):
+ action['task_index'] = i
+
+ self.populate([
+ {'nfvo_tenants': eg.tenant()}
+ ] + eg.wim_set() + [
+ {'instance_actions': eg.instance_action(**kwargs)},
+ {'vim_wim_actions': actions}
+ ])
+
+ # When we reload the tasks
+ self.thread.reload_actions(group_limit)
+
+ # Then we should still see the actions in memory properly
+ self.assertTasksEqual(self.thread.pending_tasks, actions)
+ self.assertEqual(len(self.thread.grouped_tasks.values()), 8)
+
+ @disable_foreign_keys
+ def test_process_list__refresh(self):
+ update_wan_link = MagicMock(wrap=self.persist.update_wan_link)
+ update_action = MagicMock(wrap=self.persist.update_wan_link)
+ patches = dict(update_wan_link=update_wan_link,
+ update_action=update_action)
+
+ with patch.multiple(self.persist, **patches):
+ # Given we have 2 tasks in the refresh queue
+ kwargs = {'action_id': uuid('action0')}
+ actions = (eg.wim_actions('FIND', 'DONE', **kwargs) +
+ eg.wim_actions('CREATE', 'BUILD', **kwargs))
+ for i, action in enumerate(actions):
+ action['task_index'] = i
+
+ self.populate(
+ [{'instance_wim_nets': eg.instance_wim_nets()}] +
+ [{'instance_actions':
+ eg.instance_action(num_tasks=2, **kwargs)}] +
+ [{'vim_wim_actions': actions}])
+
+ self.thread.insert_pending_tasks(actions)
+
+ # When we process the refresh list
+ processed = self.thread.process_list('refresh')
+
+ # Then we should have 2 updates
+ self.assertEqual(processed, 2)
+
+ # And the database should be updated accordingly
+ self.assertEqual(update_wan_link.call_count, 2)
+ self.assertEqual(update_action.call_count, 2)
+
+ @disable_foreign_keys
+ def test_delete_superseed_create(self):
+ # Given we insert a scheduled CREATE task
+ instance_action = eg.instance_action(num_tasks=1)
+ self.thread.pending_tasks = []
+ engine = WimEngine(persistence=self.persist)
+ self.addCleanup(engine.stop_threads)
+ wan_links = eg.instance_wim_nets()
+ create_actions = engine.create_actions(wan_links)
+ delete_actions = engine.delete_actions(wan_links)
+ engine.incorporate_actions(create_actions + delete_actions,
+ instance_action)
+
+ self.populate(instance_actions=instance_action,
+ vim_wim_actions=create_actions + delete_actions)
+
+ self.thread.insert_pending_tasks(create_actions)
+
+ assert self.thread.pending_tasks[0].is_scheduled
+
+ # When we insert the equivalent DELETE task
+ self.thread.insert_pending_tasks(delete_actions)
+
+ # Then the CREATE task should be superseded
+ self.assertEqual(self.thread.pending_tasks[0].action, 'CREATE')
+ assert self.thread.pending_tasks[0].is_superseded
+
+ self.thread.process_list('pending')
+ self.thread.process_list('refresh')
+ self.assertFalse(self.thread.pending_tasks)
+
+
+@ignore_connector
+class TestWimThread(unittest.TestCase):
+ def setUp(self):
+ wim = eg.wim(0)
+ account = eg.wim_account(0, 0)
+ account['wim'] = wim
+ self.persist = MagicMock()
+ self.thread = WimThread(self.persist, {}, account)
+ self.thread.connector = MagicMock()
+
+ super(TestWimThread, self).setUp()
+
+ def test_process_refresh(self):
+ # Given we have 30 tasks in the refresh queue
+ kwargs = {'action_id': uuid('action0')}
+ actions = eg.wim_actions('FIND', 'DONE', num_links=30, **kwargs)
+ self.thread.insert_pending_tasks(actions)
+
+ # When we process the refresh list
+ processed = self.thread.process_list('refresh')
+
+ # Then we should have REFRESH_BATCH updates
+ self.assertEqual(processed, self.thread.BATCH)
+
+ def test_process_refresh__with_superseded(self):
+ # Given we have 30 tasks but 15 of them are superseded
+ kwargs = {'action_id': uuid('action0')}
+ actions = eg.wim_actions('FIND', 'DONE', num_links=30, **kwargs)
+ self.thread.insert_pending_tasks(actions)
+ for task in self.thread.refresh_tasks[0:30:2]:
+ task.status = 'SUPERSEDED'
+
+ now = time()
+
+ # When we call the refresh_elements
+ processed = self.thread.process_list('refresh')
+
+ # Then we should have 25 updates (since SUPERSEDED updates are cheap,
+ # they are not counted for the limits)
+ self.assertEqual(processed, 25)
+
+ # The SUPERSEDED tasks should be removed, 5 tasks should be untouched,
+ # and 10 tasks should be rescheduled
+ refresh_tasks = self.thread.refresh_tasks
+ old = [t for t in refresh_tasks if t.process_at <= now]
+ new = [t for t in refresh_tasks if t.process_at > now]
+ self.assertEqual(len(old), 5)
+ self.assertEqual(len(new), 10)
+ self.assertEqual(len(self.thread.refresh_tasks), 15)
+
+
+if __name__ == '__main__':
+ unittest.main()
projects / osm / RO.git / blobdiff