4 from osdf.config.base import osdf_config
5 from osdf.adapters.policy import interface
6 from osdf.utils.interfaces import RestClient
9 from osdf.optimizers.placementopt.conductor import translation
12 class TestPolicyCalls(unittest.TestCase):
14 def test_get_subscriber_name(self):
15 req_json_obj = json.loads(open("./test/placement-tests/request_mso.json").read())
16 config_core = osdf_config.core
17 pmain = config_core['policy_info']['placement']
19 subs_name = interface.get_subscriber_name(req_json_obj, pmain)
20 print("subscriber_name=", subs_name)
21 self.assertEquals(subs_name, "Avteet_Chayal")
24 def test_get_subscriber_name_null(self):
25 req_json_file = "./test/placement-tests/request_mso_subs_name_null.json"
26 req_json_obj = json.loads(open(req_json_file).read())
27 config_core = osdf_config.core
29 pmain = config_core['policy_info']['placement']
31 subs_name = interface.get_subscriber_name(req_json_obj, pmain)
32 print("subscriber_name=", subs_name)
33 self.assertEquals(subs_name, "DEFAULT")
36 def test_get_subscriber_name_blank(self):
37 req_json_file = "./test/placement-tests/request_mso_subs_name_blank.json"
38 req_json_obj = json.loads(open(req_json_file).read())
39 config_core = osdf_config.core
41 pmain = config_core['policy_info']['placement']
43 subs_name = interface.get_subscriber_name(req_json_obj, pmain)
44 print("subscriber_name=", subs_name)
45 self.assertEquals(subs_name, "DEFAULT")
48 def test_get_subscriber_name_default(self):
49 req_json_file = "./test/placement-tests/request_mso_subs_name_default.json"
50 req_json_obj = json.loads(open(req_json_file).read())
51 config_core = osdf_config.core
53 pmain = config_core['policy_info']['placement']
55 subs_name = interface.get_subscriber_name(req_json_obj, pmain)
56 print("subscriber_name=", subs_name)
57 self.assertEquals(subs_name, "DEFAULT")
60 def test_get_subscriber_name_none(self):
61 req_json_file = "./test/placement-tests/request_mso_subs_name_none.json"
62 req_json_obj = json.loads(open(req_json_file).read())
63 config_core = osdf_config.core
65 pmain = config_core['policy_info']['placement']
67 subs_name = interface.get_subscriber_name(req_json_obj, pmain)
68 print("subscriber_name=", subs_name)
69 self.assertEquals(subs_name, "DEFAULT")
72 def test_get_by_scope(self):
73 req_json_file = "./test/placement-tests/testScoperequest.json"
74 allPolicies = "./test/placement-tests/scopePolicies.json"
75 req_json_obj = json.loads(open(req_json_file).read())
76 req_json_obj2 = json.loads(open(allPolicies).read())
77 config_core = osdf_config.core
78 yamlFile = "./test/placement-tests/test_by_scope.yaml"
80 with open(yamlFile) as yamlFile2:
81 policyConfigFile = yaml.load(yamlFile2)
82 with patch('osdf.adapters.policy.interface.get_subscriber_role', return_value=('FFA Homing', [])) as mock_open:
83 with patch('osdf.utils.interfaces.RestClient.request', return_value = req_json_obj2):
84 policiesList = interface.get_by_scope(RestClient, req_json_obj, policyConfigFile, 'placement')
86 #catches Exception if policiesList is null
87 self.assertTrue(policiesList, 'is null')
88 self.assertRaises(Exception)
90 def test_gen_demands(self):
93 req_json = "./test/placement-tests/testScoperequest.json"
94 policiesList = "./test/placement-tests/vnfGroupPolicies.txt"
95 fh = json.loads(open(policiesList).read())
97 req_json = json.loads(open(req_json).read())
98 config_core = osdf_config.core
99 service_type = req_json['placementInfo'].get('serviceType', None)
100 # service_type = data_mapping.get_request_service_type(req_json_file)
101 genDemands = translation.gen_demands(req_json['placementInfo']['demandInfo'], fh)
103 #print(req_json_file['placementInfo']['demandInfo']['placementDemand'][0])
104 for action in req_json['placementInfo']['demandInfo']['placementDemand']:
105 #print(action['resourceModuleName'])
106 actionsList.append(action['resourceModuleName'])
107 for key2,value in genDemands.items():
109 genDemandslist.append(key2)
110 #genDemandslist.remove('Primary IP_Mux_Demux updated_1 0')
111 #catches Exception if lists are not equal
112 self.assertListEqual(genDemandslist, actionsList, 'generated demands are not equal to the passed input [placementDemand][resourceModuleName] list')
114 if __name__ == '__main__':