conductor/conductor/tests/unit/data/plugins/inventory_provider/test_dcae.py

   1 import json
   2 import mock
   3 import unittest
   4
   5 from unittest.mock import patch, Mock
   6 from oslo_config import cfg
   7 import conductor.data.plugins.inventory_provider.dcae as dcae
   8 from conductor.data.plugins.inventory_provider.dcae import DCAE
   9
  10
  11
  12 class TestDCAE(unittest.TestCase):
  13
  14     def setUp(self):
  15         CONF = cfg.CONF
  16         CONF.register_opts(dcae.DCAE_OPTS, group='dcae')
  17         self.conf = CONF
  18         self.dcae_ep = DCAE()
  19
  20
  21     def tearDown(self):
  22         mock.patch.stopall()
  23
  24
  25     def test_get_dcae_response(self):
  26         nssi_candidate_file = './conductor/tests/unit/data/plugins/inventory_provider/nssi_candidate.json'
  27         uLThptDiff = 70
  28         dLThptDiff = 73
  29         nssi_candidates = json.loads(open(nssi_candidate_file).read())
  30         candidates=[]
  31         for nssi_candidate in nssi_candidates:
  32             inventory_type=nssi_candidate.get("inventory_type")
  33             if inventory_type == 'nssi':
  34                 candidates.extend(nssi_candidates)
  35         final_nssi_candidates_file = './conductor/tests/unit/data/plugins/inventory_provider/nssi_candidate_updated.json'
  36         final_nssi_candidates = json.loads(open(final_nssi_candidates_file).read())
  37         response = mock.MagicMock()
  38         response.content = None
  39         ff = open('./conductor/tests/unit/data/plugins/inventory_provider/dcae_response.json', "rb")
  40         file_res = ff.read()
  41         response.status_code = 200
  42         response.ok = True
  43         response.content=file_res
  44         self.mock_get_request = mock.patch.object(DCAE, 'get_dcae_response',
  45                 return_value=response)
  46         self.mock_get_request.start()
  47         self.maxDiff=None
  48         response1 = mock.MagicMock()
  49         response1.status_code = 200
  50         response1.ok = True
  51         response1.json.return_value = 100
  52         self.mock_get_dLThpt = mock.patch.object(DCAE, 'get_dLThpt', return_value=response1)
  53         self.mock_get_dLThpt.start()
  54         self.mock_get_uLThpt = mock.patch.object(DCAE, 'get_uLThpt', return_value=response1)
  55         self.mock_get_uLThpt.start()
  56         self.dcae_ep.get_difference = Mock(return_value = 70)
  57         self.assertEqual(final_nssi_candidates,
  58                 self.dcae_ep.capacity_filter(candidates))
  59
  60
  61     def test_get_difference(self):
  62         a = 20
  63         b = 10
  64         difference = a-b
  65         self.assertEqual(difference, self.dcae_ep.get_difference(a,b))
  66
  67
  68     def test_get_uLThpt(self):
  69         uLThpt = 30
  70         candidate_id = "cdad9f49-4201-4e3a-aac1-b0f27902c299"
  71         ff = open('./conductor/tests/unit/data/plugins/inventory_provider/dcae_response.json')
  72         file_res = ff.read()
  73         self.assertEqual(uLThpt, self.dcae_ep.get_uLThpt(file_res,candidate_id))
  74
  75
  76     def test_get_dLThpt(self):
  77         dLThpt = 27
  78         candidate_id = "cdad9f49-4201-4e3a-aac1-b0f27902c299"
  79         ff = open('./conductor/tests/unit/data/plugins/inventory_provider/dcae_response.json')
  80         file_res = ff.read()
  81         self.assertEqual(dLThpt, self.dcae_ep.get_dLThpt(file_res,candidate_id))
  82
  83
  84
  85
  86
  87
  88
  89
  90
  91
  92