test/adapters/dcae/test_des.py

   1 # -------------------------------------------------------------------------
   2 #   Copyright (C) 2020 Wipro Limited.
   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
  19 import mock
  20 from mock import patch
  21 from requests import RequestException
  22 from requests.exceptions import HTTPError
  23 import unittest
  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
  29
  30
  31 class TestDes(unittest.TestCase):
  32
  33     def setUp(self):
  34         self.config_spec = {
  35             "deployment": "config/osdf_config.yaml",
  36             "core": "config/common_config.yaml"
  37         }
  38         self.osdf_config = DotDict(config_loader.all_configs(**self.config_spec))
  39
  40     def tearDown(self):
  41         pass
  42
  43     def test_extract_data(self):
  44         response_file = 'test/adapters/dcae/des_response.json'
  45         response_json = json_from_file(response_file)
  46
  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
  53         response.ok = True
  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()
  59
  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()
  67
  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()