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