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 from mock import patch
  23
  24 from conductor.data.plugins.inventory_provider.generator import Generator
  25
  26
  27 class TestGenerator(unittest.TestCase):
  28
  29     def setUp(self):
  30         pass
  31
  32     def tearDown(self):
  33         patch.stopall()
  34
  35     def test_generate_candidate(self):
  36
  37         candidates_file = './conductor/tests/unit/data/plugins/inventory_provider/generated_candidates.json'
  38         expected_candidates = json.loads(open(candidates_file).read())
  39
  40         generator = Generator()
  41
  42         filtering_attributes = {
  43                                   'service_profile': {
  44                                         'latency': {'value': 30, 'operator': 'lte'},
  45                                         'reliability': {'value': 99.99, 'operator': 'gte'},
  46                                         'coverage_area_ta_list': {'value': 'City: Chennai', 'operator': 'eq'}
  47                                   },
  48                                   'subnets': {
  49                                      'core': {'latency': {'min': 15, 'max': 30, 'steps': 1},
  50                                               'reliability': {'values': [99.99]}},
  51                                      'ran': {'latency': {'min': 10, 'max': 30, 'steps': 1},
  52                                              'reliability': {'values': [99.99]},
  53                                              'coverage_area_ta_list': {'values': ['City: Chennai']}}
  54                                   }
  55                                 }
  56
  57         generated_candidates = generator.generate_candidates('slice_profiles', filtering_attributes,
  58                                                              candidate_uniqueness='true',
  59                                                              default_fields={"creation_cost": 0.9})
  60
  61         for candidate in generated_candidates:
  62             self.assertIsNotNone(candidate['candidate_id'])
  63             del candidate['candidate_id']
  64
  65         self.assertCountEqual(expected_candidates, generated_candidates)
  66
  67         self.assertEqual([], generator.generate_candidates('cloud', filtering_attributes,
  68                                                            candidate_uniqueness='true',
  69                                                            default_fields={"creation_cost": 0.9}))
  70
  71     def test_resolve_demands(self):
  72         demands_file = './conductor/tests/unit/data/plugins/inventory_provider/gen_demand_list.json'
  73         demands = json.loads(open(demands_file).read())
  74
  75         resolved_demands_file = './conductor/tests/unit/data/plugins/inventory_provider/resolved_demands_gen.json'
  76         expected_resolved_demands = json.loads(open(resolved_demands_file).read())
  77
  78         generator = Generator()
  79         resolved_demands = generator.resolve_demands(demands, plan_info=None, triage_translator_data=None)
  80         for _, candidate_list in resolved_demands.items():
  81             for candidate in candidate_list:
  82                 self.assertIsNotNone(candidate['candidate_id'])
  83                 del candidate['candidate_id']
  84
  85         self.assertEqual(expected_resolved_demands, resolved_demands)