1 # -------------------------------------------------------------------------
\r
2 # Copyright (C) 2020 Wipro Limited.
\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
8 # http://www.apache.org/licenses/LICENSE-2.0
\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
16 # -------------------------------------------------------------------------
\r
21 from requests import RequestException, Response
\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
30 from osdf.logging.osdf_logging import error_log, debug_log
\r
31 from osdf.adapters.policy.interface import get_policies
\r
34 class TestRemoteOptProcessor(unittest.TestCase):
\r
36 self.config_spec = {
\r
37 "deployment": "config/osdf_config.yaml",
\r
38 "core": "config/common_config.yaml"
\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
50 def test_process_nsi_selection_opt(self):
\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
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
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
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
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
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
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
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
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
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
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
128 def test_process_nssi_selection_opt(self):
\r
130 request_file = main_dir + 'test/apps/slice_selection/nssi_selection_request.json'
\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
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
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
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
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
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
164 if __name__ == "__main__":
\r