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