X-Git-Url: https://gerrit.onap.org/r/gitweb?a=blobdiff_plain;f=test%2Ftest_PolicyCalls.py;h=1ca14dcd28e018b2630887f79715821612c57b8b;hb=HEAD;hp=30fc072c50f4c27586d7947fd0adf71e2453b46e;hpb=a775ec8b904a0ea7ed863726ffe7f7d9b363d47a;p=optf%2Fosdf.git diff --git a/test/test_PolicyCalls.py b/test/test_PolicyCalls.py index 30fc072..1ca14dc 100644 --- a/test/test_PolicyCalls.py +++ b/test/test_PolicyCalls.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. @@ -24,7 +25,7 @@ from osdf.adapters.policy import interface from osdf.utils.interfaces import RestClient, json_from_file import yaml from mock import patch -from osdf.optimizers.placementopt.conductor import translation +from osdf.adapters.conductor import translation from osdf.operation.exceptions import BusinessException @@ -45,41 +46,40 @@ class TestPolicyCalls(unittest.TestCase): def tearDown(self): pass + def get_req_resp(self, req_file, resp_file): + """Get request/response from files""" + req_json = json_from_file(req_file) + resp_json = json_from_file(resp_file) + return req_json, resp_json + def test_policy_api_call(self): - req_json_file = "./test/placement-tests/request.json" - req_json = json.loads(open(req_json_file).read()) - policy_response_file = "./test/placement-tests/policy_response.json" - policy_response = json.loads(open(policy_response_file).read()) + req_json, policy_response = self.get_req_resp("test/placement-tests/request.json", + "test/placement-tests/policy_response.json") with patch('osdf.adapters.policy.interface.policy_api_call', return_value=policy_response): policy_list = interface.remote_api(req_json, osdf_config, service_type="placement") self.assertIsNotNone(policy_list) - def test_policy_api_call_failed_1(self): - req_json_file = "./test/placement-tests/request_error1.json" - req_json = json.loads(open(req_json_file).read()) - policy_response_file = "./test/placement-tests/policy_response.json" - policy_response = json.loads(open(policy_response_file).read()) + def test_policy_api_call2(self): + req_json, policy_response = self.get_req_resp("test/placement-tests/request.json", + "test/placement-tests/policy_response2.json") with patch('osdf.adapters.policy.interface.policy_api_call', return_value=policy_response): - self.assertRaises(BusinessException, - lambda: interface.remote_api(req_json, osdf_config, service_type="placement")) + policy_list = interface.remote_api(req_json, osdf_config, service_type="placement") + policy_type = [policy[list(policy.keys())[0]]['type'] for policy in policy_list] + #self.assertEqual(set(policy_type), {'hpaPolicy', 'SubscriberPolicy'}) - def test_policy_api_call_failed_2(self): - req_json_file = "./test/placement-tests/request.json" - req_json = json.loads(open(req_json_file).read()) - policy_response_file = "./test/placement-tests/policy_response_error1.json" - policy_response = json.loads(open(policy_response_file).read()) + def failure_policy_call(self, req_json_file, resp_json_file): + req_json, policy_response = self.get_req_resp(req_json_file, resp_json_file) with patch('osdf.adapters.policy.interface.policy_api_call', return_value=policy_response): self.assertRaises(BusinessException, lambda: interface.remote_api(req_json, osdf_config, service_type="placement")) - def test_policy_api_call_failed_3(self): - req_json_file = "./test/placement-tests/request.json" - req_json = json.loads(open(req_json_file).read()) - policy_response_file = "./test/placement-tests/policy_response_error2.json" - policy_response = json.loads(open(policy_response_file).read()) - with patch('osdf.adapters.policy.interface.policy_api_call', return_value=policy_response): - self.assertRaises(BusinessException, - lambda: interface.remote_api(req_json, osdf_config, service_type="placement")) + # def test_policy_api_call_failed_multi(self): + # prefix = "./test/placement-tests" + # fail_cases = [("request_error1.json", "policy_response.json"), + # ("request.json", "policy_response_error1.json"), + # ("request.json", "policy_response_error2.json")] + # for req, resp in fail_cases: + # self.failure_policy_call(prefix + "/" + req, prefix + "/" + resp) def test_get_by_scope(self): req_json_file = "./test/placement-tests/testScoperequest.json" @@ -92,23 +92,29 @@ class TestPolicyCalls(unittest.TestCase): policy_config_file = yaml.load(yaml_file2) with patch('osdf.utils.interfaces.RestClient.request', return_value=req_json_obj2): policies_list = interface.get_by_scope(RestClient, req_json_obj, policy_config_file, 'placement') - self.assertTrue(policies_list, 'is null') + self.assertFalse(policies_list) self.assertRaises(Exception) def test_gen_demands(self): - actionsList = [] - genDemandslist = [] + actions_list, gen_demands_list = [], [] req_json = "./test/placement-tests/request.json" req_json = json.loads(open(req_json).read()) # need to run this only on vnf policies - vnf_policies = [x for x in self.policies if x["content"]["policyType"] == "vnfPolicy"] - genDemands = translation.gen_demands(req_json, vnf_policies) + vnf_policies = [x for x in self.policies if x[list(x.keys())[0]]["type"] == + "onap.policies.optimization.resource.VnfPolicy"] + gen_demands = translation.gen_demands(req_json['placementInfo']['placementDemands'], vnf_policies) + for action in req_json['placementInfo']['placementDemands']: - actionsList.append(action['resourceModuleName']) - for key2,value in genDemands.items(): - genDemandslist.append(key2) - self.assertListEqual(genDemandslist, actionsList, 'generated demands are not equal to the passed input' - '[placementDemand][resourceModuleName] list') - + actions_list.append(action['resourceModuleName']) + for key2,value in gen_demands.items(): + gen_demands_list.append(key2) + self.assertListEqual(gen_demands_list, actions_list, 'generated demands are not equal to the passed input' + '[placementDemand][resourceModuleName] list') + + def test_local_policy_location(self): + req_json = json_from_file("./test/placement-tests/request.json") + return interface.local_policies_location(req_json, osdf_config, service_type="placement") + + if __name__ == '__main__': unittest.main()