1 # -------------------------------------------------------------------------
2 # Copyright (c) 2017-2018 AT&T Intellectual Property
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
16 # -------------------------------------------------------------------------
21 from osdf.adapters.local_data import local_policies
22 from osdf.config.base import osdf_config
23 from osdf.adapters.policy import interface
24 from osdf.utils.interfaces import RestClient, json_from_file
26 from mock import patch
27 from osdf.optimizers.placementopt.conductor import translation
30 class TestPolicyCalls(unittest.TestCase):
34 parameter_data_file = main_dir + "test/placement-tests/request.json"
35 policy_data_path = main_dir + "test/policy-local-files/"
36 local_config_file = main_dir + "config/common_config.yaml"
38 valid_policies_list_file = policy_data_path + '/' + 'meta-valid-policies.txt'
39 valid_policies_files = local_policies.get_policy_names_from_file(valid_policies_list_file)
41 self.request_json = json_from_file(parameter_data_file)
42 self.policies = [json_from_file(policy_data_path + '/' + name) for name in valid_policies_files]
47 def test_get_subscriber_name(self):
48 req_json_obj = json.loads(open("./test/placement-tests/request_mso.json").read())
49 config_core = osdf_config.core
50 pmain = config_core['policy_info']['placement']
52 subs_name = interface.get_subscriber_name(req_json_obj, pmain)
53 print("subscriber_name=", subs_name)
54 self.assertEquals(subs_name, "Avteet_Chayal")
56 def test_get_subscriber_name_null(self):
57 req_json_file = "./test/placement-tests/request_mso_subs_name_null.json"
58 req_json_obj = json.loads(open(req_json_file).read())
59 config_core = osdf_config.core
61 pmain = config_core['policy_info']['placement']
63 subs_name = interface.get_subscriber_name(req_json_obj, pmain)
64 print("subscriber_name=", subs_name)
65 self.assertEquals(subs_name, "DEFAULT")
67 def test_get_subscriber_name_blank(self):
68 req_json_file = "./test/placement-tests/request_mso_subs_name_blank.json"
69 req_json_obj = json.loads(open(req_json_file).read())
70 config_core = osdf_config.core
72 pmain = config_core['policy_info']['placement']
74 subs_name = interface.get_subscriber_name(req_json_obj, pmain)
75 print("subscriber_name=", subs_name)
76 self.assertEquals(subs_name, "DEFAULT")
78 def test_get_subscriber_name_default(self):
79 req_json_file = "./test/placement-tests/request_mso_subs_name_default.json"
80 req_json_obj = json.loads(open(req_json_file).read())
81 config_core = osdf_config.core
83 pmain = config_core['policy_info']['placement']
85 subs_name = interface.get_subscriber_name(req_json_obj, pmain)
86 print("subscriber_name=", subs_name)
87 self.assertEquals(subs_name, "DEFAULT")
89 def test_get_subscriber_name_none(self):
90 req_json_file = "./test/placement-tests/request_mso_subs_name_none.json"
91 req_json_obj = json.loads(open(req_json_file).read())
92 config_core = osdf_config.core
94 pmain = config_core['policy_info']['placement']
96 subs_name = interface.get_subscriber_name(req_json_obj, pmain)
97 print("subscriber_name=", subs_name)
98 self.assertEquals(subs_name, "DEFAULT")
100 def test_get_by_scope(self):
101 req_json_file = "./test/placement-tests/testScoperequest.json"
102 allPolicies = "./test/placement-tests/scopePolicies.json"
103 req_json_obj = json.loads(open(req_json_file).read())
104 req_json_obj2 = json.loads(open(allPolicies).read())
105 yamlFile = "./test/placement-tests/test_by_scope.yaml"
107 with open(yamlFile) as yamlFile2:
108 policy_config_file = yaml.load(yamlFile2)
109 with patch('osdf.adapters.policy.interface.get_subscriber_role',
110 return_value=('FFA Homing', [])) as mock_open:
111 with patch('osdf.utils.interfaces.RestClient.request', return_value=req_json_obj2):
112 policiesList = interface.get_by_scope(RestClient, req_json_obj, policy_config_file, 'placement')
113 self.assertTrue(policiesList, 'is null')
114 self.assertRaises(Exception)
116 def test_gen_demands(self):
119 req_json = "./test/placement-tests/request.json"
120 req_json = json.loads(open(req_json).read())
121 genDemands = translation.gen_demands(req_json, self.policies)
122 for action in req_json['placementInfo']['placementDemands']:
123 actionsList.append(action['resourceModuleName'])
124 for key2,value in genDemands.items():
125 genDemandslist.append(key2)
126 self.assertListEqual(genDemandslist, actionsList, 'generated demands are not equal to the passed input'
127 '[placementDemand][resourceModuleName] list')
129 if __name__ == '__main__':