functest/tests/unit/energy/test_functest_energy.py

   1 #!/usr/bin/env python
   2 # -*- coding: UTF-8 -*-
   3
   4 # Copyright (c) 2017 Orange and others.
   5 #
   6 # All rights reserved. This program and the accompanying materials
   7 # are made available under the terms of the Apache License, Version 2.0
   8 # which accompanies this distribution, and is available at
   9 # http://www.apache.org/licenses/LICENSE-2.0
  10
  11 """Unitary test for energy module."""
  12 # pylint: disable=unused-argument
  13 import logging
  14 import unittest
  15
  16 import mock
  17
  18 from functest.energy.energy import EnergyRecorder
  19 import functest.energy.energy as energy
  20
  21
  22 CASE_NAME = "UNIT_TEST_CASE"
  23 STEP_NAME = "UNIT_TEST_STEP"
  24
  25 PREVIOUS_SCENARIO = "previous_scenario"
  26 PREVIOUS_STEP = "previous_step"
  27
  28
  29 class MockHttpResponse(object):  # pylint: disable=too-few-public-methods
  30     """Mock response for Energy recorder API."""
  31
  32     def __init__(self, text, status_code):
  33         """Create an instance of MockHttpResponse."""
  34         self.text = text
  35         self.status_code = status_code
  36
  37
  38 API_OK = MockHttpResponse(
  39     '{"status": "OK"}',
  40     200
  41 )
  42 API_KO = MockHttpResponse(
  43     '{"message": "API-KO"}',
  44     500
  45 )
  46
  47 RECORDER_OK = MockHttpResponse(
  48     '{"environment": "UNIT_TEST",'
  49     ' "step": "string",'
  50     ' "scenario": "' + CASE_NAME + '"}',
  51     200
  52 )
  53 RECORDER_KO = MockHttpResponse(
  54     '{"message": "An unhandled API exception occurred (MOCK)"}',
  55     500
  56 )
  57
  58
  59 def config_loader_mock(config_key):
  60     """Return mocked config values."""
  61     if config_key == "energy_recorder.api_url":
  62         return "http://pod-uri:8888"
  63     elif config_key == "energy_recorder.api_user":
  64         return "user"
  65     elif config_key == "energy_recorder.api_password":
  66         return "password"
  67     else:
  68         raise Exception("Config not mocked")
  69
  70
  71 def config_loader_mock_no_creds(config_key):
  72     """Return mocked config values."""
  73     if config_key == "energy_recorder.api_url":
  74         return "http://pod-uri:8888"
  75     elif config_key == "energy_recorder.api_user":
  76         return ""
  77     elif config_key == "energy_recorder.api_password":
  78         return ""
  79     else:
  80         raise Exception("Config not mocked:" + config_key)
  81
  82
  83 class EnergyRecorderTest(unittest.TestCase):
  84     """Energy module unitary test suite."""
  85
  86     case_name = CASE_NAME
  87     request_headers = {'content-type': 'application/json'}
  88     returned_value_to_preserve = "value"
  89     exception_message_to_preserve = "exception_message"
  90
  91     @mock.patch('functest.energy.energy.requests.post',
  92                 return_value=RECORDER_OK)
  93     def test_start(self, post_mock=None, get_mock=None):
  94         """EnergyRecorder.start method (regular case)."""
  95         self.test_load_config()
  96         self.assertTrue(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             timeout=EnergyRecorder.CONNECTION_TIMOUT
 103         )
 104
 105     @mock.patch('functest.energy.energy.requests.post',
 106                 side_effect=Exception("Internal execution error (MOCK)"))
 107     def test_start_error(self, post_mock=None):
 108         """EnergyRecorder.start method (error in method)."""
 109         self.test_load_config()
 110         self.assertFalse(EnergyRecorder.start(self.case_name))
 111         post_mock.assert_called_once_with(
 112             EnergyRecorder.energy_recorder_api["uri"],
 113             auth=EnergyRecorder.energy_recorder_api["auth"],
 114             data=mock.ANY,
 115             headers=self.request_headers,
 116             timeout=EnergyRecorder.CONNECTION_TIMOUT
 117         )
 118
 119     @mock.patch('functest.energy.energy.requests.post',
 120                 return_value=RECORDER_KO)
 121     def test_start_api_error(self, post_mock=None):
 122         """EnergyRecorder.start method (API error)."""
 123         self.test_load_config()
 124         self.assertFalse(EnergyRecorder.start(self.case_name))
 125         post_mock.assert_called_once_with(
 126             EnergyRecorder.energy_recorder_api["uri"],
 127             auth=EnergyRecorder.energy_recorder_api["auth"],
 128             data=mock.ANY,
 129             headers=self.request_headers,
 130             timeout=EnergyRecorder.CONNECTION_TIMOUT
 131         )
 132
 133     @mock.patch('functest.energy.energy.requests.post',
 134                 return_value=RECORDER_OK)
 135     def test_set_step(self, post_mock=None):
 136         """EnergyRecorder.set_step method (regular case)."""
 137         self.test_load_config()
 138         self.assertTrue(EnergyRecorder.set_step(STEP_NAME))
 139         post_mock.assert_called_once_with(
 140             EnergyRecorder.energy_recorder_api["uri"] + "/step",
 141             auth=EnergyRecorder.energy_recorder_api["auth"],
 142             data=mock.ANY,
 143             headers=self.request_headers,
 144             timeout=EnergyRecorder.CONNECTION_TIMOUT
 145         )
 146
 147     @mock.patch('functest.energy.energy.requests.post',
 148                 return_value=RECORDER_KO)
 149     def test_set_step_api_error(self, post_mock=None):
 150         """EnergyRecorder.set_step method (API error)."""
 151         self.test_load_config()
 152         self.assertFalse(EnergyRecorder.set_step(STEP_NAME))
 153         post_mock.assert_called_once_with(
 154             EnergyRecorder.energy_recorder_api["uri"] + "/step",
 155             auth=EnergyRecorder.energy_recorder_api["auth"],
 156             data=mock.ANY,
 157             headers=self.request_headers,
 158             timeout=EnergyRecorder.CONNECTION_TIMOUT
 159         )
 160
 161     @mock.patch('functest.energy.energy.requests.post',
 162                 side_effect=Exception("Internal execution error (MOCK)"))
 163     def test_set_step_error(self, post_mock=None):
 164         """EnergyRecorder.set_step method (method error)."""
 165         self.test_load_config()
 166         self.assertFalse(EnergyRecorder.set_step(STEP_NAME))
 167         post_mock.assert_called_once_with(
 168             EnergyRecorder.energy_recorder_api["uri"] + "/step",
 169             auth=EnergyRecorder.energy_recorder_api["auth"],
 170             data=mock.ANY,
 171             headers=self.request_headers,
 172             timeout=EnergyRecorder.CONNECTION_TIMOUT
 173         )
 174
 175     @mock.patch('functest.energy.energy.requests.delete',
 176                 return_value=RECORDER_OK)
 177     def test_stop(self, delete_mock=None):
 178         """EnergyRecorder.stop method (regular case)."""
 179         self.test_load_config()
 180         self.assertTrue(EnergyRecorder.stop())
 181         delete_mock.assert_called_once_with(
 182             EnergyRecorder.energy_recorder_api["uri"],
 183             auth=EnergyRecorder.energy_recorder_api["auth"],
 184             headers=self.request_headers,
 185             timeout=EnergyRecorder.CONNECTION_TIMOUT
 186         )
 187
 188     @mock.patch('functest.energy.energy.requests.delete',
 189                 return_value=RECORDER_KO)
 190     def test_stop_api_error(self, delete_mock=None):
 191         """EnergyRecorder.stop method (API Error)."""
 192         self.test_load_config()
 193         self.assertFalse(EnergyRecorder.stop())
 194         delete_mock.assert_called_once_with(
 195             EnergyRecorder.energy_recorder_api["uri"],
 196             auth=EnergyRecorder.energy_recorder_api["auth"],
 197             headers=self.request_headers,
 198             timeout=EnergyRecorder.CONNECTION_TIMOUT
 199         )
 200
 201     @mock.patch('functest.energy.energy.requests.delete',
 202                 side_effect=Exception("Internal execution error (MOCK)"))
 203     def test_stop_error(self, delete_mock=None):
 204         """EnergyRecorder.stop method (method error)."""
 205         self.test_load_config()
 206         self.assertFalse(EnergyRecorder.stop())
 207         delete_mock.assert_called_once_with(
 208             EnergyRecorder.energy_recorder_api["uri"],
 209             auth=EnergyRecorder.energy_recorder_api["auth"],
 210             headers=self.request_headers,
 211             timeout=EnergyRecorder.CONNECTION_TIMOUT
 212         )
 213
 214     @energy.enable_recording
 215     def __decorated_method(self):
 216         """Call with to energy recorder decorators."""
 217         return self.returned_value_to_preserve
 218
 219     @energy.enable_recording
 220     def __decorated_method_with_ex(self):
 221         """Call with to energy recorder decorators."""
 222         raise Exception(self.exception_message_to_preserve)
 223
 224     @mock.patch("functest.energy.energy.EnergyRecorder.get_current_scenario",
 225                 return_value=None)
 226     @mock.patch("functest.energy.energy.EnergyRecorder")
 227     def test_decorators(self,
 228                         recorder_mock=None,
 229                         cur_scenario_mock=None):
 230         """Test energy module decorators."""
 231         self.__decorated_method()
 232         calls = [mock.call.start(self.case_name),
 233                  mock.call.stop()]
 234         recorder_mock.assert_has_calls(calls)
 235
 236     @mock.patch("functest.energy.energy.EnergyRecorder.get_current_scenario",
 237                 return_value={"scenario": PREVIOUS_SCENARIO,
 238                               "step": PREVIOUS_STEP})
 239     @mock.patch("functest.energy.energy.EnergyRecorder")
 240     @mock.patch("functest.utils.functest_utils.get_pod_name",
 241                 return_value="MOCK_POD")
 242     @mock.patch("functest.utils.functest_utils.get_functest_config",
 243                 side_effect=config_loader_mock)
 244     def test_decorators_with_previous(self,
 245                                       loader_mock=None,
 246                                       pod_mock=None,
 247                                       recorder_mock=None,
 248                                       cur_scenario_mock=None):
 249         """Test energy module decorators."""
 250         self.__decorated_method()
 251         calls = [mock.call.start(self.case_name),
 252                  mock.call.submit_scenario(PREVIOUS_SCENARIO,
 253                                            PREVIOUS_STEP)]
 254         recorder_mock.assert_has_calls(calls)
 255
 256     def test_decorator_preserve_return(self):
 257         """Test that decorator preserve method returned value."""
 258         self.test_load_config()
 259         self.assertTrue(
 260             self.__decorated_method() == self.returned_value_to_preserve
 261         )
 262
 263     @mock.patch(
 264         "functest.energy.energy.finish_session")
 265     def test_decorator_preserve_ex(self, finish_mock=None):
 266         """Test that decorator preserve method exceptions."""
 267         self.test_load_config()
 268         with self.assertRaises(Exception) as context:
 269             self.__decorated_method_with_ex()
 270         self.assertTrue(
 271             self.exception_message_to_preserve in context.exception
 272         )
 273         self.assertTrue(finish_mock.called)
 274
 275     @mock.patch("functest.utils.functest_utils.get_functest_config",
 276                 side_effect=config_loader_mock)
 277     @mock.patch("functest.utils.functest_utils.get_pod_name",
 278                 return_value="MOCK_POD")
 279     @mock.patch("functest.energy.energy.requests.get",
 280                 return_value=API_OK)
 281     def test_load_config(self, loader_mock=None, pod_mock=None,
 282                          get_mock=None):
 283         """Test load config."""
 284         EnergyRecorder.energy_recorder_api = None
 285         EnergyRecorder.load_config()
 286
 287         self.assertEquals(
 288             EnergyRecorder.energy_recorder_api["auth"],
 289             ("user", "password")
 290         )
 291         self.assertEquals(
 292             EnergyRecorder.energy_recorder_api["uri"],
 293             "http://pod-uri:8888/recorders/environment/MOCK_POD"
 294         )
 295
 296     @mock.patch("functest.utils.functest_utils.get_functest_config",
 297                 side_effect=config_loader_mock_no_creds)
 298     @mock.patch("functest.utils.functest_utils.get_pod_name",
 299                 return_value="MOCK_POD")
 300     @mock.patch("functest.energy.energy.requests.get",
 301                 return_value=API_OK)
 302     def test_load_config_no_creds(self, loader_mock=None, pod_mock=None,
 303                                   get_mock=None):
 304         """Test load config without creds."""
 305         EnergyRecorder.energy_recorder_api = None
 306         EnergyRecorder.load_config()
 307         self.assertEquals(EnergyRecorder.energy_recorder_api["auth"], None)
 308         self.assertEquals(
 309             EnergyRecorder.energy_recorder_api["uri"],
 310             "http://pod-uri:8888/recorders/environment/MOCK_POD"
 311         )
 312
 313     @mock.patch("functest.utils.functest_utils.get_functest_config",
 314                 return_value=None)
 315     @mock.patch("functest.utils.functest_utils.get_pod_name",
 316                 return_value="MOCK_POD")
 317     @mock.patch("functest.energy.energy.requests.get",
 318                 return_value=API_OK)
 319     def test_load_config_ex(self, loader_mock=None, pod_mock=None,
 320                             get_mock=None):
 321         """Test load config with exception."""
 322         with self.assertRaises(AssertionError):
 323             EnergyRecorder.energy_recorder_api = None
 324             EnergyRecorder.load_config()
 325         self.assertEquals(EnergyRecorder.energy_recorder_api, None)
 326
 327     @mock.patch("functest.utils.functest_utils.get_functest_config",
 328                 side_effect=config_loader_mock)
 329     @mock.patch("functest.utils.functest_utils.get_pod_name",
 330                 return_value="MOCK_POD")
 331     @mock.patch("functest.energy.energy.requests.get",
 332                 return_value=API_KO)
 333     def test_load_config_api_ko(self, loader_mock=None, pod_mock=None,
 334                                 get_mock=None):
 335         """Test load config with API unavailable."""
 336         EnergyRecorder.energy_recorder_api = None
 337         EnergyRecorder.load_config()
 338         self.assertEquals(EnergyRecorder.energy_recorder_api["available"],
 339                           False)
 340
 341     @mock.patch("functest.utils.functest_utils.get_functest_config",
 342                 return_value=None)
 343     @mock.patch("functest.utils.functest_utils.get_pod_name",
 344                 return_value="MOCK_POD")
 345     @mock.patch('functest.energy.energy.requests.get',
 346                 return_value=RECORDER_OK)
 347     def test_get_current_scenario(self, loader_mock=None,
 348                                   pod_mock=None, get_mock=None):
 349         """Test get_current_scenario."""
 350         self.test_load_config()
 351         scenario = EnergyRecorder.get_current_scenario()
 352         self.assertTrue(scenario is not None)
 353
 354
 355 if __name__ == "__main__":
 356     logging.disable(logging.CRITICAL)
 357     unittest.main(verbosity=2)