X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_ro%2Fwim%2Ftests%2Ftest_engine.py;fp=osm_ro%2Fwim%2Ftests%2Ftest_engine.py;h=0000000000000000000000000000000000000000;hb=7d782eff123e5b44d41437377ccca66ad1e8b21b;hp=9bb7bca00e10ad83416b3bd4b077dc72044a5bdc;hpb=5db670b68349fd1f00a5efc8c0ccd0ef9d073dca;p=osm%2FRO.git diff --git a/osm_ro/wim/tests/test_engine.py b/osm_ro/wim/tests/test_engine.py deleted file mode 100644 index 9bb7bca0..00000000 --- a/osm_ro/wim/tests/test_engine.py +++ /dev/null @@ -1,179 +0,0 @@ -# -*- 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: -# -# 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. -## - -from __future__ import unicode_literals - -import unittest - -from mock import MagicMock - -from . import fixtures as eg -from ...tests.db_helpers import TestCaseWithDatabasePerTest, uuid -from ..errors import NoWimConnectedToDatacenters -from ..engine import WimEngine -from ..persistence import WimPersistence - - -class TestWimEngineDbMethods(TestCaseWithDatabasePerTest): - def setUp(self): - super(TestWimEngineDbMethods, self).setUp() - self.persist = WimPersistence(self.db) - self.engine = WimEngine(persistence=self.persist) - self.addCleanup(self.engine.stop_threads) - - def populate(self, seeds=None): - super(TestWimEngineDbMethods, self).populate( - seeds or eg.consistent_set()) - - def test_find_common_wims(self): - # Given we have 2 WIM, 3 datacenters, but just 1 of the WIMs have - # access to them - self.populate([{'nfvo_tenants': [eg.tenant(0)]}] + - eg.wim_set(0, 0) + - eg.wim_set(1, 0) + - eg.datacenter_set(0, 0) + - eg.datacenter_set(1, 0) + - eg.datacenter_set(2, 0) + - [{'wim_port_mappings': [ - eg.wim_port_mapping(0, 0), - eg.wim_port_mapping(0, 1), - eg.wim_port_mapping(0, 2)]}]) - - # When we retrieve the wims interconnecting some datacenters - wim_ids = self.engine.find_common_wims( - [uuid('dc0'), uuid('dc1'), uuid('dc2')], tenant='tenant0') - - # Then we should have just the first wim - self.assertEqual(len(wim_ids), 1) - self.assertEqual(wim_ids[0], uuid('wim0')) - - def test_find_common_wims_multiple(self): - # Given we have 2 WIM, 3 datacenters, and all the WIMs have access to - # all datacenters - self.populate([{'nfvo_tenants': [eg.tenant(0)]}] + - eg.wim_set(0, 0) + - eg.wim_set(1, 0) + - eg.datacenter_set(0, 0) + - eg.datacenter_set(1, 0) + - eg.datacenter_set(2, 0) + - [{'wim_port_mappings': [ - eg.wim_port_mapping(0, 0), - eg.wim_port_mapping(0, 1), - eg.wim_port_mapping(0, 2), - eg.wim_port_mapping(1, 0), - eg.wim_port_mapping(1, 1), - eg.wim_port_mapping(1, 2)]}]) - - # When we retrieve the wims interconnecting tree datacenters - wim_ids = self.engine.find_common_wims( - [uuid('dc0'), uuid('dc1'), uuid('dc2')], tenant='tenant0') - - # Then we should have all the wims - self.assertEqual(len(wim_ids), 2) - self.assertItemsEqual(wim_ids, [uuid('wim0'), uuid('wim1')]) - - def test_find_common_wim(self): - # Given we have 1 WIM, 3 datacenters but the WIM have access to just 2 - # of them - self.populate([{'nfvo_tenants': [eg.tenant(0)]}] + - eg.wim_set(0, 0) + - eg.datacenter_set(0, 0) + - eg.datacenter_set(1, 0) + - eg.datacenter_set(2, 0) + - [{'wim_port_mappings': [ - eg.wim_port_mapping(0, 0), - eg.wim_port_mapping(0, 1)]}]) - - # When we retrieve the common wim for the 2 datacenter that are - # interconnected - wim_id = self.engine.find_common_wim( - [uuid('dc0'), uuid('dc1')], tenant='tenant0') - - # Then we should find the wim - self.assertEqual(wim_id, uuid('wim0')) - - # When we try to retrieve the common wim for the all the datacenters - # Then a NoWimConnectedToDatacenters exception should be raised - with self.assertRaises(NoWimConnectedToDatacenters): - self.engine.find_common_wim( - [uuid('dc0'), uuid('dc1'), uuid('dc2')], tenant='tenant0') - - def test_find_common_wim__different_tenants(self): - # Given we have 1 WIM and 2 datacenters connected but the WIMs don't - # belong to the tenant we have access to... - self.populate([{'nfvo_tenants': [eg.tenant(0), eg.tenant(1)]}] + - eg.wim_set(0, 0) + - eg.datacenter_set(0, 0) + - eg.datacenter_set(1, 0) + - [{'wim_port_mappings': [ - eg.wim_port_mapping(0, 0), - eg.wim_port_mapping(0, 1)]}]) - - # When we retrieve the common wim for the 2 datacenter that are - # interconnected, but using another tenant, - # Then we should get an exception - with self.assertRaises(NoWimConnectedToDatacenters): - self.engine.find_common_wim( - [uuid('dc0'), uuid('dc1')], tenant='tenant1') - - -class TestWimEngine(unittest.TestCase): - def test_derive_wan_link(self): - # Given we have 2 datacenters connected by the same WIM, with port - # mappings registered - mappings = [eg.processed_port_mapping(0, 0), - eg.processed_port_mapping(0, 1)] - persist = MagicMock( - get_wim_port_mappings=MagicMock(return_value=mappings)) - - engine = WimEngine(persistence=persist) - self.addCleanup(engine.stop_threads) - - # When we receive a list of 4 instance nets, representing - # 2 VLDs connecting 2 datacenters each - instance_nets = eg.instance_nets(2, 2) - wan_links = engine.derive_wan_links({}, instance_nets, uuid('tenant0')) - - # Then we should derive 2 wan_links with the same instance_scenario_id - # and different scenario_network_id - self.assertEqual(len(wan_links), 2) - for link in wan_links: - self.assertEqual(link['instance_scenario_id'], uuid('nsr0')) - # Each VLD needs a network to be created in each datacenter - self.assertItemsEqual([l['sce_net_id'] for l in wan_links], - [uuid('vld0'), uuid('vld1')]) - - -if __name__ == '__main__': - unittest.main()