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
22 from requests.exceptions import HTTPError
24 from osdf.adapters.dcae import des
25 from osdf.adapters.dcae.des import DESException
26 import osdf.config.loader as config_loader
27 from osdf.utils.interfaces import json_from_file
28 from osdf.utils.programming_utils import DotDict
31 class TestDes(unittest.TestCase):
35 "deployment": "config/osdf_config.yaml",
36 "core": "config/common_config.yaml"
38 self.osdf_config = DotDict(config_loader.all_configs(**self.config_spec))
43 def test_extract_data(self):
44 response_file = 'test/adapters/dcae/des_response.json'
45 response_json = json_from_file(response_file)
47 des_config = self.osdf_config.core['PCI']['DES']
48 service_id = des_config['service_id']
49 data = des_config['filter']
50 expected = response_json['result']
51 response = mock.MagicMock()
52 response.status_code = 200
54 response.json.return_value = response_json
55 self.patcher_req = patch('requests.request', return_value=response)
56 self.Mock_req = self.patcher_req.start()
57 self.assertEqual(expected, des.extract_data(service_id, data))
58 self.patcher_req.stop()
60 response = mock.MagicMock()
61 response.status_code = 404
62 response.raise_for_status.side_effect = HTTPError("404")
63 self.patcher_req = patch('requests.request', return_value=response)
64 self.Mock_req = self.patcher_req.start()
65 self.assertRaises(DESException, des.extract_data, service_id, data)
66 self.patcher_req.stop()
68 self.patcher_req = patch('requests.request', side_effect=RequestException("error"))
69 self.Mock_req = self.patcher_req.start()
70 self.assertRaises(DESException, des.extract_data, service_id, data)
71 self.patcher_req.stop()