conductor/conductor/tests/unit/data/plugins/inventory_provider/test_aai_utils.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
  23 from mock import patch
  24
  25 from conductor.data.plugins.inventory_provider.utils import aai_utils
  26
  27 class TestUtils(unittest.TestCase):
  28
  29     def setUp(self):
  30         pass
  31
  32     def tearDown(self):
  33         patch.stopall()
  34
  35     def test_convert_hyphen_to_under_score(self):
  36         slice_profile_file = './conductor/tests/unit/data/plugins/inventory_provider/slice_profile.json'
  37         slice_profile_hyphened = json.loads(open(slice_profile_file).read())
  38
  39         converted_slice_profile_file = './conductor/tests/unit/data/plugins/inventory_provider' \
  40                                        '/slice_profile_converted.json'
  41         converted_slice_profile = json.loads(open(converted_slice_profile_file).read())
  42         self.assertEqual(converted_slice_profile, aai_utils.convert_hyphen_to_under_score(slice_profile_hyphened))
  43
  44     def test_get_first_level_and_second_level_filter(self):
  45         first_level_filter_file = './conductor/tests/unit/data/plugins/inventory_provider/first_level_filter.json'
  46         first_level_filter = json.loads(open(first_level_filter_file).read())
  47         filtering_attributes = dict()
  48         filtering_attributes['orchestration-status'] = "active"
  49         filtering_attributes['service-role'] = "nssi"
  50         filtering_attributes['model-invariant-id'] = "123644"
  51         filtering_attributes['model-version-id'] = "524846"
  52         filtering_attributes['environment-context'] = 'shared'
  53
  54         second_level_filter = {'service-role': 'nssi'}
  55
  56         self.assertEqual(second_level_filter, aai_utils.get_first_level_and_second_level_filter(filtering_attributes,
  57                                                                                                   "service_instance"))
  58
  59         self.assertEqual(first_level_filter, filtering_attributes)
  60
  61         self.assertEqual({}, aai_utils.get_first_level_and_second_level_filter(filtering_attributes,
  62                                                                                "service_instance"))
  63
  64     def test_add_query_params(self):
  65         first_level_filter_file = './conductor/tests/unit/data/plugins/inventory_provider/first_level_filter.json'
  66         first_level_filter = json.loads(open(first_level_filter_file).read())
  67
  68         query_params = "?orchestration-status=active&model-invariant-id=123644&model-version-id=524846&" \
  69                        "environment-context=shared&"
  70
  71         self.assertEqual(query_params, aai_utils.add_query_params(first_level_filter))
  72         first_level_filter = {}
  73         self.assertEqual('', aai_utils.add_query_params(first_level_filter))
  74
  75     def test_add_query_params_and_depth(self):
  76         first_level_filter_file = './conductor/tests/unit/data/plugins/inventory_provider/first_level_filter.json'
  77         first_level_filter = json.loads(open(first_level_filter_file).read())
  78
  79         query_params_with_depth = "?orchestration-status=active&model-invariant-id=123644&model-version-id=524846&" \
  80                                   "environment-context=shared&depth=2"
  81
  82         self.assertEqual(query_params_with_depth, aai_utils.add_query_params_and_depth(first_level_filter, "2"))
  83
  84         only_depth = "?depth=2"
  85         first_level_filter = {}
  86         self.assertEqual(only_depth, aai_utils.add_query_params_and_depth(first_level_filter, "2"))
  87
  88     def test_get_inv_values_for_second_level_filter(self):
  89         nssi_response_file = './conductor/tests/unit/data/plugins/inventory_provider/nssi_response.json'
  90         nssi_response = json.loads(open(nssi_response_file).read())
  91         nssi_instance = nssi_response.get("service-instance")[0]
  92         second_level_filter = {'service-role': 'nsi'}
  93         inventory_attribute = {'service-role': 'nssi'}
  94         self.assertEqual(inventory_attribute, aai_utils.get_inv_values_for_second_level_filter(second_level_filter,
  95                                                                                                nssi_instance))