test/apps/slice_selection/test_remote_opt_processor.py

   1 # -------------------------------------------------------------------------
   2 #   Copyright (C) 2020 Wipro Limited.
   3 #
   4 #   Licensed under the Apache License, Version 2.0 (the "License");
   5 #   you may not use this file except in compliance with the License.
   6 #   You may obtain a copy of the License at
   7 #
   8 #       http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 #   Unless required by applicable law or agreed to in writing, software
  11 #   distributed under the License is distributed on an "AS IS" BASIS,
  12 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 #   See the License for the specific language governing permissions and
  14 #   limitations under the License.
  15 #
  16 # -------------------------------------------------------------------------
  17 #
  18
  19 import json
  20 import unittest
  21 from requests import RequestException
  22
  23 from apps.slice_selection.optimizers.conductor.remote_opt_processor import process_nsi_selection_opt
  24 from osdf.adapters.local_data import local_policies
  25 from osdf.utils.interfaces import json_from_file, yaml_from_file
  26 from osdf.utils.programming_utils import DotDict
  27 import osdf.config.loader as config_loader
  28 from mock import patch, MagicMock
  29 import json
  30 from osdf.logging.osdf_logging import error_log, debug_log
  31 from osdf.adapters.policy.interface import get_policies
  32
  33
  34 class TestRemoteOptProcessor(unittest.TestCase):
  35     def setUp(self):
  36         self.config_spec = {
  37             "deployment": "config/osdf_config.yaml",
  38             "core": "config/common_config.yaml"
  39         }
  40         self.osdf_config = DotDict(config_loader.all_configs(**self.config_spec))
  41
  42     def tearDown(self):
  43         patch.stopall()
  44
  45     def test_process_nsi_selection_opt(self):
  46         main_dir = ""
  47         request_file = main_dir + 'test/apps/slice_selection/nsi_request.json'
  48         new_solution_response_file = main_dir + 'test/apps/slice_selection/new_solution_nsi_response.json'
  49         shared_solution_response_file = main_dir + 'test/apps/slice_selection/shared_solution_nsi_response.json'
  50         error_response_file = main_dir + 'test/apps/slice_selection/nsi_error_response.json'
  51
  52         request_json = json_from_file(request_file)
  53         new_solution_response_json = json_from_file(new_solution_response_file)
  54         shared_solution_response_json = json_from_file(shared_solution_response_file)
  55         error_response_json = json_from_file(error_response_file)
  56
  57         policies_path = main_dir + 'test/policy-local-files'
  58         slice_policies_file = main_dir + 'test/apps/slice_selection/slice_policies.txt'
  59
  60         valid_policies_files = local_policies.get_policy_names_from_file(slice_policies_file)
  61         policies = [json_from_file(policies_path + '/' + name) for name in valid_policies_files]
  62         self.patcher_get_policies = patch('osdf.adapters.policy.interface.remote_api',
  63                                           return_value=policies)
  64         self.Mock_get_policies = self.patcher_get_policies.start()
  65
  66         new_solution_conductor_response_file = 'test/apps/slice_selection/new_solution_conductor_response.json'
  67         new_solution_conductor_response = json_from_file(new_solution_conductor_response_file)
  68         self.patcher_req = patch('osdf.adapters.conductor.conductor.request',
  69                                  return_value=new_solution_conductor_response)
  70         self.Mock_req = self.patcher_req.start()
  71         self.assertEquals(new_solution_response_json, process_nsi_selection_opt(request_json, self.osdf_config))
  72         self.patcher_req.stop()
  73
  74         shared_solution_conductor_response_file = 'test/apps/slice_selection/shared_solution_conductor_response.json'
  75         shared_solution_conductor_response = json_from_file(shared_solution_conductor_response_file)
  76         self.patcher_req = patch('osdf.adapters.conductor.conductor.request',
  77                                  return_value=shared_solution_conductor_response)
  78         self.Mock_req = self.patcher_req.start()
  79         self.assertEquals(shared_solution_response_json,
  80                           process_nsi_selection_opt(request_json, self.osdf_config))
  81         self.patcher_req.stop()
  82
  83         conductor_error_response_file = 'test/apps/slice_selection/conductor_error_response.json'
  84         conductor_error_response = json_from_file(conductor_error_response_file)
  85
  86         self.patcher_req = patch('osdf.adapters.conductor.conductor.request',
  87                                  side_effect=RequestException(response=json.dumps(conductor_error_response)))
  88         self.Mock_req = self.patcher_req.start()
  89         self.assertEquals(error_response_json, process_nsi_selection_opt(request_json, self.osdf_config))
  90         self.patcher_req.stop()
  91
  92         self.patcher_req = patch('osdf.adapters.conductor.conductor.request',
  93                                  side_effect=Exception("test_exception"))
  94         self.Mock_req = self.patcher_req.start()
  95         self.assertEquals('test_exception',
  96                           process_nsi_selection_opt(request_json, self.osdf_config).get('statusMessage'))
  97         self.patcher_req.stop()
  98
  99
 100 if __name__ == "__main__":
 101     unittest.main()
 102