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 os
  15 import unittest
  16
  17 import mock
  18 import requests
  19
  20 from functest.energy.energy import EnergyRecorder
  21 import functest.energy.energy as energy
  22
  23 CASE_NAME = "UNIT_TEST_CASE"
  24 STEP_NAME = "UNIT_TEST_STEP"
  25
  26 PREVIOUS_SCENARIO = "previous_scenario"
  27 PREVIOUS_STEP = "previous_step"
  28
  29
  30 class MockHttpResponse(object):  # pylint: disable=too-few-public-methods
  31     """Mock response for Energy recorder API."""
  32
  33     def __init__(self, text, status_code):
  34         """Create an instance of MockHttpResponse."""
  35         self.text = text
  36         self.status_code = status_code
  37
  38
  39 API_OK = MockHttpResponse(
  40     '{"status": "OK"}',
  41     200
  42 )
  43 API_KO = MockHttpResponse(
  44     '{"message": "API-KO"}',
  45     500
  46 )
  47
  48 RECORDER_OK = MockHttpResponse(
  49     '{"environment": "UNIT_TEST",'
  50     ' "step": "string",'
  51     ' "scenario": "' + CASE_NAME + '"}',
  52     200
  53 )
  54 RECORDER_KO = MockHttpResponse(
  55     '{"message": "An unhandled API exception occurred (MOCK)"}',
  56     500
  57 )
  58 RECORDER_NOT_FOUND = MockHttpResponse(
  59     '{"message": "Recorder not found (MOCK)"}',
  60     404
  61 )
  62
  63
  64 # pylint: disable=too-many-public-methods
  65 class EnergyRecorderTest(unittest.TestCase):
  66     """Energy module unitary test suite."""
  67
  68     case_name = CASE_NAME
  69     request_headers = {'content-type': 'application/json'}
  70     returned_value_to_preserve = "value"
  71     exception_message_to_preserve = "exception_message"
  72
  73     @staticmethod
  74     def _set_env_creds():
  75         """Set config values."""
  76         os.environ["ENERGY_RECORDER_API_URL"] = "http://pod-uri:8888"
  77         os.environ["ENERGY_RECORDER_API_USER"] = "user"
  78         os.environ["ENERGY_RECORDER_API_PASSWORD"] = "password"
  79
  80     @staticmethod
  81     def _set_env_nocreds():
  82         """Set config values."""
  83         os.environ["ENERGY_RECORDER_API_URL"] = "http://pod-uri:8888"
  84         del os.environ["ENERGY_RECORDER_API_USER"]
  85         del os.environ["ENERGY_RECORDER_API_PASSWORD"]
  86
  87     @mock.patch('functest.energy.energy.requests.post',
  88                 return_value=RECORDER_OK)
  89     def test_start(self, post_mock=None, get_mock=None):
  90         """EnergyRecorder.start method (regular case)."""
  91         self.test_load_config()
  92         self.assertTrue(EnergyRecorder.start(self.case_name))
  93         post_mock.assert_called_once_with(
  94             EnergyRecorder.energy_recorder_api["uri"],
  95             auth=EnergyRecorder.energy_recorder_api["auth"],
  96             data=mock.ANY,
  97             headers=self.request_headers,
  98             timeout=EnergyRecorder.CONNECTION_TIMEOUT
  99         )
 100
 101     @mock.patch('functest.energy.energy.requests.post',
 102                 side_effect=Exception("Internal execution error (MOCK)"))
 103     def test_start_error(self, post_mock=None):
 104         """EnergyRecorder.start method (error in method)."""
 105         self.test_load_config()
 106         self.assertFalse(EnergyRecorder.start(self.case_name))
 107         post_mock.assert_called_once_with(
 108             EnergyRecorder.energy_recorder_api["uri"],
 109             auth=EnergyRecorder.energy_recorder_api["auth"],
 110             data=mock.ANY,
 111             headers=self.request_headers,
 112             timeout=EnergyRecorder.CONNECTION_TIMEOUT
 113         )
 114
 115     @mock.patch('functest.energy.energy.EnergyRecorder.load_config',
 116                 side_effect=Exception("Internal execution error (MOCK)"))
 117     def test_start_exception(self, conf_loader_mock=None):
 118         """EnergyRecorder.start test with exception during execution."""
 119         start_status = EnergyRecorder.start(CASE_NAME)
 120         self.assertFalse(start_status)
 121
 122     @mock.patch('functest.energy.energy.requests.post',
 123                 return_value=RECORDER_KO)
 124     def test_start_api_error(self, post_mock=None):
 125         """EnergyRecorder.start method (API error)."""
 126         self.test_load_config()
 127         self.assertFalse(EnergyRecorder.start(self.case_name))
 128         post_mock.assert_called_once_with(
 129             EnergyRecorder.energy_recorder_api["uri"],
 130             auth=EnergyRecorder.energy_recorder_api["auth"],
 131             data=mock.ANY,
 132             headers=self.request_headers,
 133             timeout=EnergyRecorder.CONNECTION_TIMEOUT
 134         )
 135
 136     @mock.patch('functest.energy.energy.requests.post',
 137                 return_value=RECORDER_OK)
 138     def test_set_step(self, post_mock=None):
 139         """EnergyRecorder.set_step method (regular case)."""
 140         self.test_load_config()
 141         self.assertTrue(EnergyRecorder.set_step(STEP_NAME))
 142         post_mock.assert_called_once_with(
 143             EnergyRecorder.energy_recorder_api["uri"] + "/step",
 144             auth=EnergyRecorder.energy_recorder_api["auth"],
 145             data=mock.ANY,
 146             headers=self.request_headers,
 147             timeout=EnergyRecorder.CONNECTION_TIMEOUT
 148         )
 149
 150     @mock.patch('functest.energy.energy.requests.post',
 151                 return_value=RECORDER_KO)
 152     def test_set_step_api_error(self, post_mock=None):
 153         """EnergyRecorder.set_step method (API error)."""
 154         self.test_load_config()
 155         self.assertFalse(EnergyRecorder.set_step(STEP_NAME))
 156         post_mock.assert_called_once_with(
 157             EnergyRecorder.energy_recorder_api["uri"] + "/step",
 158             auth=EnergyRecorder.energy_recorder_api["auth"],
 159             data=mock.ANY,
 160             headers=self.request_headers,
 161             timeout=EnergyRecorder.CONNECTION_TIMEOUT
 162         )
 163
 164     @mock.patch('functest.energy.energy.requests.post',
 165                 side_effect=Exception("Internal execution error (MOCK)"))
 166     def test_set_step_error(self, post_mock=None):
 167         """EnergyRecorder.set_step method (method error)."""
 168         self.test_load_config()
 169         self.assertFalse(EnergyRecorder.set_step(STEP_NAME))
 170         post_mock.assert_called_once_with(
 171             EnergyRecorder.energy_recorder_api["uri"] + "/step",
 172             auth=EnergyRecorder.energy_recorder_api["auth"],
 173             data=mock.ANY,
 174             headers=self.request_headers,
 175             timeout=EnergyRecorder.CONNECTION_TIMEOUT
 176         )
 177
 178     @mock.patch('functest.energy.energy.EnergyRecorder.load_config',
 179                 side_effect=requests.exceptions.ConnectionError())
 180     def test_set_step_connection_error(self, conf_loader_mock=None):
 181         """EnergyRecorder.start test with exception during execution."""
 182         step_status = EnergyRecorder.set_step(STEP_NAME)
 183         self.assertFalse(step_status)
 184
 185     @mock.patch('functest.energy.energy.requests.delete',
 186                 return_value=RECORDER_OK)
 187     def test_stop(self, delete_mock=None):
 188         """EnergyRecorder.stop method (regular case)."""
 189         self.test_load_config()
 190         self.assertTrue(EnergyRecorder.stop())
 191         delete_mock.assert_called_once_with(
 192             EnergyRecorder.energy_recorder_api["uri"],
 193             auth=EnergyRecorder.energy_recorder_api["auth"],
 194             headers=self.request_headers,
 195             timeout=EnergyRecorder.CONNECTION_TIMEOUT
 196         )
 197
 198     @mock.patch('functest.energy.energy.requests.delete',
 199                 return_value=RECORDER_KO)
 200     def test_stop_api_error(self, delete_mock=None):
 201         """EnergyRecorder.stop method (API Error)."""
 202         self.test_load_config()
 203         self.assertFalse(EnergyRecorder.stop())
 204         delete_mock.assert_called_once_with(
 205             EnergyRecorder.energy_recorder_api["uri"],
 206             auth=EnergyRecorder.energy_recorder_api["auth"],
 207             headers=self.request_headers,
 208             timeout=EnergyRecorder.CONNECTION_TIMEOUT
 209         )
 210
 211     @mock.patch('functest.energy.energy.requests.delete',
 212                 side_effect=Exception("Internal execution error (MOCK)"))
 213     def test_stop_error(self, delete_mock=None):
 214         """EnergyRecorder.stop method (method error)."""
 215         self.test_load_config()
 216         self.assertFalse(EnergyRecorder.stop())
 217         delete_mock.assert_called_once_with(
 218             EnergyRecorder.energy_recorder_api["uri"],
 219             auth=EnergyRecorder.energy_recorder_api["auth"],
 220             headers=self.request_headers,
 221             timeout=EnergyRecorder.CONNECTION_TIMEOUT
 222         )
 223
 224     @energy.enable_recording
 225     def __decorated_method(self):
 226         """Call with to energy recorder decorators."""
 227         return self.returned_value_to_preserve
 228
 229     @energy.enable_recording
 230     def __decorated_method_with_ex(self):
 231         """Call with to energy recorder decorators."""
 232         raise Exception(self.exception_message_to_preserve)
 233
 234     @mock.patch("functest.energy.energy.EnergyRecorder.get_current_scenario",
 235                 return_value=None)
 236     @mock.patch("functest.energy.energy.EnergyRecorder")
 237     def test_decorators(self,
 238                         recorder_mock=None,
 239                         cur_scenario_mock=None):
 240         """Test energy module decorators."""
 241         self.__decorated_method()
 242         calls = [mock.call.start(self.case_name),
 243                  mock.call.stop()]
 244         recorder_mock.assert_has_calls(calls)
 245
 246     @mock.patch("functest.energy.energy.EnergyRecorder.get_current_scenario",
 247                 return_value={"scenario": PREVIOUS_SCENARIO,
 248                               "step": PREVIOUS_STEP})
 249     @mock.patch("functest.energy.energy.EnergyRecorder")
 250     def test_decorators_with_previous(self,
 251                                       recorder_mock=None,
 252                                       cur_scenario_mock=None):
 253         """Test energy module decorators."""
 254         os.environ['NODE_NAME'] = 'MOCK_POD'
 255         self._set_env_creds()
 256         self.__decorated_method()
 257         calls = [mock.call.start(self.case_name),
 258                  mock.call.submit_scenario(PREVIOUS_SCENARIO,
 259                                            PREVIOUS_STEP)]
 260         recorder_mock.assert_has_calls(calls, True)
 261
 262     def test_decorator_preserve_return(self):
 263         """Test that decorator preserve method returned value."""
 264         self.test_load_config()
 265         self.assertTrue(
 266             self.__decorated_method() == self.returned_value_to_preserve
 267         )
 268
 269     @mock.patch(
 270         "functest.energy.energy.finish_session")
 271     def test_decorator_preserve_ex(self, finish_mock=None):
 272         """Test that decorator preserve method exceptions."""
 273         self.test_load_config()
 274         with self.assertRaises(Exception) as context:
 275             self.__decorated_method_with_ex()
 276         self.assertTrue(
 277             self.exception_message_to_preserve in str(context.exception)
 278         )
 279         self.assertTrue(finish_mock.called)
 280
 281     @mock.patch("functest.energy.energy.requests.get",
 282                 return_value=API_OK)
 283     def test_load_config(self, loader_mock=None, get_mock=None):
 284         """Test load config."""
 285         os.environ['NODE_NAME'] = 'MOCK_POD'
 286         self._set_env_creds()
 287         EnergyRecorder.energy_recorder_api = None
 288         EnergyRecorder.load_config()
 289
 290         self.assertEquals(
 291             EnergyRecorder.energy_recorder_api["auth"],
 292             ("user", "password")
 293         )
 294         self.assertEquals(
 295             EnergyRecorder.energy_recorder_api["uri"],
 296             "http://pod-uri:8888/recorders/environment/MOCK_POD"
 297         )
 298
 299     @mock.patch("functest.energy.energy.requests.get",
 300                 return_value=API_OK)
 301     def test_load_config_no_creds(self, loader_mock=None, get_mock=None):
 302         """Test load config without creds."""
 303         os.environ['NODE_NAME'] = 'MOCK_POD'
 304         self._set_env_nocreds()
 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.energy.energy.requests.get",
 314                 return_value=API_OK)
 315     def test_load_config_ex(self, loader_mock=None, get_mock=None):
 316         """Test load config with exception."""
 317         for key in ['NODE_NAME', 'ENERGY_RECORDER_API_URL']:
 318             os.environ[key] = ''
 319             with self.assertRaises(AssertionError):
 320                 EnergyRecorder.energy_recorder_api = None
 321                 EnergyRecorder.load_config()
 322             self.assertEquals(EnergyRecorder.energy_recorder_api, None)
 323
 324     @mock.patch("functest.energy.energy.requests.get",
 325                 return_value=API_KO)
 326     def test_load_config_api_ko(self, loader_mock=None, get_mock=None):
 327         """Test load config with API unavailable."""
 328         os.environ['NODE_NAME'] = 'MOCK_POD'
 329         self._set_env_creds()
 330         EnergyRecorder.energy_recorder_api = None
 331         EnergyRecorder.load_config()
 332         self.assertEquals(EnergyRecorder.energy_recorder_api["available"],
 333                           False)
 334
 335     @mock.patch('functest.energy.energy.requests.get',
 336                 return_value=RECORDER_OK)
 337     def test_get_current_scenario(self, loader_mock=None, get_mock=None):
 338         """Test get_current_scenario."""
 339         os.environ['NODE_NAME'] = 'MOCK_POD'
 340         self.test_load_config()
 341         scenario = EnergyRecorder.get_current_scenario()
 342         self.assertTrue(scenario is not None)
 343
 344     @mock.patch('functest.energy.energy.requests.get',
 345                 return_value=RECORDER_NOT_FOUND)
 346     def test_current_scenario_not_found(self, get_mock=None):
 347         """Test get current scenario not existing."""
 348         os.environ['NODE_NAME'] = 'MOCK_POD'
 349         self.test_load_config()
 350         scenario = EnergyRecorder.get_current_scenario()
 351         self.assertTrue(scenario is None)
 352
 353     @mock.patch('functest.energy.energy.requests.get',
 354                 return_value=RECORDER_KO)
 355     def test_current_scenario_api_error(self, get_mock=None):
 356         """Test get current scenario with API error."""
 357         os.environ['NODE_NAME'] = 'MOCK_POD'
 358         self.test_load_config()
 359         scenario = EnergyRecorder.get_current_scenario()
 360         self.assertTrue(scenario is None)
 361
 362     @mock.patch('functest.energy.energy.EnergyRecorder.load_config',
 363                 side_effect=Exception("Internal execution error (MOCK)"))
 364     def test_current_scenario_exception(self, get_mock=None):
 365         """Test get current scenario with exception."""
 366         scenario = EnergyRecorder.get_current_scenario()
 367         self.assertTrue(scenario is None)
 368
 369 if __name__ == "__main__":
 370     logging.disable(logging.CRITICAL)
 371     unittest.main(verbosity=2)