from requests import RequestException
from apps.slice_selection.optimizers.conductor.response_processor \
- import conductor_response_processor, conductor_error_response_processor
+ import conductor_response_processor, conductor_error_response_processor, solution_with_only_slice_profile, get_nsi_selection_response
from osdf.adapters.conductor import conductor
from osdf.adapters.policy.interface import get_policies
from osdf.adapters.policy.utils import group_policies_gen
overall_recommendations = dict()
nst_info_map = dict()
+ new_nsi_solutions = list()
for nst_info in request_json["NSTInfoList"]:
nst_name = nst_info["modelName"]
- nst_info_map["nst_name"] = {"NSTName": nst_name,
- "UUID": nst_info["modelVersionId"],
- "invariantUUID": nst_info["modelInvariantId"]}
+ nst_info_map[nst_name] = {"NSTName": nst_name,
+ "UUID": nst_info["modelVersionId"],
+ "invariantUUID": nst_info["modelInvariantId"]}
- policy_request_json = request_json.copy()
- policy_request_json['serviceInfo']['serviceName'] = nst_name
+ if request_json["serviceProfile"]["resourceSharingLevel"] == "non-shared":
+ new_nsi_solution = solution_with_only_slice_profile(request_json['serviceProfile'], nst_info_map.get(nst_name))
+ new_nsi_solutions.append(new_nsi_solution)
+ else:
+ policy_request_json = request_json.copy()
+ policy_request_json['serviceInfo']['serviceName'] = nst_name
+ policies = get_policies(policy_request_json, "slice_selection")
- policies = get_policies(policy_request_json, "slice_selection")
+ demands = get_slice_demands(nst_name, policies, osdf_config.core)
- demands = get_slice_demands(nst_name, policies, osdf_config.core)
-
- request_parameters = request_json.get('serviceProfile',{})
- service_info = {}
- req_info['numSolutions'] = 'all'
- resp = conductor.request(req_info, demands, request_parameters, service_info, False,
+ request_parameters = request_json.get('serviceProfile',{})
+ service_info = {}
+ req_info['numSolutions'] = 'all'
+ resp = conductor.request(req_info, demands, request_parameters, service_info, False,
osdf_config, policies)
- debug_log.debug("Response from conductor {}".format(str(resp)))
- overall_recommendations[nst_name] = resp["plans"][0].get("recommendations")
-
- return conductor_response_processor(overall_recommendations, nst_info_map, req_info)
+ if resp["plans"][0].get("status") == "not found":
+ resp["recommendations"] = list()
+ debug_log.debug("Response from conductor {}".format(str(resp)))
+ overall_recommendations[nst_name] = resp["plans"][0].get("recommendations")
+ if request_json["serviceProfile"]["resourceSharingLevel"] == "non-shared":
+ solutions = dict()
+ solutions['newNSISolutions'] = new_nsi_solutions
+ solutions['sharedNSISolutions'] = []
+ return get_nsi_selection_response(req_info, solutions)
+ else:
+ return conductor_response_processor(overall_recommendations, nst_info_map, req_info, request_json["serviceProfile"])
except Exception as ex:
error_log.error("Error for {} {}".format(req_info.get('requestId'),
traceback.format_exc()))
from osdf.logging.osdf_logging import debug_log
-SLICE_PROFILE_FIELDS = ["latency", "max_number_of_ues", "coverage_area_ta_list",
- "ue_mobility_level", "resource_sharing_level", "exp_data_rate_ul",
- "exp_data_rate_dl", "area_traffic_cap_ul", "area_traffic_cap_dl",
- "activity_factor", "e2e_latency", "jitter", "survival_time",
- "exp_data_rate", "payload_size", "traffic_density", "conn_density",
- "reliability", "service_area_dimension", "cs_availability"]
+SLICE_PROFILE_FIELDS = {"latency":"latency", "max_number_of_ues":"maxNumberOfUEs", "coverage_area_ta_list": "coverageAreaTAList",
+ "ue_mobility_level":"uEMobilityLevel", "resource_sharing_level":"resourceSharingLevel", "exp_data_rate_ul": "expDataRateUL",
+ "exp_data_rate_dl":"expDataRateDL", "area_traffic_cap_ul":"areaTrafficCapUL", "area_traffic_cap_dl": "areaTrafficCapDL",
+ "activity_factor":"activityFactor", "e2e_latency":"e2eLatency", "jitter":"jitter", "survival_time": "survivalTime",
+ "exp_data_rate":"expDataRate", "payload_size":"payloadSize", "traffic_density":"trafficDensity", "conn_density":"connDensity",
+ "reliability":"reliability", "service_area_dimension":"serviceAreaDimension", "cs_availability": "csAvailability"}
-def conductor_response_processor(overall_recommendations, nst_info_map, request_info):
+def conductor_response_processor(overall_recommendations, nst_info_map, request_info, service_profile):
"""Process conductor response to form the response for the API request
:param overall_recommendations: recommendations from conductor
:param nst_info_map: NST info from the request
"""
shared_nsi_solutions = list()
new_nsi_solutions = list()
-
for nst_name, recommendations in overall_recommendations.items():
+ if not (recommendations):
+ new_nsi_solution = solution_with_only_slice_profile(service_profile, nst_info_map.get(nst_name))
+ new_nsi_solutions.append(new_nsi_solution)
+ continue
+
for recommendation in recommendations:
nsi_set = set(values['candidate']['nsi_id'] for key, values in recommendation.items())
if len(nsi_set) == 1:
return get_nsi_selection_response(request_info, solutions)
+def solution_with_only_slice_profile(service_profile, nst_info):
+ nssi_solutions = get_slice_profile_from_service_profile(service_profile)
+ new_nsi_solution = dict()
+ new_nsi_solution['matchLevel'] = ""
+ new_nsi_solution['NSTInfo'] = nst_info
+ new_nsi_solution['NSSISolutions'] = nssi_solutions
+ return new_nsi_solution
+
def conductor_error_response_processor(request_info, error_message):
"""Form response message from the error message
:param request_info: request info
'statusMessage': error_message}
+def get_slice_profile_from_service_profile(service_profile):
+ nssi_solutions = list()
+ service_profile["domainType"] = "cn"
+ nssi_solution = {"sliceProfile": service_profile}
+ nssi_solutions.append(nssi_solution)
+ return nssi_solutions
+
+
def get_nssi_solutions(recommendation):
"""Get nssi solutions from recommendation
:param recommendation: recommendation from conductor
for field in SLICE_PROFILE_FIELDS:
if candidate[field]:
- slice_profile[field] = candidate[field]
+ slice_profile[SLICE_PROFILE_FIELDS[field]] = candidate[field]
return nssi_info, slice_profile
'requestStatus': 'completed',
'statusMessage': '',
'solutions': solutions}
+
"newNSISolutions":[
{
"matchLevel":"",
- "NSTInfo":null,
+ "NSTInfo":{"invariantUUID": "fda3c1e8-7653-4acd-80ef-f5755c1d3859",
+ "UUID": "a6906768-1cae-4e78-acd1-d753ac61f3e8",
+ "NSTName": "URLLC_1"
+ },
"NSSISolutions":[
{
"sliceProfile":{
"latency":20,
- "coverage_area_ta_list":"[{\"province\":\"??\",\"city\":\"???\",\"county\":\"???\",\"street\":\"?????\"}]",
- "ue_mobility_level":"stationary",
- "resource_sharing_level":"0",
- "exp_data_rate_ul":100,
- "exp_data_rate_dl":100
+ "coverageAreaTAList":"[{\"province\":\"??\",\"city\":\"???\",\"county\":\"???\",\"street\":\"?????\"}]",
+ "uEMobilityLevel":"stationary",
+ "resourceSharingLevel":"0",
+ "expDataRateUL":100,
+ "expDataRateDL":100
},
"NSSTInfo":{
"NSSTName":"URLLC_Core_1"
{
"sliceProfile":{
"latency":15,
- "coverage_area_ta_list":"[{\"province\":\"??\",\"city\":\"???\",\"county\":\"???\",\"street\":\"?????\"}]",
- "ue_mobility_level":"stationary",
- "resource_sharing_level":"0",
- "exp_data_rate_ul":100,
- "exp_data_rate_dl":100
+ "coverageAreaTAList":"[{\"province\":\"??\",\"city\":\"???\",\"county\":\"???\",\"street\":\"?????\"}]",
+ "uEMobilityLevel":"stationary",
+ "resourceSharingLevel":"0",
+ "expDataRateUL":100,
+ "expDataRateDL":100
},
"NSSTInfo":{
"NSSTName":"URLLC_Ran_1"
}
]
}
-}
\ No newline at end of file
+}
--- /dev/null
+{
+ "plans": [
+ {
+ "id": "d8c07237-5f66-4aa6-871c-a04221d99458",
+ "links": [
+ [
+ {
+ "href": "https://oof-has-api:8091/v1/plans/d8c07237-5f66-4aa6-871c-a04221d99458",
+ "rel": "self"
+ }
+ ]
+ ],
+ "message": "Plan d8c07237-5f66-4aa6-871c-a04221d99458 search failed, no recommendations found by machine dev-oof-has-solver-65d478b6d-ql5fp",
+ "name": "d290f1ee-6c54-4b01-90e6-d701748f0851",
+ "status": "not found"
+ }
+ ]
+}
--- /dev/null
+{\r
+ "plans":[\r
+ {\r
+ "status":"done",\r
+ "id":"plan_id",\r
+ "name":"Plan Name 1",\r
+ "links":[\r
+ [\r
+ {\r
+ "href":"http://conductor:8091/v1/plans/plan_id",\r
+ "rel":"self"\r
+ }\r
+ ]\r
+ ],\r
+ "recommendations": []\r
+ }\r
+ ]\r
+}\r
--- /dev/null
+{\r
+ "requestId":"d290f1ee-6c54-4b01-90e6-d701748f0851",\r
+ "transactionId":"d290f1ee-6c54-4b01-90e6-d701748f0851",\r
+ "requestStatus":"completed",\r
+ "statusMessage":"",\r
+ "solutions":{\r
+ "sharedNSISolutions":[\r
+\r
+ ],\r
+ "newNSISolutions":[\r
+ {\r
+ "matchLevel":"",\r
+ "NSTInfo":{"invariantUUID": "fda3c1e8-7653-4acd-80ef-f5755c1d3859",\r
+ "UUID": "a6906768-1cae-4e78-acd1-d753ac61f3e8",\r
+ "NSTName": "URLLC_1"\r
+ },\r
+\r
+ "NSSISolutions":[\r
+ {\r
+ "sliceProfile": {\r
+ "latency": 2,\r
+ "security": "High",\r
+ "reliability": 99.9999,\r
+ "trafficDensity": 1,\r
+ "connDensity": 100000,\r
+ "expDataRate": 50,\r
+ "jitter": 1,\r
+ "survivalTime": 0,\r
+ "domainType":"cn",\r
+ "resourceSharingLevel":"shared"\r
+ }\r
+ }\r
+ ]\r
+ }\r
+ ]\r
+ }\r
+}\r
--- /dev/null
+{\r
+ "serviceProfile": {\r
+ "latency": 2,\r
+ "security": "High",\r
+ "reliability": 99.9999,\r
+ "trafficDensity": 1,\r
+ "connDensity": 100000,\r
+ "expDataRate": 50,\r
+ "jitter": 1,\r
+ "survivalTime": 0,\r
+ "resourceSharingLevel": "non-shared"\r
+ },\r
+ "serviceInfo":{\r
+ "serviceInstanceId": "209fb01e-60ca-4325-b074-c5ad4e0499f8",\r
+ "serviceName": ""\r
+ },\r
+ "requestInfo": {\r
+ "transactionId": "d290f1ee-6c54-4b01-90e6-d701748f0851",\r
+ "requestId": "d290f1ee-6c54-4b01-90e6-d701748f0851",\r
+ "callbackUrl": "http://0.0.0.0:9000/osdfCallback/",\r
+ "sourceId": "SO",\r
+ "timeout": 5\r
+ },\r
+ "NSTInfoList": [\r
+ {\r
+ "modelInvariantId": "fda3c1e8-7653-4acd-80ef-f5755c1d3859",\r
+ "modelVersionId": "a6906768-1cae-4e78-acd1-d753ac61f3e8",\r
+ "modelName": "URLLC_1"\r
+ }\r
+ ]\r
+}\r
--- /dev/null
+{\r
+ "requestId":"d290f1ee-6c54-4b01-90e6-d701748f0851",\r
+ "transactionId":"d290f1ee-6c54-4b01-90e6-d701748f0851",\r
+ "requestStatus":"completed",\r
+ "statusMessage":"",\r
+ "solutions":{\r
+ "sharedNSISolutions":[\r
+\r
+ ],\r
+ "newNSISolutions":[\r
+ {\r
+ "matchLevel":"",\r
+ "NSTInfo":{"invariantUUID": "fda3c1e8-7653-4acd-80ef-f5755c1d3859",\r
+ "UUID": "a6906768-1cae-4e78-acd1-d753ac61f3e8",\r
+ "NSTName": "URLLC_1"\r
+ },\r
+\r
+ "NSSISolutions":[\r
+ {\r
+ "sliceProfile": {\r
+ "latency": 2,\r
+ "security": "High",\r
+ "reliability": 99.9999,\r
+ "trafficDensity": 1,\r
+ "connDensity": 100000,\r
+ "expDataRate": 50,\r
+ "jitter": 1,\r
+ "survivalTime": 0,\r
+ "domainType": "cn",\r
+ "resourceSharingLevel": "non-shared"\r
+ }\r
+ }\r
+ ]\r
+ }\r
+ ]\r
+ }\r
+}\r
"connDensity": 100000,
"expDataRate": 50,
"jitter": 1,
- "survivalTime": 0
+ "survivalTime": 0,
+ "resourceSharingLevel":"shared"
},
"serviceInfo":{
"serviceInstanceId": "209fb01e-60ca-4325-b074-c5ad4e0499f8",
"modelName": "URLLC_1"
}
]
-}
\ No newline at end of file
+}
"invariantUUID": "",
"sliceProfile": [
{
- "coverage_area_ta_list": "[{\"province\":\"??\",\"city\":\"???\",\"county\":\"???\",\"street\":\"?????\"}]",
- "exp_data_rate_dl": 100,
- "exp_data_rate_ul": 100,
+ "coverageAreaTAList": "[{\"province\":\"??\",\"city\":\"???\",\"county\":\"???\",\"street\":\"?????\"}]",
+ "expDataRateDL": 100,
+ "expDataRateUL": 100,
"latency": 20,
- "resource_sharing_level": "0",
- "ue_mobility_level": "stationary"
+ "resourceSharingLevel": "0",
+ "uEMobilityLevel": "stationary"
}
]
},
"invariantUUID": "",
"sliceProfile": [
{
- "coverage_area_ta_list": "[{\"province\":\"??\",\"city\":\"???\",\"county\":\"???\",\"street\":\"?????\"}]",
- "exp_data_rate_dl": 100,
- "exp_data_rate_ul": 100,
+ "coverageAreaTAList": "[{\"province\":\"??\",\"city\":\"???\",\"county\":\"???\",\"street\":\"?????\"}]",
+ "expDataRateDL": 100,
+ "expDataRateUL": 100,
"latency": 15,
- "resource_sharing_level": "0",
- "ue_mobility_level": "stationary"
+ "resourceSharingLevel": "0",
+ "uEMobilityLevel": "stationary"
}
]
}
-# -------------------------------------------------------------------------
-# Copyright (C) 2020 Wipro Limited.
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-# -------------------------------------------------------------------------
-#
-
-import json
-import unittest
-from requests import RequestException
-
-from apps.slice_selection.optimizers.conductor.remote_opt_processor import process_nsi_selection_opt
-from osdf.adapters.local_data import local_policies
-from osdf.utils.interfaces import json_from_file, yaml_from_file
-from osdf.utils.programming_utils import DotDict
-import osdf.config.loader as config_loader
-from mock import patch, MagicMock
-import json
-from osdf.logging.osdf_logging import error_log, debug_log
-from osdf.adapters.policy.interface import get_policies
-
-
-class TestRemoteOptProcessor(unittest.TestCase):
- def setUp(self):
- self.config_spec = {
- "deployment": "config/osdf_config.yaml",
- "core": "config/common_config.yaml"
- }
- self.osdf_config = DotDict(config_loader.all_configs(**self.config_spec))
-
- def tearDown(self):
- patch.stopall()
-
- def test_process_nsi_selection_opt(self):
- main_dir = ""
- request_file = main_dir + 'test/apps/slice_selection/nsi_request.json'
- new_solution_response_file = main_dir + 'test/apps/slice_selection/new_solution_nsi_response.json'
- shared_solution_response_file = main_dir + 'test/apps/slice_selection/shared_solution_nsi_response.json'
- error_response_file = main_dir + 'test/apps/slice_selection/nsi_error_response.json'
-
- request_json = json_from_file(request_file)
- new_solution_response_json = json_from_file(new_solution_response_file)
- shared_solution_response_json = json_from_file(shared_solution_response_file)
- error_response_json = json_from_file(error_response_file)
-
- policies_path = main_dir + 'test/policy-local-files'
- slice_policies_file = main_dir + 'test/apps/slice_selection/slice_policies.txt'
-
- valid_policies_files = local_policies.get_policy_names_from_file(slice_policies_file)
- policies = [json_from_file(policies_path + '/' + name) for name in valid_policies_files]
- self.patcher_get_policies = patch('osdf.adapters.policy.interface.remote_api',
- return_value=policies)
- self.Mock_get_policies = self.patcher_get_policies.start()
-
- new_solution_conductor_response_file = 'test/apps/slice_selection/new_solution_conductor_response.json'
- new_solution_conductor_response = json_from_file(new_solution_conductor_response_file)
- self.patcher_req = patch('osdf.adapters.conductor.conductor.request',
- return_value=new_solution_conductor_response)
- self.Mock_req = self.patcher_req.start()
- self.assertEquals(new_solution_response_json, process_nsi_selection_opt(request_json, self.osdf_config))
- self.patcher_req.stop()
-
- shared_solution_conductor_response_file = 'test/apps/slice_selection/shared_solution_conductor_response.json'
- shared_solution_conductor_response = json_from_file(shared_solution_conductor_response_file)
- self.patcher_req = patch('osdf.adapters.conductor.conductor.request',
- return_value=shared_solution_conductor_response)
- self.Mock_req = self.patcher_req.start()
- self.assertEquals(shared_solution_response_json,
- process_nsi_selection_opt(request_json, self.osdf_config))
- self.patcher_req.stop()
-
- conductor_error_response_file = 'test/apps/slice_selection/conductor_error_response.json'
- conductor_error_response = json_from_file(conductor_error_response_file)
-
- self.patcher_req = patch('osdf.adapters.conductor.conductor.request',
- side_effect=RequestException(response=json.dumps(conductor_error_response)))
- self.Mock_req = self.patcher_req.start()
- self.assertEquals(error_response_json, process_nsi_selection_opt(request_json, self.osdf_config))
- self.patcher_req.stop()
-
- self.patcher_req = patch('osdf.adapters.conductor.conductor.request',
- side_effect=Exception("test_exception"))
- self.Mock_req = self.patcher_req.start()
- self.assertEquals('test_exception',
- process_nsi_selection_opt(request_json, self.osdf_config).get('statusMessage'))
- self.patcher_req.stop()
-
-
-if __name__ == "__main__":
- unittest.main()
-
+# -------------------------------------------------------------------------\r
+# Copyright (C) 2020 Wipro Limited.\r
+#\r
+# Licensed under the Apache License, Version 2.0 (the "License");\r
+# you may not use this file except in compliance with the License.\r
+# You may obtain a copy of the License at\r
+#\r
+# http://www.apache.org/licenses/LICENSE-2.0\r
+#\r
+# Unless required by applicable law or agreed to in writing, software\r
+# distributed under the License is distributed on an "AS IS" BASIS,\r
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
+# See the License for the specific language governing permissions and\r
+# limitations under the License.\r
+#\r
+# -------------------------------------------------------------------------\r
+#\r
+\r
+import json\r
+import unittest\r
+from requests import RequestException\r
+\r
+from apps.slice_selection.optimizers.conductor.remote_opt_processor import process_nsi_selection_opt\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
+import osdf.config.loader as config_loader\r
+from mock import patch, MagicMock\r
+import json\r
+from osdf.logging.osdf_logging import error_log, debug_log\r
+from osdf.adapters.policy.interface import get_policies\r
+\r
+\r
+class TestRemoteOptProcessor(unittest.TestCase):\r
+ def setUp(self):\r
+ self.config_spec = {\r
+ "deployment": "config/osdf_config.yaml",\r
+ "core": "config/common_config.yaml"\r
+ }\r
+ self.osdf_config = DotDict(config_loader.all_configs(**self.config_spec))\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
+ 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
+ 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
+ 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
+ # 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
+ self.patcher_req.stop()\r
+ # shared solution\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
+ 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
+ # 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
+ self.patcher_req.stop()\r
+\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=json.dumps(conductor_error_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
+ self.patcher_req.stop()\r
+\r
+ self.patcher_req = patch('osdf.adapters.conductor.conductor.request',\r
+ side_effect=Exception("test_exception"))\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
+ self.patcher_req.stop()\r
+\r
+\r
+if __name__ == "__main__":\r
+ unittest.main()\r
+\r
Code Review / optf / osdf.git / commitdiff