RO/osm_ro/wim/tests/test_engine.py

   1 # -*- coding: utf-8 -*-
   2 ##
   3 # Copyright 2018 University of Bristol - High Performance Networks Research
   4 # Group
   5 # All Rights Reserved.
   6 #
   7 # Contributors: Anderson Bravalheri, Dimitrios Gkounis, Abubakar Siddique
   8 # Muqaddas, Navdeep Uniyal, Reza Nejabati and Dimitra Simeonidou
   9 #
  10 # Licensed under the Apache License, Version 2.0 (the "License"); you may
  11 # not use this file except in compliance with the License. You may obtain
  12 # a copy of the License at
  13 #
  14 #         http://www.apache.org/licenses/LICENSE-2.0
  15 #
  16 # Unless required by applicable law or agreed to in writing, software
  17 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  18 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  19 # License for the specific language governing permissions and limitations
  20 # under the License.
  21 #
  22 # For those usages not covered by the Apache License, Version 2.0 please
  23 # contact with: <highperformance-networks@bristol.ac.uk>
  24 #
  25 # Neither the name of the University of Bristol nor the names of its
  26 # contributors may be used to endorse or promote products derived from
  27 # this software without specific prior written permission.
  28 #
  29 # This work has been performed in the context of DCMS UK 5G Testbeds
  30 # & Trials Programme and in the framework of the Metro-Haul project -
  31 # funded by the European Commission under Grant number 761727 through the
  32 # Horizon 2020 and 5G-PPP programmes.
  33 ##
  34
  35 import unittest
  36
  37 from unittest.mock import MagicMock
  38
  39 from . import fixtures as eg
  40 from ...tests.db_helpers import TestCaseWithDatabasePerTest, uuid
  41 from ..errors import NoWimConnectedToDatacenters
  42 from ..engine import WimEngine
  43 from ..persistence import WimPersistence
  44
  45
  46 class TestWimEngineDbMethods(TestCaseWithDatabasePerTest):
  47     def setUp(self):
  48         super(TestWimEngineDbMethods, self).setUp()
  49         self.persist = WimPersistence(self.db)
  50         self.engine = WimEngine(persistence=self.persist)
  51         self.addCleanup(self.engine.stop_threads)
  52
  53     def populate(self, seeds=None):
  54         super(TestWimEngineDbMethods, self).populate(
  55             seeds or eg.consistent_set())
  56
  57     def test_find_common_wims(self):
  58         # Given we have 2 WIM, 3 datacenters, but just 1 of the WIMs have
  59         # access to them
  60         self.populate([{'nfvo_tenants': [eg.tenant(0)]}] +
  61                       eg.wim_set(0, 0) +
  62                       eg.wim_set(1, 0) +
  63                       eg.datacenter_set(0, 0) +
  64                       eg.datacenter_set(1, 0) +
  65                       eg.datacenter_set(2, 0) +
  66                       [{'wim_port_mappings': [
  67                           eg.wim_port_mapping(0, 0),
  68                           eg.wim_port_mapping(0, 1),
  69                           eg.wim_port_mapping(0, 2)]}])
  70
  71         # When we retrieve the wims interconnecting some datacenters
  72         wim_ids = self.engine.find_common_wims(
  73             [uuid('dc0'), uuid('dc1'), uuid('dc2')], tenant='tenant0')
  74
  75         # Then we should have just the first wim
  76         self.assertEqual(len(wim_ids), 1)
  77         self.assertEqual(wim_ids[0], uuid('wim0'))
  78
  79     def test_find_common_wims_multiple(self):
  80         # Given we have 2 WIM, 3 datacenters, and all the WIMs have access to
  81         # all datacenters
  82         self.populate([{'nfvo_tenants': [eg.tenant(0)]}] +
  83                       eg.wim_set(0, 0) +
  84                       eg.wim_set(1, 0) +
  85                       eg.datacenter_set(0, 0) +
  86                       eg.datacenter_set(1, 0) +
  87                       eg.datacenter_set(2, 0) +
  88                       [{'wim_port_mappings': [
  89                           eg.wim_port_mapping(0, 0),
  90                           eg.wim_port_mapping(0, 1),
  91                           eg.wim_port_mapping(0, 2),
  92                           eg.wim_port_mapping(1, 0),
  93                           eg.wim_port_mapping(1, 1),
  94                           eg.wim_port_mapping(1, 2)]}])
  95
  96         # When we retrieve the wims interconnecting tree datacenters
  97         wim_ids = self.engine.find_common_wims(
  98             [uuid('dc0'), uuid('dc1'), uuid('dc2')], tenant='tenant0')
  99
 100         # Then we should have all the wims
 101         self.assertEqual(len(wim_ids), 2)
 102         self.assertItemsEqual(wim_ids, [uuid('wim0'), uuid('wim1')])
 103
 104     def test_find_common_wim(self):
 105         # Given we have 1 WIM, 3 datacenters but the WIM have access to just 2
 106         # of them
 107         self.populate([{'nfvo_tenants': [eg.tenant(0)]}] +
 108                       eg.wim_set(0, 0) +
 109                       eg.datacenter_set(0, 0) +
 110                       eg.datacenter_set(1, 0) +
 111                       eg.datacenter_set(2, 0) +
 112                       [{'wim_port_mappings': [
 113                           eg.wim_port_mapping(0, 0),
 114                           eg.wim_port_mapping(0, 1)]}])
 115
 116         # When we retrieve the common wim for the 2 datacenter that are
 117         # interconnected
 118         wim_id = self.engine.find_common_wim(
 119             [uuid('dc0'), uuid('dc1')], tenant='tenant0')
 120
 121         # Then we should find the wim
 122         self.assertEqual(wim_id, uuid('wim0'))
 123
 124         # When we try to retrieve the common wim for the all the datacenters
 125         # Then a NoWimConnectedToDatacenters exception should be raised
 126         with self.assertRaises(NoWimConnectedToDatacenters):
 127             self.engine.find_common_wim(
 128                 [uuid('dc0'), uuid('dc1'), uuid('dc2')], tenant='tenant0')
 129
 130     def test_find_common_wim__different_tenants(self):
 131         # Given we have 1 WIM and 2 datacenters connected but the WIMs don't
 132         # belong to the tenant we have access to...
 133         self.populate([{'nfvo_tenants': [eg.tenant(0), eg.tenant(1)]}] +
 134                       eg.wim_set(0, 0) +
 135                       eg.datacenter_set(0, 0) +
 136                       eg.datacenter_set(1, 0) +
 137                       [{'wim_port_mappings': [
 138                           eg.wim_port_mapping(0, 0),
 139                           eg.wim_port_mapping(0, 1)]}])
 140
 141         # When we retrieve the common wim for the 2 datacenter that are
 142         # interconnected, but using another tenant,
 143         # Then we should get an exception
 144         with self.assertRaises(NoWimConnectedToDatacenters):
 145             self.engine.find_common_wim(
 146                 [uuid('dc0'), uuid('dc1')], tenant='tenant1')
 147
 148
 149 class TestWimEngine(unittest.TestCase):
 150     def test_derive_wan_link(self):
 151         # Given we have 2 datacenters connected by the same WIM, with port
 152         # mappings registered
 153         mappings = [eg.processed_port_mapping(0, 0),
 154                     eg.processed_port_mapping(0, 1)]
 155         persist = MagicMock(
 156             get_wim_port_mappings=MagicMock(return_value=mappings))
 157
 158         engine = WimEngine(persistence=persist)
 159         self.addCleanup(engine.stop_threads)
 160
 161         # When we receive a list of 4 instance nets, representing
 162         # 2 VLDs connecting 2 datacenters each
 163         instance_nets = eg.instance_nets(2, 2)
 164         wan_links = engine.derive_wan_links({}, instance_nets, uuid('tenant0'))
 165
 166         # Then we should derive 2 wan_links with the same instance_scenario_id
 167         # and different scenario_network_id
 168         self.assertEqual(len(wan_links), 2)
 169         for link in wan_links:
 170             self.assertEqual(link['instance_scenario_id'], uuid('nsr0'))
 171         # Each VLD needs a network to be created in each datacenter
 172         self.assertItemsEqual([l['sce_net_id'] for l in wan_links],
 173                               [uuid('vld0'), uuid('vld1')])
 174
 175
 176 if __name__ == '__main__':
 177     unittest.main()