1 # -------------------------------------------------------------------------
2 # Copyright (C) 2020 Wipro Limited.
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 # -------------------------------------------------------------------------
20 from mock import patch
21 from requests import RequestException
23 from osdf.adapters.dcae import des
24 from osdf.adapters.dcae.des import DESException
25 import osdf.config.loader as config_loader
26 from osdf.utils.interfaces import json_from_file
27 from osdf.utils.programming_utils import DotDict
30 class TestDes(unittest.TestCase):
34 "deployment": "config/osdf_config.yaml",
35 "core": "config/common_config.yaml"
37 self.osdf_config = DotDict(config_loader.all_configs(**self.config_spec))
42 def test_extract_data(self):
43 response_file = 'test/adapters/dcae/des_response.json'
44 response_json = json_from_file(response_file)
46 des_config = self.osdf_config.core['PCI']['DES']
47 service_id = des_config['service_id']
48 data = des_config['filter']
49 expected = response_json['result']
50 response = mock.MagicMock()
51 response.status_code = 200
53 response.json.return_value = response_json
54 self.patcher_req = patch('requests.post', return_value=response)
55 self.Mock_req = self.patcher_req.start()
56 self.assertEqual(expected, des.extract_data(service_id, data))
57 self.patcher_req.stop()
59 response = mock.MagicMock()
60 response.status_code = 404
61 self.patcher_req = patch('requests.post', return_value=response)
62 self.Mock_req = self.patcher_req.start()
63 self.assertRaises(DESException, des.extract_data, service_id, data)
64 self.patcher_req.stop()
66 self.patcher_req = patch('requests.post', side_effect=RequestException("error"))
67 self.Mock_req = self.patcher_req.start()
68 self.assertRaises(DESException, des.extract_data, service_id, data)
69 self.patcher_req.stop()