import unittest\r
from requests import RequestException, Response\r
\r
-from apps.slice_selection.optimizers.conductor.remote_opt_processor import process_nsi_selection_opt\r
+from apps.slice_selection.optimizers.conductor.remote_opt_processor import SliceSelectionOptimizer\r
from osdf.adapters.local_data import local_policies\r
from osdf.utils.interfaces import json_from_file, yaml_from_file\r
from osdf.utils.programming_utils import DotDict\r
"deployment": "config/osdf_config.yaml",\r
"core": "config/common_config.yaml"\r
}\r
+ slice_spec = "config/slicing_config.yaml"\r
+ self.slice_config = config_loader.load_config_file(slice_spec)\r
self.osdf_config = DotDict(config_loader.all_configs(**self.config_spec))\r
+ self.patcher_RestClient = patch(\r
+ 'osdf.utils.interfaces.RestClient.request', return_value=MagicMock())\r
+ self.mock_rc = self.patcher_RestClient.start()\r
\r
def tearDown(self):\r
patch.stopall()\r
\r
def test_process_nsi_selection_opt(self):\r
main_dir = ""\r
- request_file = main_dir + 'test/apps/slice_selection/nsi_request.json'\r
+ request_file = main_dir + 'test/apps/slice_selection/nsi_selection_request.json'\r
not_shared_request_file = main_dir + 'test/apps/slice_selection/not_shared_nsi_request.json'\r
#response files\r
new_solution_response_file = main_dir + 'test/apps/slice_selection/new_solution_nsi_response.json'\r
shared_solution_response_file = main_dir + 'test/apps/slice_selection/shared_solution_nsi_response.json'\r
no_solution_response_file = main_dir + 'test/apps/slice_selection/no_recomm_nsi_response.json'\r
- not_shared_response_file = main_dir + 'test/apps/slice_selection/not_shared_nsi_response.json'\r
error_response_file = main_dir + 'test/apps/slice_selection/nsi_error_response.json'\r
\r
- not_shared_request_json = json_from_file(not_shared_request_file)\r
- not_shared_response_json = json_from_file(not_shared_response_file)\r
request_json = json_from_file(request_file)\r
new_solution_response_json = json_from_file(new_solution_response_file)\r
shared_solution_response_json = json_from_file(shared_solution_response_file)\r
no_solution_response_json = json_from_file(no_solution_response_file)\r
error_response_json = json_from_file(error_response_file)\r
\r
- policies_path = main_dir + 'test/policy-local-files'\r
+ policies_path = main_dir + 'test/policy-local-files/slice-selection-files'\r
slice_policies_file = main_dir + 'test/apps/slice_selection/slice_policies.txt'\r
\r
valid_policies_files = local_policies.get_policy_names_from_file(slice_policies_file)\r
self.patcher_get_policies = patch('osdf.adapters.policy.interface.remote_api',\r
return_value=policies)\r
self.Mock_get_policies = self.patcher_get_policies.start()\r
+\r
# new solution\r
new_solution_conductor_response_file = 'test/apps/slice_selection/new_solution_conductor_response.json'\r
new_solution_conductor_response = json_from_file(new_solution_conductor_response_file)\r
self.patcher_req = patch('osdf.adapters.conductor.conductor.request',\r
return_value=new_solution_conductor_response)\r
self.Mock_req = self.patcher_req.start()\r
- self.assertEquals(new_solution_response_json, process_nsi_selection_opt(request_json, self.osdf_config))\r
+ slice_select_opt = SliceSelectionOptimizer(self.osdf_config, self.slice_config, request_json, 'NSI')\r
+ slice_select_opt.process_slice_selection_opt()\r
+ self.mock_rc.assert_called_with(json=new_solution_response_json, noresponse=True)\r
self.patcher_req.stop()\r
+\r
# shared solution\r
+ request_json['preferReuse'] = True\r
shared_solution_conductor_response_file = 'test/apps/slice_selection/shared_solution_conductor_response.json'\r
shared_solution_conductor_response = json_from_file(shared_solution_conductor_response_file)\r
self.patcher_req = patch('osdf.adapters.conductor.conductor.request',\r
return_value=shared_solution_conductor_response)\r
self.Mock_req = self.patcher_req.start()\r
- self.assertEquals(shared_solution_response_json,\r
- process_nsi_selection_opt(request_json, self.osdf_config))\r
+ slice_select_opt = SliceSelectionOptimizer(self.osdf_config, self.slice_config, request_json, 'NSI')\r
+ slice_select_opt.process_slice_selection_opt()\r
+ self.mock_rc.assert_called_with(json=shared_solution_response_json, noresponse=True)\r
self.patcher_req.stop()\r
- # not-shared solution\r
- self.assertEquals(not_shared_response_json,\r
- process_nsi_selection_opt(not_shared_request_json, self.osdf_config))\r
+\r
# no recommendation\r
no_solution_conductor_response_file = 'test/apps/slice_selection/no_rec.json'\r
no_solution_conductor_response = json_from_file(no_solution_conductor_response_file)\r
self.patcher_req = patch('osdf.adapters.conductor.conductor.request',\r
return_value=no_solution_conductor_response)\r
self.Mock_req = self.patcher_req.start()\r
- self.assertEquals(no_solution_response_json,\r
- process_nsi_selection_opt(request_json, self.osdf_config))\r
+ slice_select_opt.process_slice_selection_opt()\r
+ self.mock_rc.assert_called_with(json=no_solution_response_json, noresponse=True)\r
self.patcher_req.stop()\r
\r
+ # Exception\r
conductor_error_response_file = 'test/apps/slice_selection/conductor_error_response.json'\r
conductor_error_response = json_from_file(conductor_error_response_file)\r
\r
self.patcher_req = patch('osdf.adapters.conductor.conductor.request',\r
side_effect=RequestException(response=response))\r
self.Mock_req = self.patcher_req.start()\r
- self.assertEquals(error_response_json, process_nsi_selection_opt(request_json, self.osdf_config))\r
+ slice_select_opt.process_slice_selection_opt()\r
+ self.mock_rc.assert_called_with(json=error_response_json, noresponse=True)\r
+ self.patcher_req.stop()\r
+\r
+ self.patcher_req = patch('osdf.adapters.conductor.conductor.request',\r
+ side_effect=Exception("Some error message"))\r
+ self.Mock_req = self.patcher_req.start()\r
+ slice_select_opt.process_slice_selection_opt()\r
+ self.mock_rc.assert_called_with(json=error_response_json, noresponse=True)\r
self.patcher_req.stop()\r
\r
+ def test_process_nssi_selection_opt(self):\r
+ main_dir = ""\r
+ request_file = main_dir + 'test/apps/slice_selection/nssi_selection_request.json'\r
+ # response files\r
+ shared_solution_response_file = main_dir + 'test/apps/slice_selection/shared_solution_nssi_response.json'\r
+ error_response_file = main_dir + 'test/apps/slice_selection/nssi_error_response.json'\r
+\r
+ request_json = json_from_file(request_file)\r
+ shared_solution_response_json = json_from_file(shared_solution_response_file)\r
+ error_response_json = json_from_file(error_response_file)\r
+\r
+ policies_path = main_dir + 'test/policy-local-files/slice-selection-files'\r
+ slice_policies_file = main_dir + 'test/apps/slice_selection/subnet_policies.txt'\r
+\r
+ valid_policies_files = local_policies.get_policy_names_from_file(slice_policies_file)\r
+ policies = [json_from_file(policies_path + '/' + name) for name in valid_policies_files]\r
+ self.patcher_get_policies = patch('osdf.adapters.policy.interface.remote_api',\r
+ return_value=policies)\r
+ self.Mock_get_policies = self.patcher_get_policies.start()\r
+\r
+ shared_solution_conductor_response_file = 'test/apps/slice_selection/nssi_conductor_response.json'\r
+ shared_solution_conductor_response = json_from_file(shared_solution_conductor_response_file)\r
self.patcher_req = patch('osdf.adapters.conductor.conductor.request',\r
- side_effect=Exception("test_exception"))\r
+ return_value=shared_solution_conductor_response)\r
self.Mock_req = self.patcher_req.start()\r
- self.assertEquals('test_exception',\r
- process_nsi_selection_opt(request_json, self.osdf_config).get('statusMessage'))\r
+ slice_select_opt = SliceSelectionOptimizer(self.osdf_config, self.slice_config, request_json, 'NSSI')\r
+ slice_select_opt.process_slice_selection_opt()\r
+ self.mock_rc.assert_called_with(json=shared_solution_response_json, noresponse=True)\r
self.patcher_req.stop()\r
\r
+ request_json['sliceProfile']['resourceSharingLevel'] = "not-shared"\r
+ slice_select_opt = SliceSelectionOptimizer(self.osdf_config, self.slice_config, request_json, 'NSSI')\r
+ slice_select_opt.process_slice_selection_opt()\r
+ self.mock_rc.assert_called_with(json=error_response_json, noresponse=True)\r
+\r
\r
if __name__ == "__main__":\r
unittest.main()\r
-\r
Code Review / optf / osdf.git / blobdiff