It's also renamed source_envfile().
Change-Id: I0e7c38c2def125961f86dc2bc9a63cfb6ad87c03
Signed-off-by: Cédric Ollivier <cedric.ollivier@orange.com>
import collections
import logging
-import pkg_resources
import socket
from flask import jsonify
from flasgger.utils import swag_from
+import pkg_resources
from functest.api.base import ApiResource
from functest.api.common import api_utils
+from functest.ci import run_tests
from functest.cli.commands.cli_os import OpenStack
-from functest.utils import openstack_utils as os_utils
from functest.utils.constants import CONST
LOGGER = logging.getLogger(__name__)
endpoint='{0}/credentials'.format(ENDPOINT_CREDS))
def get(self): # pylint: disable=no-self-use
""" Get credentials """
- os_utils.source_credentials(CONST.__getattribute__('env_file'))
+ run_tests.Runner.source_envfile(CONST.__getattribute__('env_file'))
credentials_show = OpenStack.show_credentials()
return jsonify(credentials_show)
LOGGER.info("Sourcing the OpenStack RC file...")
try:
- os_utils.source_credentials(rc_file)
+ run_tests.Runner.source_envfile(rc_file)
except Exception as err: # pylint: disable=broad-except
LOGGER.exception('Failed to source the OpenStack RC file')
return api_utils.result_handler(status=0, data=str(err))
import functest.ci.tier_builder as tb
import functest.core.testcase as testcase
import functest.utils.functest_utils as ft_utils
-import functest.utils.openstack_utils as os_utils
from functest.utils.constants import CONST
# __name__ cannot be used here
CONST.__getattribute__('DEPLOY_SCENARIO'),
pkg_resources.resource_filename('functest', 'ci/testcases.yaml'))
+ @staticmethod
+ def source_envfile(rc_file):
+ """Source the env file passed as arg"""
+ with open(rc_file, "r") as rcfd:
+ for line in rcfd:
+ var = (line.rstrip('"\n').replace('export ', '').split(
+ "=") if re.search(r'(.*)=(.*)', line) else None)
+ # The two next lines should be modified as soon as rc_file
+ # conforms with common rules. Be aware that it could induce
+ # issues if value starts with '
+ if var:
+ key = re.sub(r'^["\' ]*|[ \'"]*$', '', var[0])
+ value = re.sub(r'^["\' ]*|[ \'"]*$', '', "".join(var[1:]))
+ os.environ[key] = value
+ setattr(CONST, key, value)
+
@staticmethod
def get_run_dict(testname):
"""Obtain the 'run' block of the testcase from testcases.yaml"""
try:
if 'test' in kwargs:
LOGGER.debug("Sourcing the credential file...")
- os_utils.source_credentials(CONST.__getattribute__('env_file'))
+ self.source_envfile(getattr(CONST, 'env_file'))
LOGGER.debug("Test args: %s", kwargs['test'])
if self.tiers.get_tier(kwargs['test']):
import pkg_resources
from functest.opnfv_tests.openstack.tempest import conf_utils
-from functest.utils import openstack_utils
-from functest.utils.constants import CONST
from functest.opnfv_tests.openstack.tempest.tempest \
import TempestResourcesManager
def generate_tempestconf(self):
""" Generate tempest.conf file"""
try:
- openstack_utils.source_credentials(
- CONST.__getattribute__('env_file'))
resources = self.resources.create(create_project=True,
use_custom_images=True,
use_custom_flavors=True)
import logging
import unittest
+import os
import mock
args[1].assert_called_once_with(
"Cannot get %s's config options", testname)
+ def _test_source_envfile(self, msg, key='OS_TENANT_NAME', value='admin'):
+ try:
+ del os.environ[key]
+ except Exception: # pylint: disable=broad-except
+ pass
+ envfile = 'rc_file'
+ with mock.patch('six.moves.builtins.open',
+ mock.mock_open(read_data=msg),
+ create=True) as mock_method:
+ mock_method.return_value.__iter__ = lambda self: iter(
+ self.readline, '')
+ self.runner.source_envfile(envfile)
+ mock_method.assert_called_once_with(envfile, 'r')
+ self.assertEqual(os.environ[key], value)
+
+ def test_source_envfile(self):
+ self._test_source_envfile('OS_TENANT_NAME=admin')
+ self._test_source_envfile('OS_TENANT_NAME= admin')
+ self._test_source_envfile('OS_TENANT_NAME = admin')
+ self._test_source_envfile('OS_TENANT_NAME = "admin"')
+ self._test_source_envfile('export OS_TENANT_NAME=admin')
+ self._test_source_envfile('export OS_TENANT_NAME =admin')
+ self._test_source_envfile('export OS_TENANT_NAME = admin')
+ self._test_source_envfile('export OS_TENANT_NAME = "admin"')
+ # This test will fail as soon as rc_file is fixed
+ self._test_source_envfile(
+ 'export "\'OS_TENANT_NAME\'" = "\'admin\'"')
+
@mock.patch('functest.ci.run_tests.Runner.get_run_dict',
return_value=None)
def test_run_tests_import_exception(self, *args):
self.runner.run_all()
self.assertTrue(mock_methods[1].called)
- @mock.patch('functest.utils.openstack_utils.source_credentials',
+ @mock.patch('functest.ci.run_tests.Runner.source_envfile',
side_effect=Exception)
@mock.patch('functest.ci.run_tests.Runner.summary')
def test_main_failed(self, *mock_methods):
mock_methods[1].assert_called_once_with(
'/home/opnfv/functest/conf/env_file')
- @mock.patch('functest.utils.openstack_utils.source_credentials')
+ @mock.patch('functest.ci.run_tests.Runner.source_envfile')
@mock.patch('functest.ci.run_tests.Runner.run_test',
return_value=TestCase.EX_OK)
@mock.patch('functest.ci.run_tests.Runner.summary')
run_tests.Result.EX_OK)
mock_methods[1].assert_called()
- @mock.patch('functest.utils.openstack_utils.source_credentials')
+ @mock.patch('functest.ci.run_tests.Runner.source_envfile')
@mock.patch('functest.ci.run_tests.Runner.run_test',
return_value=TestCase.EX_OK)
def test_main_test(self, *mock_methods):
run_tests.Result.EX_OK)
mock_methods[0].assert_called_once_with('test_name')
- @mock.patch('functest.utils.openstack_utils.source_credentials')
+ @mock.patch('functest.ci.run_tests.Runner.source_envfile')
@mock.patch('functest.ci.run_tests.Runner.run_all')
@mock.patch('functest.ci.run_tests.Runner.summary')
def test_main_all_tier(self, *args):
args[1].assert_called_once_with()
args[2].assert_called_once_with('/home/opnfv/functest/conf/env_file')
- @mock.patch('functest.utils.openstack_utils.source_credentials')
+ @mock.patch('functest.ci.run_tests.Runner.source_envfile')
def test_main_any_tier_test_ko(self, *args):
kwargs = {'get_tier.return_value': None,
'get_test.return_value': None}
import copy
import logging
-import os
import unittest
import mock
def test_get_credentials_missing_endpoint_type(self):
self._get_credentials_missing_env('OS_ENDPOINT_TYPE')
- def _test_source_credentials(self, msg, key='OS_TENANT_NAME',
- value='admin'):
- try:
- del os.environ[key]
- except:
- pass
- f = 'rc_file'
- with mock.patch('six.moves.builtins.open',
- mock.mock_open(read_data=msg),
- create=True) as m:
- m.return_value.__iter__ = lambda self: iter(self.readline, '')
- openstack_utils.source_credentials(f)
- m.assert_called_once_with(f, 'r')
- self.assertEqual(os.environ[key], value)
-
- def test_source_credentials(self):
- self._test_source_credentials('OS_TENANT_NAME=admin')
- self._test_source_credentials('OS_TENANT_NAME= admin')
- self._test_source_credentials('OS_TENANT_NAME = admin')
- self._test_source_credentials('OS_TENANT_NAME = "admin"')
- self._test_source_credentials('export OS_TENANT_NAME=admin')
- self._test_source_credentials('export OS_TENANT_NAME =admin')
- self._test_source_credentials('export OS_TENANT_NAME = admin')
- self._test_source_credentials('export OS_TENANT_NAME = "admin"')
- # This test will fail as soon as rc_file is fixed
- self._test_source_credentials(
- 'export "\'OS_TENANT_NAME\'" = "\'admin\'"')
-
@mock.patch('functest.utils.openstack_utils.os.getenv',
return_value=None)
def test_get_keystone_client_version_missing_env(self, mock_os_getenv):
import logging
import os.path
-import re
import sys
import time
return creds
-def source_credentials(rc_file):
- with open(rc_file, "r") as f:
- for line in f:
- var = (line.rstrip('"\n').replace('export ', '').split("=")
- if re.search(r'(.*)=(.*)', line) else None)
- # The two next lines should be modified as soon as rc_file
- # conforms with common rules. Be aware that it could induce
- # issues if value starts with '
- if var:
- key = re.sub(r'^["\' ]*|[ \'"]*$', '', var[0])
- value = re.sub(r'^["\' ]*|[ \'"]*$', '', "".join(var[1:]))
- os.environ[key] = value
- CONST.__setattr__(key, value)
-
-
def get_session_auth(other_creds={}):
loader = loading.get_plugin_loader('password')
creds = get_credentials(other_creds)