def enable_recording(method):
"""
+ Record energy during method execution.
+
Decorator to record energy during "method" exection.
param method: Method to suround with start and stop
attribute
"""
def wrapper(*args):
- """Wrapper for decorator to handle method arguments."""
+ """
+ Record energy during method execution (implementation).
+
+ Wrapper for decorator to handle method arguments.
+ """
+ current_scenario = EnergyRecorder.get_current_scenario()
EnergyRecorder.start(args[0].case_name)
return_value = method(*args)
- EnergyRecorder.stop()
+ if current_scenario is None:
+ EnergyRecorder.stop()
+ else:
+ EnergyRecorder.submit_scenario(
+ current_scenario["scenario"],
+ current_scenario["step"]
+ )
return return_value
return wrapper
energy_recorder_api = None
# Default initial step
- INITIAL_STEP = "starting"
+ INITIAL_STEP = "running"
@staticmethod
def load_config():
}
@staticmethod
- def start(scenario):
+ def submit_scenario(scenario, step):
"""
- Start a recording session for scenario.
+ Submit a complet scenario definition to Energy recorder API.
- param scenario: Starting scenario
+ param scenario: Scenario name
:type scenario: string
+ param step: Step name
+ :type step: string
"""
return_status = True
try:
- EnergyRecorder.logger.debug("Starting recording")
+ EnergyRecorder.logger.debug("Submitting scenario")
# Ensure that connectyvity settings are loaded
EnergyRecorder.load_config()
# Create API payload
payload = {
- "step": EnergyRecorder.INITIAL_STEP,
+ "step": step,
"scenario": scenario
}
# Call API to start energy recording
}
)
if response.status_code != 200:
- log_msg = "Error while starting energy recording session\n{}"
+ log_msg = "Error while submitting scenario\n{}"
log_msg = log_msg.format(response.text)
EnergyRecorder.logger.info(log_msg)
return_status = False
+ except Exception: # pylint: disable=broad-except
+ # Default exception handler to ensure that method
+ # is safe for caller
+ EnergyRecorder.logger.exception(
+ "Error while submitting scenarion to energy recorder API"
+ )
+ return_status = False
+ return return_status
+
+ @staticmethod
+ def start(scenario):
+ """
+ Start a recording session for scenario.
+
+ param scenario: Starting scenario
+ :type scenario: string
+ """
+ return_status = True
+ try:
+ EnergyRecorder.logger.debug("Starting recording")
+ return_status = EnergyRecorder.submit_scenario(
+ scenario,
+ EnergyRecorder.INITIAL_STEP
+ )
except Exception: # pylint: disable=broad-except
# Default exception handler to ensure that method
)
return_status = False
return return_status
+
+ @staticmethod
+ def get_current_scenario():
+ """Get current running scenario (if any, None else)."""
+ EnergyRecorder.logger.debug("Getting current scenario")
+ return_value = None
+ print "In get current"
+ try:
+ # Ensure that connectyvity settings are loaded
+ EnergyRecorder.load_config()
+
+ # Call API get running scenario
+ response = requests.get(
+ EnergyRecorder.energy_recorder_api["uri"],
+ auth=EnergyRecorder.energy_recorder_api["auth"]
+ )
+ if response.status_code == 200:
+ return_value = json.loads(response.text)
+ elif response.status_code == 404:
+ log_msg = "No current running scenario at {}"
+ log_msg = log_msg.format(
+ EnergyRecorder.energy_recorder_api["uri"])
+ EnergyRecorder.logger.error(log_msg)
+ print log_msg
+ return_value = None
+ else:
+ log_msg = "Error while getting current scenario\n{}"
+ log_msg = log_msg.format(response.text)
+ EnergyRecorder.logger.error(log_msg)
+ print log_msg
+ return_value = None
+ except Exception: # pylint: disable=broad-except
+ # Default exception handler to ensure that method
+ # is safe for caller
+ EnergyRecorder.logger.exception(
+ "Error while getting current scenario from energy recorder API"
+ )
+ return_value = None
+ return return_value
#!/usr/bin/env python
+# -*- coding: UTF-8 -*-
+# Copyright (c) 2017 Orange and others.
+#
# All rights reserved. This program and the accompanying materials
# are made available under the terms of the Apache License, Version 2.0
# which accompanies this distribution, and is available at
import functest.energy.energy as energy
-CASE_NAME = "UNIT_test_CASE"
-STEP_NAME = "UNIT_test_STEP"
+CASE_NAME = "UNIT_TEST_CASE"
+STEP_NAME = "UNIT_TEST_STEP"
+
+PREVIOUS_SCENARIO = "previous_scenario"
+PREVIOUS_STEP = "previous_step"
class MockHttpResponse(object): # pylint: disable=too-few-public-methods
"""Call with to energy recorder decorators."""
raise Exception(self.exception_message_to_preserve)
+ @mock.patch("functest.energy.energy.EnergyRecorder.get_current_scenario",
+ return_value=None)
@mock.patch("functest.energy.energy.EnergyRecorder")
@mock.patch("functest.utils.functest_utils.get_pod_name",
return_value="MOCK_POD")
def test_decorators(self,
loader_mock=None,
pod_mock=None,
- recorder_mock=None):
+ recorder_mock=None,
+ cur_scenario_mock=None):
"""Test energy module decorators."""
self.__decorated_method()
calls = [mock.call.start(self.case_name),
mock.call.stop()]
recorder_mock.assert_has_calls(calls)
+ @mock.patch("functest.energy.energy.EnergyRecorder.get_current_scenario",
+ return_value={"scenario": PREVIOUS_SCENARIO,
+ "step": PREVIOUS_STEP})
+ @mock.patch("functest.energy.energy.EnergyRecorder")
+ @mock.patch("functest.utils.functest_utils.get_pod_name",
+ return_value="MOCK_POD")
+ @mock.patch("functest.utils.functest_utils.get_functest_config",
+ side_effect=config_loader_mock)
+ def test_decorators_with_previous(self,
+ loader_mock=None,
+ pod_mock=None,
+ recorder_mock=None,
+ cur_scenario_mock=None):
+ """Test energy module decorators."""
+ self.__decorated_method()
+ calls = [mock.call.start(self.case_name),
+ mock.call.submit_scenario(PREVIOUS_SCENARIO,
+ PREVIOUS_STEP)]
+ recorder_mock.assert_has_calls(calls)
+
def test_decorator_preserve_return(self):
"""Test that decorator preserve method returned value."""
self.test_load_config()
EnergyRecorder.load_config()
self.assertEquals(EnergyRecorder.energy_recorder_api, None)
+ @mock.patch("functest.utils.functest_utils.get_functest_config",
+ return_value=None)
+ @mock.patch("functest.utils.functest_utils.get_pod_name",
+ return_value="MOCK_POD")
+ @mock.patch('functest.energy.energy.requests.get',
+ return_value=RECORDER_OK)
+ def test_get_current_scenario(self, loader_mock=None,
+ pod_mock=None, get_mock=None):
+ """Test get_current_scenario."""
+ self.test_load_config()
+ scenario = EnergyRecorder.get_current_scenario()
+ self.assertTrue(scenario is not None)
+
if __name__ == "__main__":
logging.disable(logging.CRITICAL)
Code Review / functest.git / commitdiff