test/apps/slice_selection/test_remote_opt_processor.py

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