import json
import logging
+import os
from functools import wraps
import requests
from six.moves import urllib
-from functest.utils.constants import CONST
-import functest.utils.functest_utils as ft_utils
-
def finish_session(current_scenario):
"""Finish a recording session."""
# Singleton pattern for energy_recorder_api static member
# Load only if not previouly done
if EnergyRecorder.energy_recorder_api is None:
- environment = CONST.__getattribute__('NODE_NAME')
+ environment = os.environ['NODE_NAME']
+ assert environment
# API URL
- energy_recorder_uri = ft_utils.get_functest_config(
- "energy_recorder.api_url")
+ energy_recorder_uri = os.environ["ENERGY_RECORDER_API_URL"]
assert energy_recorder_uri
- assert environment
+
+ # Creds
+ creds_usr = os.environ.get("ENERGY_RECORDER_API_USER", "")
+ creds_pass = os.environ.get("ENERGY_RECORDER_API_PASSWORD", "")
uri_comp = "/recorders/environment/"
uri_comp += urllib.parse.quote_plus(environment)
- # Creds
- creds_usr = ft_utils.get_functest_config(
- "energy_recorder.api_user")
- creds_pass = ft_utils.get_functest_config(
- "energy_recorder.api_password")
-
if creds_usr != "" and creds_pass != "":
energy_recorder_api_auth = (creds_usr, creds_pass)
else:
"""Unitary test for energy module."""
# pylint: disable=unused-argument
import logging
-import requests
+import os
import unittest
import mock
+import requests
from functest.energy.energy import EnergyRecorder
import functest.energy.energy as energy
-from functest.utils.constants import CONST
CASE_NAME = "UNIT_TEST_CASE"
STEP_NAME = "UNIT_TEST_STEP"
)
-def config_loader_mock(config_key):
- """Return mocked config values."""
- if config_key == "energy_recorder.api_url":
- return "http://pod-uri:8888"
- elif config_key == "energy_recorder.api_user":
- return "user"
- elif config_key == "energy_recorder.api_password":
- return "password"
-
-
-def config_loader_mock_no_creds(config_key):
- """Return mocked config values."""
- if config_key == "energy_recorder.api_url":
- return "http://pod-uri:8888"
- elif config_key == "energy_recorder.api_user":
- return ""
- elif config_key == "energy_recorder.api_password":
- return ""
-
-
# pylint: disable=too-many-public-methods
class EnergyRecorderTest(unittest.TestCase):
"""Energy module unitary test suite."""
returned_value_to_preserve = "value"
exception_message_to_preserve = "exception_message"
+ @staticmethod
+ def _set_env_creds():
+ """Set config values."""
+ os.environ["ENERGY_RECORDER_API_URL"] = "http://pod-uri:8888"
+ os.environ["ENERGY_RECORDER_API_USER"] = "user"
+ os.environ["ENERGY_RECORDER_API_PASSWORD"] = "password"
+
+ @staticmethod
+ def _set_env_nocreds():
+ """Set config values."""
+ os.environ["ENERGY_RECORDER_API_URL"] = "http://pod-uri:8888"
+ del os.environ["ENERGY_RECORDER_API_USER"]
+ del os.environ["ENERGY_RECORDER_API_PASSWORD"]
+
@mock.patch('functest.energy.energy.requests.post',
return_value=RECORDER_OK)
def test_start(self, post_mock=None, get_mock=None):
return_value={"scenario": PREVIOUS_SCENARIO,
"step": PREVIOUS_STEP})
@mock.patch("functest.energy.energy.EnergyRecorder")
- @mock.patch("functest.utils.functest_utils.get_functest_config",
- side_effect=config_loader_mock)
def test_decorators_with_previous(self,
- loader_mock=None,
recorder_mock=None,
cur_scenario_mock=None):
"""Test energy module decorators."""
- CONST.__setattr__('NODE_NAME', 'MOCK_POD')
+ os.environ['NODE_NAME'] = 'MOCK_POD'
+ self._set_env_creds()
self.__decorated_method()
calls = [mock.call.start(self.case_name),
mock.call.submit_scenario(PREVIOUS_SCENARIO,
)
self.assertTrue(finish_mock.called)
- @mock.patch("functest.utils.functest_utils.get_functest_config",
- side_effect=config_loader_mock)
@mock.patch("functest.energy.energy.requests.get",
return_value=API_OK)
def test_load_config(self, loader_mock=None, get_mock=None):
"""Test load config."""
- CONST.__setattr__('NODE_NAME', 'MOCK_POD')
+ os.environ['NODE_NAME'] = 'MOCK_POD'
+ self._set_env_creds()
EnergyRecorder.energy_recorder_api = None
EnergyRecorder.load_config()
"http://pod-uri:8888/recorders/environment/MOCK_POD"
)
- @mock.patch("functest.utils.functest_utils.get_functest_config",
- side_effect=config_loader_mock_no_creds)
@mock.patch("functest.energy.energy.requests.get",
return_value=API_OK)
def test_load_config_no_creds(self, loader_mock=None, get_mock=None):
"""Test load config without creds."""
- CONST.__setattr__('NODE_NAME', 'MOCK_POD')
+ os.environ['NODE_NAME'] = 'MOCK_POD'
+ self._set_env_nocreds()
EnergyRecorder.energy_recorder_api = None
EnergyRecorder.load_config()
self.assertEquals(EnergyRecorder.energy_recorder_api["auth"], None)
"http://pod-uri:8888/recorders/environment/MOCK_POD"
)
- @mock.patch("functest.utils.functest_utils.get_functest_config",
- return_value=None)
@mock.patch("functest.energy.energy.requests.get",
return_value=API_OK)
def test_load_config_ex(self, loader_mock=None, get_mock=None):
"""Test load config with exception."""
- CONST.__setattr__('NODE_NAME', 'MOCK_POD')
- with self.assertRaises(AssertionError):
- EnergyRecorder.energy_recorder_api = None
- EnergyRecorder.load_config()
- self.assertEquals(EnergyRecorder.energy_recorder_api, None)
-
- @mock.patch("functest.utils.functest_utils.get_functest_config",
- side_effect=config_loader_mock)
+ for key in ['NODE_NAME', 'ENERGY_RECORDER_API_URL']:
+ os.environ[key] = ''
+ with self.assertRaises(AssertionError):
+ EnergyRecorder.energy_recorder_api = None
+ EnergyRecorder.load_config()
+ self.assertEquals(EnergyRecorder.energy_recorder_api, None)
+
@mock.patch("functest.energy.energy.requests.get",
return_value=API_KO)
def test_load_config_api_ko(self, loader_mock=None, get_mock=None):
"""Test load config with API unavailable."""
- CONST.__setattr__('NODE_NAME', 'MOCK_POD')
+ os.environ['NODE_NAME'] = 'MOCK_POD'
+ self._set_env_creds()
EnergyRecorder.energy_recorder_api = None
EnergyRecorder.load_config()
self.assertEquals(EnergyRecorder.energy_recorder_api["available"],
False)
- @mock.patch("functest.utils.functest_utils.get_functest_config",
- return_value=None)
@mock.patch('functest.energy.energy.requests.get',
return_value=RECORDER_OK)
def test_get_current_scenario(self, loader_mock=None, get_mock=None):
"""Test get_current_scenario."""
- CONST.__setattr__('NODE_NAME', 'MOCK_POD')
+ os.environ['NODE_NAME'] = 'MOCK_POD'
self.test_load_config()
scenario = EnergyRecorder.get_current_scenario()
self.assertTrue(scenario is not None)
return_value=RECORDER_NOT_FOUND)
def test_current_scenario_not_found(self, get_mock=None):
"""Test get current scenario not existing."""
- CONST.__setattr__('NODE_NAME', 'MOCK_POD')
+ os.environ['NODE_NAME'] = 'MOCK_POD'
self.test_load_config()
scenario = EnergyRecorder.get_current_scenario()
self.assertTrue(scenario is None)
return_value=RECORDER_KO)
def test_current_scenario_api_error(self, get_mock=None):
"""Test get current scenario with API error."""
- CONST.__setattr__('NODE_NAME', 'MOCK_POD')
+ os.environ['NODE_NAME'] = 'MOCK_POD'
self.test_load_config()
scenario = EnergyRecorder.get_current_scenario()
self.assertTrue(scenario is None)