test/test_PolicyCalls.py

   1 # -------------------------------------------------------------------------
   2 #   Copyright (c) 2017-2018 AT&T Intellectual Property
   3 #
   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
   7 #
   8 #       http://www.apache.org/licenses/LICENSE-2.0
   9 #
  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.
  15 #
  16 # -------------------------------------------------------------------------
  17 #
  18 import json
  19 import unittest
  20
  21 from osdf.config.base import osdf_config
  22 from osdf.adapters.policy import interface
  23 from osdf.utils.interfaces import RestClient
  24 import yaml
  25 from mock import patch
  26 from osdf.optimizers.placementopt.conductor import translation
  27
  28
  29 class TestPolicyCalls(unittest.TestCase):
  30
  31     def test_get_subscriber_name(self):
  32         req_json_obj = json.loads(open("./test/placement-tests/request_mso.json").read())
  33         config_core = osdf_config.core
  34         pmain = config_core['policy_info']['placement']
  35         print(pmain)
  36         subs_name = interface.get_subscriber_name(req_json_obj, pmain)
  37         print("subscriber_name=", subs_name)
  38         self.assertEquals(subs_name, "Avteet_Chayal")
  39
  40
  41     def test_get_subscriber_name_null(self):
  42         req_json_file = "./test/placement-tests/request_mso_subs_name_null.json"
  43         req_json_obj = json.loads(open(req_json_file).read())
  44         config_core = osdf_config.core
  45
  46         pmain = config_core['policy_info']['placement']
  47         print(pmain)
  48         subs_name = interface.get_subscriber_name(req_json_obj, pmain)
  49         print("subscriber_name=", subs_name)
  50         self.assertEquals(subs_name, "DEFAULT")
  51
  52
  53     def test_get_subscriber_name_blank(self):
  54         req_json_file = "./test/placement-tests/request_mso_subs_name_blank.json"
  55         req_json_obj = json.loads(open(req_json_file).read())
  56         config_core = osdf_config.core
  57
  58         pmain = config_core['policy_info']['placement']
  59         print(pmain)
  60         subs_name = interface.get_subscriber_name(req_json_obj, pmain)
  61         print("subscriber_name=", subs_name)
  62         self.assertEquals(subs_name, "DEFAULT")
  63
  64
  65     def test_get_subscriber_name_default(self):
  66         req_json_file = "./test/placement-tests/request_mso_subs_name_default.json"
  67         req_json_obj = json.loads(open(req_json_file).read())
  68         config_core = osdf_config.core
  69
  70         pmain = config_core['policy_info']['placement']
  71         print(pmain)
  72         subs_name = interface.get_subscriber_name(req_json_obj, pmain)
  73         print("subscriber_name=", subs_name)
  74         self.assertEquals(subs_name, "DEFAULT")
  75
  76
  77     def test_get_subscriber_name_none(self):
  78         req_json_file = "./test/placement-tests/request_mso_subs_name_none.json"
  79         req_json_obj = json.loads(open(req_json_file).read())
  80         config_core = osdf_config.core
  81
  82         pmain = config_core['policy_info']['placement']
  83         print(pmain)
  84         subs_name = interface.get_subscriber_name(req_json_obj, pmain)
  85         print("subscriber_name=", subs_name)
  86         self.assertEquals(subs_name, "DEFAULT")
  87
  88
  89     def test_get_by_scope(self):
  90         req_json_file = "./test/placement-tests/testScoperequest.json"
  91         allPolicies = "./test/placement-tests/scopePolicies.json"
  92         req_json_obj = json.loads(open(req_json_file).read())
  93         req_json_obj2 = json.loads(open(allPolicies).read())
  94         config_core = osdf_config.core
  95         yamlFile = "./test/placement-tests/test_by_scope.yaml"
  96
  97         with open(yamlFile) as yamlFile2:
  98             policyConfigFile = yaml.load(yamlFile2)
  99             with patch('osdf.adapters.policy.interface.get_subscriber_role', return_value=('FFA Homing', [])) as mock_open:
 100                 with patch('osdf.utils.interfaces.RestClient.request', return_value = req_json_obj2):
 101                     policiesList = interface.get_by_scope(RestClient, req_json_obj, policyConfigFile, 'placement')
 102                     print(policiesList)
 103                     #catches Exception if policiesList is null
 104                     self.assertTrue(policiesList, 'is null')
 105                     self.assertRaises(Exception)
 106
 107     def test_gen_demands(self):
 108         actionsList = []
 109         genDemandslist = []
 110         req_json = "./test/placement-tests/testScoperequest.json"
 111         policiesList = "./test/placement-tests/vnfGroupPolicies.txt"
 112         fh = json.loads(open(policiesList).read())
 113         #print(fh)
 114         req_json = json.loads(open(req_json).read())
 115         config_core = osdf_config.core
 116         service_type = req_json['placementInfo'].get('serviceType', None)
 117         # service_type = data_mapping.get_request_service_type(req_json_file)
 118         genDemands = translation.gen_demands(req_json['placementInfo']['demandInfo'], fh)
 119         #print(genDemands)
 120         #print(req_json_file['placementInfo']['demandInfo']['placementDemand'][0])
 121         for action in req_json['placementInfo']['demandInfo']['placementDemand']:
 122             #print(action['resourceModuleName'])
 123             actionsList.append(action['resourceModuleName'])
 124         for key2,value in genDemands.items():
 125             #print(key2)
 126             genDemandslist.append(key2)
 127         #genDemandslist.remove('Primary IP_Mux_Demux updated_1 0')
 128         #catches Exception if lists are not equal
 129         self.assertListEqual(genDemandslist, actionsList, 'generated demands are not equal to the passed input [placementDemand][resourceModuleName] list')
 130
 131 if __name__ == '__main__':
 132     unittest.main()