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 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
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 self.osdf_config = DotDict(config_loader.all_configs(**self.config_spec))
\r
45 def test_process_nsi_selection_opt(self):
\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
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
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
64 policies_path = main_dir + 'test/policy-local-files'
\r
65 slice_policies_file = main_dir + 'test/apps/slice_selection/slice_policies.txt'
\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
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
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
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
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
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
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
121 if __name__ == "__main__":
\r