diff options
Diffstat (limited to 'test/conductor')
| -rw-r--r-- | test/conductor/test_conductor_calls.py | 21 | ||||
| -rw-r--r-- | test/conductor/test_conductor_translation.py | 42 |
2 files changed, 51 insertions, 12 deletions
diff --git a/test/conductor/test_conductor_calls.py b/test/conductor/test_conductor_calls.py index 77d9a72..8b4411d 100644 --- a/test/conductor/test_conductor_calls.py +++ b/test/conductor/test_conductor_calls.py @@ -1,5 +1,6 @@ # ------------------------------------------------------------------------- # Copyright (c) 2018 AT&T Intellectual Property +# 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. @@ -17,7 +18,7 @@ # import unittest -from osdf.optimizers.placementopt.conductor import conductor +from osdf.adapters.conductor import conductor import osdf.config.loader as config_loader from osdf.utils.interfaces import json_from_file from osdf.utils.programming_utils import DotDict @@ -34,6 +35,10 @@ class TestConductorCalls(unittest.TestCase): self.osdf_config = DotDict(config_loader.all_configs(**self.config_spec)) self.lp = self.osdf_config.core.get('osdf_temp', {}).get('local_policies', {} ).get('placement_policy_files_vcpe') + self.template_fields = { + 'location_enabled': True, + 'version': '2017-10-10' + } def tearDown(self): pass @@ -41,12 +46,22 @@ class TestConductorCalls(unittest.TestCase): def test_request(self): req_json = json_from_file("./test/placement-tests/request.json") policies = pol.get_local_policies("test/policy-local-files/", self.lp) - conductor.request(req_json, self.osdf_config, policies) + req_info = req_json['requestInfo'] + demands = req_json['placementInfo']['placementDemands'] + request_parameters = req_json['placementInfo']['requestParameters'] + service_info = req_json['serviceInfo'] + conductor.request(req_info, demands, request_parameters, service_info, self.template_fields, + self.osdf_config, policies) def test_request_vfmod(self): req_json = json_from_file("./test/placement-tests/request_vfmod.json") policies = pol.get_local_policies("test/policy-local-files/", self.lp) - conductor.request(req_json, self.osdf_config, policies) + req_info = req_json['requestInfo'] + demands = req_json['placementInfo']['placementDemands'] + request_parameters = req_json['placementInfo']['requestParameters'] + service_info = req_json['serviceInfo'] + conductor.request(req_info, demands, request_parameters, service_info, self.template_fields, + self.osdf_config, policies) if __name__ == "__main__": diff --git a/test/conductor/test_conductor_translation.py b/test/conductor/test_conductor_translation.py index 27711f5..2a600e4 100644 --- a/test/conductor/test_conductor_translation.py +++ b/test/conductor/test_conductor_translation.py @@ -1,5 +1,6 @@ # ------------------------------------------------------------------------- # Copyright (c) 2017-2018 AT&T Intellectual Property +# 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. @@ -15,14 +16,11 @@ # # ------------------------------------------------------------------------- # -import mock import unittest -from flask import Response -from mock import patch from osdf.adapters.local_data import local_policies -from osdf.optimizers.placementopt.conductor import translation as tr -from osdf.utils.interfaces import json_from_file, yaml_from_file +from osdf.adapters.conductor import translation as tr +from osdf.utils.interfaces import json_from_file class TestConductorTranslation(unittest.TestCase): @@ -42,21 +40,47 @@ class TestConductorTranslation(unittest.TestCase): self.request_vfmod_json = json_from_file(parameter_data_file) self.policies = [json_from_file(policy_data_path + '/' + name) for name in valid_policies_files] + self.optimization_policies = [json_from_file(policy_data_path + '/' + + "slice-selection-files/opt_policy_nsi_reuse.json")] + def tearDown(self): pass def test_gen_demands(self): # need to run this only on vnf policies - vnf_policies = [x for x in self.policies if x["content"]["policyType"] == "vnfPolicy"] - res = tr.gen_demands(self.request_json, vnf_policies) + vnf_policies = [x for x in self.policies if x[list(x.keys())[0]]["type"] + == "onap.policies.optimization.VnfPolicy"] + res = tr.gen_demands(self.request_json['placementInfo']['placementDemands'], vnf_policies) + assert res is not None def test_gen_vfmod_demands(self): # need to run this only on vnf policies - vnf_policies = [x for x in self.policies if x["content"]["policyType"] == "vnfPolicy"] - res = tr.gen_demands(self.request_vfmod_json, vnf_policies) + vnf_policies = [x for x in self.policies if x[list(x.keys())[0]]["type"] + == "onap.policies.optimization.VnfPolicy"] + res = tr.gen_demands(self.request_vfmod_json['placementInfo']['placementDemands'], vnf_policies) assert res is not None + def test_gen_optimization_policy(self): + expected = [{ + "goal": "minimize", + "operation_function": { + "operator": "sum", + "operands": [ + { + "function": "attribute", + "params": { + "attribute": "creation_cost", + "demand": "embb-nst" + } + } + ] + } + }] + self.assertEqual(expected, + tr.gen_optimization_policy(self.request_vfmod_json['placementInfo']['placementDemands'], + self.optimization_policies)) + if __name__ == "__main__": unittest.main() |