functest/tests/unit/energy/test_functest_energy.py

   1 #!/usr/bin/env python
   2
   3 # All rights reserved. This program and the accompanying materials
   4 # are made available under the terms of the Apache License, Version 2.0
   5 # which accompanies this distribution, and is available at
   6 # http://www.apache.org/licenses/LICENSE-2.0
   7
   8 """Unitary test for energy module."""
   9 # pylint: disable=unused-argument
  10 import logging
  11 import unittest
  12
  13 import mock
  14
  15 from functest.energy.energy import EnergyRecorder
  16 import functest.energy.energy as energy
  17
  18
  19 CASE_NAME = "UNIT_test_CASE"
  20 STEP_NAME = "UNIT_test_STEP"
  21
  22
  23 class MockHttpResponse(object):  # pylint: disable=too-few-public-methods
  24     """Mock response for Energy recorder API."""
  25
  26     def __init__(self, text, status_code):
  27         """Create an instance of MockHttpResponse."""
  28         self.text = text
  29         self.status_code = status_code
  30
  31
  32 RECORDER_OK = MockHttpResponse(
  33     '{"environment": "UNIT_TEST",'
  34     ' "step": "string",'
  35     ' "scenario": "' + CASE_NAME + '"}',
  36     200
  37 )
  38 RECORDER_KO = MockHttpResponse(
  39     '{"message": "An unhandled API exception occurred (MOCK)"}',
  40     500
  41 )
  42
  43
  44 def config_loader_mock(config_key):
  45     """Return mocked config values."""
  46     if config_key == "energy_recorder.api_url":
  47         return "http://pod-uri:8888"
  48     elif config_key == "energy_recorder.api_user":
  49         return "user"
  50     elif config_key == "energy_recorder.api_password":
  51         return "password"
  52     else:
  53         raise Exception("Config not mocked")
  54
  55
  56 def config_loader_mock_no_creds(config_key):
  57     """Return mocked config values."""
  58     if config_key == "energy_recorder.api_url":
  59         return "http://pod-uri:8888"
  60     elif config_key == "energy_recorder.api_user":
  61         return ""
  62     elif config_key == "energy_recorder.api_password":
  63         return ""
  64     else:
  65         raise Exception("Config not mocked:" + config_key)
  66
  67
  68 class EnergyRecorderTest(unittest.TestCase):
  69     """Energy module unitary test suite."""
  70
  71     case_name = CASE_NAME
  72     request_headers = {'content-type': 'application/json'}
  73     returned_value_to_preserve = "value"
  74     exception_message_to_preserve = "exception_message"
  75
  76     @mock.patch('functest.energy.energy.requests.post',
  77                 return_value=RECORDER_OK)
  78     def test_start(self, post_mock=None):
  79         """EnergyRecorder.start method (regular case)."""
  80         self.test_load_config()
  81         self.assertTrue(EnergyRecorder.start(self.case_name))
  82         post_mock.assert_called_once_with(
  83             EnergyRecorder.energy_recorder_api["uri"],
  84             auth=EnergyRecorder.energy_recorder_api["auth"],
  85             data=mock.ANY,
  86             headers=self.request_headers
  87         )
  88
  89     @mock.patch('functest.energy.energy.requests.post',
  90                 side_effect=Exception("Internal execution error (MOCK)"))
  91     def test_start_error(self, post_mock=None):
  92         """EnergyRecorder.start method (error in method)."""
  93         self.test_load_config()
  94         self.assertFalse(EnergyRecorder.start(self.case_name))
  95         post_mock.assert_called_once_with(
  96             EnergyRecorder.energy_recorder_api["uri"],
  97             auth=EnergyRecorder.energy_recorder_api["auth"],
  98             data=mock.ANY,
  99             headers=self.request_headers
 100         )
 101
 102     @mock.patch('functest.energy.energy.requests.post',
 103                 return_value=RECORDER_KO)
 104     def test_start_api_error(self, post_mock=None):
 105         """EnergyRecorder.start method (API error)."""
 106         self.test_load_config()
 107         self.assertFalse(EnergyRecorder.start(self.case_name))
 108         post_mock.assert_called_once_with(
 109             EnergyRecorder.energy_recorder_api["uri"],
 110             auth=EnergyRecorder.energy_recorder_api["auth"],
 111             data=mock.ANY,
 112             headers=self.request_headers
 113         )
 114
 115     @mock.patch('functest.energy.energy.requests.post',
 116                 return_value=RECORDER_OK)
 117     def test_set_step(self, post_mock=None):
 118         """EnergyRecorder.set_step method (regular case)."""
 119         self.test_load_config()
 120         self.assertTrue(EnergyRecorder.set_step(STEP_NAME))
 121         post_mock.assert_called_once_with(
 122             EnergyRecorder.energy_recorder_api["uri"] + "/step",
 123             auth=EnergyRecorder.energy_recorder_api["auth"],
 124             data=mock.ANY,
 125             headers=self.request_headers
 126         )
 127
 128     @mock.patch('functest.energy.energy.requests.post',
 129                 return_value=RECORDER_KO)
 130     def test_set_step_api_error(self, post_mock=None):
 131         """EnergyRecorder.set_step method (API error)."""
 132         self.test_load_config()
 133         self.assertFalse(EnergyRecorder.set_step(STEP_NAME))
 134         post_mock.assert_called_once_with(
 135             EnergyRecorder.energy_recorder_api["uri"] + "/step",
 136             auth=EnergyRecorder.energy_recorder_api["auth"],
 137             data=mock.ANY,
 138             headers=self.request_headers
 139         )
 140
 141     @mock.patch('functest.energy.energy.requests.post',
 142                 side_effect=Exception("Internal execution error (MOCK)"))
 143     def test_set_step_error(self, post_mock=None):
 144         """EnergyRecorder.set_step method (method error)."""
 145         self.test_load_config()
 146         self.assertFalse(EnergyRecorder.set_step(STEP_NAME))
 147         post_mock.assert_called_once_with(
 148             EnergyRecorder.energy_recorder_api["uri"] + "/step",
 149             auth=EnergyRecorder.energy_recorder_api["auth"],
 150             data=mock.ANY,
 151             headers=self.request_headers
 152         )
 153
 154     @mock.patch('functest.energy.energy.requests.delete',
 155                 return_value=RECORDER_OK)
 156     def test_stop(self, delete_mock=None):
 157         """EnergyRecorder.stop method (regular case)."""
 158         self.test_load_config()
 159         self.assertTrue(EnergyRecorder.stop())
 160         delete_mock.assert_called_once_with(
 161             EnergyRecorder.energy_recorder_api["uri"],
 162             auth=EnergyRecorder.energy_recorder_api["auth"],
 163             headers=self.request_headers
 164         )
 165
 166     @mock.patch('functest.energy.energy.requests.delete',
 167                 return_value=RECORDER_KO)
 168     def test_stop_api_error(self, delete_mock=None):
 169         """EnergyRecorder.stop method (API Error)."""
 170         self.test_load_config()
 171         self.assertFalse(EnergyRecorder.stop())
 172         delete_mock.assert_called_once_with(
 173             EnergyRecorder.energy_recorder_api["uri"],
 174             auth=EnergyRecorder.energy_recorder_api["auth"],
 175             headers=self.request_headers
 176         )
 177
 178     @mock.patch('functest.energy.energy.requests.delete',
 179                 side_effect=Exception("Internal execution error (MOCK)"))
 180     def test_stop_error(self, delete_mock=None):
 181         """EnergyRecorder.stop method (method error)."""
 182         self.test_load_config()
 183         self.assertFalse(EnergyRecorder.stop())
 184         delete_mock.assert_called_once_with(
 185             EnergyRecorder.energy_recorder_api["uri"],
 186             auth=EnergyRecorder.energy_recorder_api["auth"],
 187             headers=self.request_headers
 188         )
 189
 190     @energy.enable_recording
 191     def __decorated_method(self):
 192         """Call with to energy recorder decorators."""
 193         return self.returned_value_to_preserve
 194
 195     @energy.enable_recording
 196     def __decorated_method_with_ex(self):
 197         """Call with to energy recorder decorators."""
 198         raise Exception(self.exception_message_to_preserve)
 199
 200     @mock.patch("functest.energy.energy.EnergyRecorder")
 201     @mock.patch("functest.utils.functest_utils.get_pod_name",
 202                 return_value="MOCK_POD")
 203     @mock.patch("functest.utils.functest_utils.get_functest_config",
 204                 side_effect=config_loader_mock)
 205     def test_decorators(self,
 206                         loader_mock=None,
 207                         pod_mock=None,
 208                         recorder_mock=None):
 209         """Test energy module decorators."""
 210         self.__decorated_method()
 211         calls = [mock.call.start(self.case_name),
 212                  mock.call.stop()]
 213         recorder_mock.assert_has_calls(calls)
 214
 215     def test_decorator_preserve_return(self):
 216         """Test that decorator preserve method returned value."""
 217         self.test_load_config()
 218         self.assertTrue(
 219             self.__decorated_method() == self.returned_value_to_preserve
 220         )
 221
 222     def test_decorator_preserve_ex(self):
 223         """Test that decorator preserve method exceptions."""
 224         self.test_load_config()
 225         with self.assertRaises(Exception) as context:
 226             self.__decorated_method_with_ex()
 227         self.assertTrue(
 228             self.exception_message_to_preserve in context.exception
 229         )
 230
 231     @mock.patch("functest.utils.functest_utils.get_functest_config",
 232                 side_effect=config_loader_mock)
 233     @mock.patch("functest.utils.functest_utils.get_pod_name",
 234                 return_value="MOCK_POD")
 235     def test_load_config(self, loader_mock=None, pod_mock=None):
 236         """Test load config."""
 237         EnergyRecorder.energy_recorder_api = None
 238         EnergyRecorder.load_config()
 239         self.assertEquals(
 240             EnergyRecorder.energy_recorder_api["auth"],
 241             ("user", "password")
 242         )
 243         self.assertEquals(
 244             EnergyRecorder.energy_recorder_api["uri"],
 245             "http://pod-uri:8888/recorders/environment/MOCK_POD"
 246         )
 247
 248     @mock.patch("functest.utils.functest_utils.get_functest_config",
 249                 side_effect=config_loader_mock_no_creds)
 250     @mock.patch("functest.utils.functest_utils.get_pod_name",
 251                 return_value="MOCK_POD")
 252     def test_load_config_no_creds(self, loader_mock=None, pod_mock=None):
 253         """Test load config without creds."""
 254         EnergyRecorder.energy_recorder_api = None
 255         EnergyRecorder.load_config()
 256         self.assertEquals(EnergyRecorder.energy_recorder_api["auth"], None)
 257         self.assertEquals(
 258             EnergyRecorder.energy_recorder_api["uri"],
 259             "http://pod-uri:8888/recorders/environment/MOCK_POD"
 260         )
 261
 262     @mock.patch("functest.utils.functest_utils.get_functest_config",
 263                 return_value=None)
 264     @mock.patch("functest.utils.functest_utils.get_pod_name",
 265                 return_value="MOCK_POD")
 266     def test_load_config_ex(self, loader_mock=None, pod_mock=None):
 267         """Test load config with exception."""
 268         with self.assertRaises(AssertionError):
 269             EnergyRecorder.energy_recorder_api = None
 270             EnergyRecorder.load_config()
 271         self.assertEquals(EnergyRecorder.energy_recorder_api, None)
 272
 273
 274 if __name__ == "__main__":
 275     logging.disable(logging.CRITICAL)
 276     unittest.main(verbosity=2)