conductor/conductor/tests/unit/data/plugins/inventory_provider/test_generator.py

   1 #
   2 # -------------------------------------------------------------------------
   3 #   Copyright (C) 2020 Wipro Limited.
   4 #
   5 #   Licensed under the Apache License, Version 2.0 (the "License");
   6 #   you may not use this file except in compliance with the License.
   7 #   You may obtain a copy of the License at
   8 #
   9 #       http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 #   Unless required by applicable law or agreed to in writing, software
  12 #   distributed under the License is distributed on an "AS IS" BASIS,
  13 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 #   See the License for the specific language governing permissions and
  15 #   limitations under the License.
  16 #
  17 # -------------------------------------------------------------------------
  18 #
  19
  20 import unittest
  21 import json
  22 import mock
  23 from mock import patch
  24 from conductor.data.plugins.inventory_provider.cps import CPS
  25
  26 from conductor.data.plugins.inventory_provider.generator import Generator
  27
  28
  29 class TestGenerator(unittest.TestCase):
  30
  31     def setUp(self):
  32         pass
  33
  34     def tearDown(self):
  35         patch.stopall()
  36
  37     def test_generate_candidate(self):
  38
  39         candidates_file = './conductor/tests/unit/data/plugins/inventory_provider/generated_candidates.json'
  40         expected_candidates = json.loads(open(candidates_file).read())
  41         coverage_area_file='./conductor/tests/unit/data/plugins/inventory_provider/coverage_area.json'
  42         coverage_area_json = json.loads(open(coverage_area_file).read())
  43         generator = Generator()
  44
  45         filtering_attributes = {
  46                                   'service_profile': {
  47                                         'latency': {'value': 30, 'operator': 'lte'},
  48                                         'reliability': {'value': 99.99, 'operator': 'gte'},
  49                                         'coverage_area': {'value': 'City: Chennai-1,2,3', 'operator': 'eq'}
  50                                   },
  51                                   'subnets': {
  52                                      'core': {'latency': {'min': 15, 'max': 30, 'steps': 1},
  53                                               'reliability': {'values': [99.99]}},
  54                                      'ran': {'latency': {'min': 10, 'max': 30, 'steps': 1},
  55                                              'reliability': {'values': [99.99]},
  56                                              'coverage_area_ta_list': {"derive_from":{"method":"get_tracking_area",
  57                                                                                 "args": {
  58                                                                          "coverage_area": "Chennai-1,2,3"
  59                                                                              }}}}
  60                                   }
  61                                 }
  62         self.mock_get_coverage_area_ta = mock.patch.object(CPS, 'get_coveragearea_ta',
  63                                                          return_value=coverage_area_json)
  64
  65         self.mock_get_coverage_area_ta.start()
  66
  67         self.maxDiff=None
  68
  69         generated_candidates = generator.generate_candidates('slice_profiles', filtering_attributes,
  70                                                              candidate_uniqueness='true',
  71                                                              default_fields={"creation_cost": 0.9})
  72
  73         for candidate in generated_candidates:
  74             self.assertIsNotNone(candidate['candidate_id'])
  75             del candidate['candidate_id']
  76
  77         self.assertEqual(expected_candidates, generated_candidates)
  78
  79         self.assertEqual([], generator.generate_candidates('cloud', filtering_attributes,
  80                                                            candidate_uniqueness='true',
  81                                                            default_fields={"creation_cost": 0.9}))
  82
  83     def test_resolve_demands(self):
  84         demands_file = './conductor/tests/unit/data/plugins/inventory_provider/gen_demand_list.json'
  85         demands = json.loads(open(demands_file).read())
  86         coverage_area_file = './conductor/tests/unit/data/plugins/inventory_provider/coverage_area.json'
  87         coverage_area_json = json.loads(open(coverage_area_file).read())
  88
  89         resolved_demands_file = './conductor/tests/unit/data/plugins/inventory_provider/resolved_demands_gen.json'
  90         expected_resolved_demands = json.loads(open(resolved_demands_file).read())
  91
  92         generator = Generator()
  93         self.mock_get_coverage_area_ta = mock.patch.object(CPS, 'get_coveragearea_ta',
  94                                                          return_value=coverage_area_json)
  95
  96         self.mock_get_coverage_area_ta.start()
  97
  98         self.maxDiff=None
  99         resolved_demands = generator.resolve_demands(demands, plan_info=None, triage_translator_data=None)
 100         for _, candidate_list in resolved_demands.items():
 101             for candidate in candidate_list:
 102                 self.assertIsNotNone(candidate['candidate_id'])
 103                 del candidate['candidate_id']
 104
 105         self.assertEqual(expected_resolved_demands, resolved_demands)