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