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 SliceSelectionOptimizer\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         slice_spec = "config/slicing_config.yaml"\r
  41         self.slice_config = config_loader.load_config_file(slice_spec)\r
  42         self.osdf_config = DotDict(config_loader.all_configs(**self.config_spec))\r
  43         self.patcher_RestClient = patch(\r
  44             'osdf.utils.interfaces.RestClient.request', return_value=MagicMock())\r
  45         self.mock_rc = self.patcher_RestClient.start()\r
  46 \r
  47     def tearDown(self):\r
  48         patch.stopall()\r
  49 \r
  50     def test_process_nsi_selection_opt(self):\r
  51         main_dir = ""\r
  52         request_file = main_dir + 'test/apps/slice_selection/nsi_selection_request.json'\r
  53         not_shared_request_file = main_dir + 'test/apps/slice_selection/not_shared_nsi_request.json'\r
  54         #response files\r
  55         new_solution_response_file = main_dir + 'test/apps/slice_selection/new_solution_nsi_response.json'\r
  56         shared_solution_response_file = main_dir + 'test/apps/slice_selection/shared_solution_nsi_response.json'\r
  57         no_solution_response_file = main_dir + 'test/apps/slice_selection/no_recomm_nsi_response.json'\r
  58         error_response_file = main_dir + 'test/apps/slice_selection/nsi_error_response.json'\r
  59 \r
  60         request_json = json_from_file(request_file)\r
  61         new_solution_response_json = json_from_file(new_solution_response_file)\r
  62         shared_solution_response_json = json_from_file(shared_solution_response_file)\r
  63         no_solution_response_json = json_from_file(no_solution_response_file)\r
  64         error_response_json = json_from_file(error_response_file)\r
  65 \r
  66         policies_path = main_dir + 'test/policy-local-files/slice-selection-files'\r
  67         slice_policies_file = main_dir + 'test/apps/slice_selection/slice_policies.txt'\r
  68 \r
  69         valid_policies_files = local_policies.get_policy_names_from_file(slice_policies_file)\r
  70         policies = [json_from_file(policies_path + '/' + name) for name in valid_policies_files]\r
  71         self.patcher_get_policies = patch('osdf.adapters.policy.interface.remote_api',\r
  72                                           return_value=policies)\r
  73         self.Mock_get_policies = self.patcher_get_policies.start()\r
  74 \r
  75         # new solution\r
  76         new_solution_conductor_response_file = 'test/apps/slice_selection/new_solution_conductor_response.json'\r
  77         new_solution_conductor_response = json_from_file(new_solution_conductor_response_file)\r
  78         self.patcher_req = patch('osdf.adapters.conductor.conductor.request',\r
  79                                  return_value=new_solution_conductor_response)\r
  80         self.Mock_req = self.patcher_req.start()\r
  81         slice_select_opt = SliceSelectionOptimizer(self.osdf_config, self.slice_config, request_json, 'NSI')\r
  82         slice_select_opt.process_slice_selection_opt()\r
  83         self.mock_rc.assert_called_with(json=new_solution_response_json, noresponse=True)\r
  84         self.patcher_req.stop()\r
  85 \r
  86         # shared solution\r
  87         request_json['preferReuse'] = True\r
  88         shared_solution_conductor_response_file = 'test/apps/slice_selection/shared_solution_conductor_response.json'\r
  89         shared_solution_conductor_response = json_from_file(shared_solution_conductor_response_file)\r
  90         self.patcher_req = patch('osdf.adapters.conductor.conductor.request',\r
  91                                  return_value=shared_solution_conductor_response)\r
  92         self.Mock_req = self.patcher_req.start()\r
  93         slice_select_opt = SliceSelectionOptimizer(self.osdf_config, self.slice_config, request_json, 'NSI')\r
  94         slice_select_opt.process_slice_selection_opt()\r
  95         self.mock_rc.assert_called_with(json=shared_solution_response_json, noresponse=True)\r
  96         self.patcher_req.stop()\r
  97 \r
  98         # no recommendation\r
  99         no_solution_conductor_response_file = 'test/apps/slice_selection/no_rec.json'\r
 100         no_solution_conductor_response = json_from_file(no_solution_conductor_response_file)\r
 101         self.patcher_req = patch('osdf.adapters.conductor.conductor.request',\r
 102                                  return_value=no_solution_conductor_response)\r
 103         self.Mock_req = self.patcher_req.start()\r
 104         slice_select_opt.process_slice_selection_opt()\r
 105         self.mock_rc.assert_called_with(json=no_solution_response_json, noresponse=True)\r
 106         self.patcher_req.stop()\r
 107 \r
 108         # Exception\r
 109         conductor_error_response_file = 'test/apps/slice_selection/conductor_error_response.json'\r
 110         conductor_error_response = json_from_file(conductor_error_response_file)\r
 111 \r
 112         response = Response()\r
 113         response._content = json.dumps(conductor_error_response).encode()\r
 114         self.patcher_req = patch('osdf.adapters.conductor.conductor.request',\r
 115                                  side_effect=RequestException(response=response))\r
 116         self.Mock_req = self.patcher_req.start()\r
 117         slice_select_opt.process_slice_selection_opt()\r
 118         self.mock_rc.assert_called_with(json=error_response_json, noresponse=True)\r
 119         self.patcher_req.stop()\r
 120 \r
 121         self.patcher_req = patch('osdf.adapters.conductor.conductor.request',\r
 122                                  side_effect=Exception("Some error message"))\r
 123         self.Mock_req = self.patcher_req.start()\r
 124         slice_select_opt.process_slice_selection_opt()\r
 125         self.mock_rc.assert_called_with(json=error_response_json, noresponse=True)\r
 126         self.patcher_req.stop()\r
 127 \r
 128     def test_process_nssi_selection_opt(self):\r
 129         main_dir = ""\r
 130         request_file = main_dir + 'test/apps/slice_selection/nssi_selection_request.json'\r
 131         # response files\r
 132         shared_solution_response_file = main_dir + 'test/apps/slice_selection/shared_solution_nssi_response.json'\r
 133         error_response_file = main_dir + 'test/apps/slice_selection/nssi_error_response.json'\r
 134 \r
 135         request_json = json_from_file(request_file)\r
 136         shared_solution_response_json = json_from_file(shared_solution_response_file)\r
 137         error_response_json = json_from_file(error_response_file)\r
 138 \r
 139         policies_path = main_dir + 'test/policy-local-files/slice-selection-files'\r
 140         slice_policies_file = main_dir + 'test/apps/slice_selection/subnet_policies.txt'\r
 141 \r
 142         valid_policies_files = local_policies.get_policy_names_from_file(slice_policies_file)\r
 143         policies = [json_from_file(policies_path + '/' + name) for name in valid_policies_files]\r
 144         self.patcher_get_policies = patch('osdf.adapters.policy.interface.remote_api',\r
 145                                           return_value=policies)\r
 146         self.Mock_get_policies = self.patcher_get_policies.start()\r
 147 \r
 148         shared_solution_conductor_response_file = 'test/apps/slice_selection/nssi_conductor_response.json'\r
 149         shared_solution_conductor_response = json_from_file(shared_solution_conductor_response_file)\r
 150         self.patcher_req = patch('osdf.adapters.conductor.conductor.request',\r
 151                                  return_value=shared_solution_conductor_response)\r
 152         self.Mock_req = self.patcher_req.start()\r
 153         slice_select_opt = SliceSelectionOptimizer(self.osdf_config, self.slice_config, request_json, 'NSSI')\r
 154         slice_select_opt.process_slice_selection_opt()\r
 155         self.mock_rc.assert_called_with(json=shared_solution_response_json, noresponse=True)\r
 156         self.patcher_req.stop()\r
 157 \r
 158         request_json['sliceProfile']['resourceSharingLevel'] = "not-shared"\r
 159         slice_select_opt = SliceSelectionOptimizer(self.osdf_config, self.slice_config, request_json, 'NSSI')\r
 160         slice_select_opt.process_slice_selection_opt()\r
 161         self.mock_rc.assert_called_with(json=error_response_json, noresponse=True)\r
 162 \r
 163 \r
 164 if __name__ == "__main__":\r
 165     unittest.main()\r