Several environement variables may be specified:
* INSTALLER_TYPE=(apex|compass|daisy|fuel|joid)
+ * INSTALLER_IP=<Specific IP Address>
* DEPLOY_SCENARIO=<vim>-<controller>-<nfv_feature>-<ha_mode>
+INSTALLER_IP is required by Barometer in order to access the installer node and
+the deployment.
+
The format for the DEPLOY_SCENARIO env variable can be described as follows:
* vim: (os|k8s) = OpenStack or Kubernetes
* controller is one of ( nosdn | odl )
Resources to handle testcase related requests
"""
+import ConfigParser
import logging
import os
import re
-import pkg_resources
import socket
import uuid
-import ConfigParser
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, thread
from functest.cli.commands.cli_testcase import Testcase
from functest.api.database.v1.handlers import TasksHandler
from functest.utils.constants import CONST
+from functest.utils import env
import functest.utils.functest_utils as ft_utils
LOGGER = logging.getLogger(__name__)
result = 'FAIL'
env_info = {
- 'installer': os.environ.get('INSTALLER_TYPE', None),
- 'scenario': os.environ.get('DEPLOY_SCENARIO', None),
- 'build_tag': os.environ.get('BUILD_TAG', None),
- 'ci_loop': os.environ.get('CI_LOOP', 'daily')
+ 'installer': env.get('INSTALLER_TYPE'),
+ 'scenario': env.get('DEPLOY_SCENARIO'),
+ 'build_tag': env.get('BUILD_TAG'),
+ 'ci_loop': env.get('CI_LOOP')
}
result = {
'task_id': args.get('task_id'),
import prettytable
import yaml
-import functest.ci.tier_builder as tb
-import functest.core.testcase as testcase
+from functest.ci import tier_builder
+from functest.core import testcase
+from functest.utils import env
LOGGER = logging.getLogger('functest.ci.run_tests')
ENV_FILE = "/home/opnfv/functest/conf/env_file"
self.overall_result = Result.EX_OK
self.clean_flag = True
self.report_flag = False
- self.tiers = tb.TierBuilder(
- os.environ.get('INSTALLER_TYPE', None),
- os.environ.get('DEPLOY_SCENARIO', None),
+ self.tiers = tier_builder.TierBuilder(
+ env.get('INSTALLER_TYPE'),
+ env.get('DEPLOY_SCENARIO'),
pkg_resources.resource_filename('functest', 'ci/testcases.yaml'))
@staticmethod
field_names=['tiers', 'order', 'CI Loop', 'description',
'testcases'])
for tier in self.tiers.get_tiers():
- ci_loop = os.environ.get('CI_LOOP', 'daily')
+ ci_loop = env.get('CI_LOOP')
if (tier.get_tests() and
re.search(ci_loop, tier.get_ci_loop()) is not None):
tiers_to_run.append(tier)
LOGGER.error("Unknown test case or tier '%s', or not "
"supported by the given scenario '%s'.",
kwargs['test'],
- os.environ.get('DEPLOY_SCENARIO', ""))
+ env.get('DEPLOY_SCENARIO'))
LOGGER.debug("Available tiers are:\n\n%s",
self.tiers)
return Result.EX_ERROR
field_names=['env var', 'value'])
for env_var in ['INSTALLER_TYPE', 'DEPLOY_SCENARIO', 'BUILD_TAG',
'CI_LOOP']:
- msg.add_row([env_var, os.environ.get(env_var, "")])
+ msg.add_row([env_var, env.get(env_var)])
LOGGER.info("Deployment description:\n\n%s\n", msg)
msg = prettytable.PrettyTable(
header_style='upper', padding_width=5,
# pylint: disable=missing-docstring
-import os
-
import click
import prettytable
import six
+from functest.utils import env
+
class Env(object): # pylint: disable=too-few-public-methods
@staticmethod
def show():
- install_type = os.environ.get('INSTALLER_TYPE', 'Unknown')
- scenario = os.environ.get('DEPLOY_SCENARIO', 'Unknown')
- node = os.environ.get('NODE_NAME', 'Unknown')
- build_tag = os.environ.get('BUILD_TAG', None)
+ install_type = env.get('INSTALLER_TYPE')
+ scenario = env.get('DEPLOY_SCENARIO')
+ node = env.get('NODE_NAME')
+ build_tag = env.get('BUILD_TAG')
if build_tag:
build_tag = build_tag.lstrip(
"jenkins-").lstrip("functest").lstrip("-")
# pylint: disable=missing-docstring
-import os
import pkg_resources
import click
from functest.ci import tier_builder
from functest.utils import functest_utils
+from functest.utils import env
class Tier(object):
def __init__(self):
self.tiers = tier_builder.TierBuilder(
- os.environ['INSTALLER_TYPE'],
- os.environ['DEPLOY_SCENARIO'],
+ env.get('INSTALLER_TYPE'),
+ env.get('DEPLOY_SCENARIO'),
pkg_resources.resource_filename('functest', 'ci/testcases.yaml'))
def list(self):
import requests
from functest.utils import decorators
+from functest.utils import env
import prettytable
assert self.case_name
assert self.start_time
assert self.stop_time
- url = os.environ['TEST_DB_URL']
+ url = env.get('TEST_DB_URL')
data = {"project_name": self.project_name,
"case_name": self.case_name,
"details": self.details}
- data["installer"] = os.environ['INSTALLER_TYPE']
- data["scenario"] = os.environ['DEPLOY_SCENARIO']
- data["pod_name"] = os.environ['NODE_NAME']
- data["build_tag"] = os.environ['BUILD_TAG']
+ data["installer"] = env.get('INSTALLER_TYPE')
+ data["scenario"] = env.get('DEPLOY_SCENARIO')
+ data["pod_name"] = env.get('NODE_NAME')
+ data["build_tag"] = env.get('BUILD_TAG')
data["criteria"] = 'PASS' if self.is_successful(
) == TestCase.EX_OK else 'FAIL'
data["start_date"] = datetime.fromtimestamp(
try:
data["version"] = re.search(
TestCase._job_name_rule,
- os.environ['BUILD_TAG']).group(2)
+ env.get('BUILD_TAG')).group(2)
except Exception: # pylint: disable=broad-except
data["version"] = "unknown"
req = requests.post(
self.__logger.exception(
"Please run test before publishing the results")
return TestCase.EX_PUSH_TO_DB_ERROR
- except KeyError as exc:
- self.__logger.error("Please set env var: " + str(exc))
- return TestCase.EX_PUSH_TO_DB_ERROR
except requests.exceptions.HTTPError:
self.__logger.exception("The HTTP request raises issues")
return TestCase.EX_PUSH_TO_DB_ERROR
import json
import logging
-import os
import traceback
from functools import wraps
import requests
from six.moves import urllib
+from functest.utils import env
+
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:
- assert os.environ['NODE_NAME']
- assert os.environ["ENERGY_RECORDER_API_URL"]
- environment = os.environ['NODE_NAME']
- energy_recorder_uri = os.environ["ENERGY_RECORDER_API_URL"]
+ assert env.get('NODE_NAME')
+ assert env.get('ENERGY_RECORDER_API_URL')
+ environment = env.get('NODE_NAME')
+ energy_recorder_uri = env.get(
+ 'ENERGY_RECORDER_API_URL')
# Creds
- creds_usr = os.environ.get("ENERGY_RECORDER_API_USER", "")
- creds_pass = os.environ.get("ENERGY_RECORDER_API_PASSWORD", "")
+ creds_usr = env.get("ENERGY_RECORDER_API_USER")
+ creds_pass = env.get("ENERGY_RECORDER_API_PASSWORD")
uri_comp = "/recorders/environment/"
uri_comp += urllib.parse.quote_plus(environment)
from functest.opnfv_tests.openstack.snaps import snaps_utils
from functest.opnfv_tests.openstack.tempest import conf_utils
from functest.utils.constants import CONST
+from functest.utils import env
from snaps.config.flavor import FlavorConfig
from snaps.config.image import ImageConfig
with open(RallyBase.BLACKLIST_FILE, 'r') as black_list_file:
black_list_yaml = yaml.safe_load(black_list_file)
- installer_type = os.getenv('INSTALLER_TYPE', None)
- deploy_scenario = os.getenv('DEPLOY_SCENARIO', None)
+ installer_type = env.get('INSTALLER_TYPE')
+ deploy_scenario = env.get('DEPLOY_SCENARIO')
if (bool(installer_type) and bool(deploy_scenario) and
'scenario' in black_list_yaml.keys()):
for item in black_list_yaml['scenario']:
from functest.core import unit
from functest.opnfv_tests.openstack.snaps import snaps_utils
from functest.utils.constants import CONST
+from functest.utils import env
from snaps.openstack import create_flavor
CONST.__getattribute__('snaps_use_floating_ips') == 'True')
self.use_keystone = (
CONST.__getattribute__('snaps_use_keystone') == 'True')
- scenario = CONST.__getattribute__('DEPLOY_SCENARIO')
+ scenario = env.get('DEPLOY_SCENARIO')
self.flavor_metadata = None
if 'ovs' in scenario or 'fdio' in scenario:
"""Some common utils wrapping snaps functions """
from functest.utils.constants import CONST
+from functest.utils import env
from snaps.openstack.tests import openstack_tests
from snaps.openstack.utils import neutron_utils, nova_utils
"""
neutron = neutron_utils.neutron_client(os_creds)
ext_nets = neutron_utils.get_external_networks(neutron)
- if hasattr(CONST, 'EXTERNAL_NETWORK'):
- extnet_config = CONST.__getattribute__('EXTERNAL_NETWORK')
+ if env.get('EXTERNAL_NETWORK'):
+ extnet_config = env.get('EXTERNAL_NETWORK')
for ext_net in ext_nets:
if ext_net.name == extnet_config:
return extnet_config
import logging
import fileinput
import os
-import pkg_resources
import shutil
import subprocess
+import pkg_resources
import yaml
from functest.utils.constants import CONST
+from functest.utils import env
import functest.utils.functest_utils as ft_utils
'functest',
'opnfv_tests/openstack/tempest/custom_tests/test_accounts.yaml')
-CI_INSTALLER_TYPE = CONST.__getattribute__('INSTALLER_TYPE')
+CI_INSTALLER_TYPE = env.get('INSTALLER_TYPE')
""" logging configuration """
LOGGER = logging.getLogger(__name__)
def create_rally_deployment():
"""Create new rally deployment"""
# set the architecture to default
- pod_arch = os.getenv("POD_ARCH", None)
+ pod_arch = env.get("POD_ARCH")
arch_filter = ['aarch64']
if pod_arch and pod_arch in arch_filter:
config.set('compute', 'min_compute_nodes', compute_cnt)
config.set('compute-feature-enabled', 'live_migration', True)
- config.set('identity', 'region',
- CONST.__getattribute__('OS_REGION_NAME'))
- identity_api_version = os.getenv(
- "OS_IDENTITY_API_VERSION", os.getenv("IDENTITY_API_VERSION"))
+ config.set('identity', 'region', os.environ.get('OS_REGION_NAME'))
+ identity_api_version = os.environ.get(
+ "OS_IDENTITY_API_VERSION", os.environ.get("IDENTITY_API_VERSION"))
if identity_api_version == '3':
auth_version = 'v3'
config.set('identity-feature-enabled', 'api_v2', False)
config.set('object-storage', 'operator_role',
CONST.__getattribute__('tempest_object_storage_operator_role'))
- if CONST.__getattribute__('OS_ENDPOINT_TYPE') is not None:
+ if os.environ.get('OS_ENDPOINT_TYPE') is not None:
config.set('identity', 'v3_endpoint_type',
- CONST.__getattribute__('OS_ENDPOINT_TYPE'))
+ os.environ.get('OS_ENDPOINT_TYPE'))
- if CONST.__getattribute__('OS_ENDPOINT_TYPE') is not None:
+ if os.environ.get('OS_ENDPOINT_TYPE') is not None:
sections = config.sections()
services_list = ['compute',
'volume',
if service not in sections:
config.add_section(service)
config.set(service, 'endpoint_type',
- CONST.__getattribute__('OS_ENDPOINT_TYPE'))
+ os.environ.get('OS_ENDPOINT_TYPE'))
LOGGER.debug('Add/Update required params defined in tempest_conf.yaml '
'into tempest.conf file')
from functest.opnfv_tests.openstack.snaps import snaps_utils
from functest.opnfv_tests.openstack.tempest import conf_utils
from functest.utils.constants import CONST
+from functest.utils import env
import functest.utils.functest_utils as ft_utils
from snaps.config.flavor import FlavorConfig
result_file = open(conf_utils.TEMPEST_LIST, 'w')
black_tests = []
try:
- installer_type = CONST.__getattribute__('INSTALLER_TYPE')
- deploy_scenario = CONST.__getattribute__('DEPLOY_SCENARIO')
+ installer_type = env.get('INSTALLER_TYPE')
+ deploy_scenario = env.get('DEPLOY_SCENARIO')
if bool(installer_type) * bool(deploy_scenario):
# if INSTALLER_TYPE and DEPLOY_SCENARIO are set we read the
# file
def _create_flavor(self, name):
"""Create flavor for tests."""
- scenario = CONST.__getattribute__('DEPLOY_SCENARIO')
+ scenario = env.get('DEPLOY_SCENARIO')
flavor_metadata = None
if 'ovs' in scenario or 'fdio' in scenario:
flavor_metadata = create_flavor.MEM_PAGE_SIZE_LARGE
if use_custom_flavors:
LOGGER.info("Creating 2nd flavor for Tempest suite")
- scenario = CONST.__getattribute__('DEPLOY_SCENARIO')
+ scenario = env.get('DEPLOY_SCENARIO')
if 'ovs' in scenario or 'fdio' in scenario:
CONST.__setattr__('openstack_flavor_ram', 1024)
name = CONST.__getattribute__(
from functest.core import testcase
from functest.opnfv_tests.openstack.snaps import snaps_utils
from functest.utils.constants import CONST
+from functest.utils import env
from snaps.config.flavor import FlavorConfig
from snaps.config.network import NetworkConfig, SubnetConfig
self.logger.info(
"Creating flavor with name: '%s'", self.flavor_name)
- scenario = getattr(CONST, 'DEPLOY_SCENARIO')
+ scenario = env.get('DEPLOY_SCENARIO')
flavor_metadata = None
flavor_ram = 512
if 'ovs' in scenario or 'fdio' in scenario:
from functest.core import robotframework
from functest.opnfv_tests.openstack.snaps import snaps_utils
from functest.utils import constants
+from functest.utils import env
__author__ = "Cedric Ollivier <cedric.ollivier@orange.com>"
kwargs['odlrestconfport'] = '8181'
kwargs['odlusername'] = 'admin'
kwargs['odlpassword'] = 'admin'
- installer_type = None
- if 'INSTALLER_TYPE' in os.environ:
- installer_type = os.environ['INSTALLER_TYPE']
+ installer_type = env.get('INSTALLER_TYPE')
kwargs['osusername'] = os.environ['OS_USERNAME']
kwargs['osuserdomainname'] = os.environ.get(
'OS_USER_DOMAIN_NAME', 'Default')
kwargs['odlwebport'] = '8181'
kwargs['odlrestconfport'] = '8282'
elif installer_type == 'apex' or installer_type == 'netvirt':
- kwargs['odlip'] = os.environ['SDN_CONTROLLER_IP']
+ kwargs['odlip'] = env.get('SDN_CONTROLLER_IP')
kwargs['odlwebport'] = '8081'
kwargs['odlrestconfport'] = '8081'
elif installer_type == 'compass':
kwargs['odlrestconfport'] = '8080'
elif installer_type == 'daisy':
- kwargs['odlip'] = os.environ['SDN_CONTROLLER_IP']
+ kwargs['odlip'] = env.get('SDN_CONTROLLER_IP')
kwargs['odlwebport'] = '8181'
kwargs['odlrestconfport'] = '8087'
else:
- kwargs['odlip'] = os.environ['SDN_CONTROLLER_IP']
+ kwargs['odlip'] = env.get('SDN_CONTROLLER_IP')
except KeyError as ex:
self.__logger.error("Cannot run ODL testcases. "
"Please check env var: "
# pylint: disable=missing-docstring
import logging
+import os
import unittest
import mock
from functest.core import testcase
from functest.opnfv_tests.openstack.tempest import tempest
from functest.opnfv_tests.openstack.tempest import conf_utils
-from functest.utils.constants import CONST
from snaps.openstack.os_credentials import OSCreds
mock.patch.object(self.tempestcommon, 'read_file',
return_value=['test1', 'test2']):
conf_utils.TEMPEST_BLACKLIST = Exception
- CONST.__setattr__('INSTALLER_TYPE', 'installer_type')
- CONST.__setattr__('DEPLOY_SCENARIO', 'deploy_scenario')
+ os.environ['INSTALLER_TYPE'] = 'installer_type'
+ os.environ['DEPLOY_SCENARIO'] = 'deploy_scenario'
self.tempestcommon.apply_tempest_blacklist()
obj = mock_open()
obj.write.assert_any_call('test1\n')
return_value=['test1', 'test2']), \
mock.patch('functest.opnfv_tests.openstack.tempest.tempest.'
'yaml.safe_load', return_value=item_dict):
- CONST.__setattr__('INSTALLER_TYPE', 'installer_type')
- CONST.__setattr__('DEPLOY_SCENARIO', 'deploy_scenario')
+ os.environ['INSTALLER_TYPE'] = 'installer_type'
+ os.environ['DEPLOY_SCENARIO'] = 'deploy_scenario'
self.tempestcommon.apply_tempest_blacklist()
obj = mock_open()
obj.write.assert_any_call('test1\n')
--- /dev/null
+#!/usr/bin/env python
+
+# Copyright (c) 2018 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
+# http://www.apache.org/licenses/LICENSE-2.0
+
+# pylint: disable=missing-docstring
+
+import logging
+import os
+import unittest
+
+from six.moves import reload_module
+
+from functest.utils import env
+from functest.utils import constants
+
+
+class EnvTesting(unittest.TestCase):
+ # pylint: disable=missing-docstring
+
+ def setUp(self):
+ os.environ['FOO'] = 'foo'
+ os.environ['BUILD_TAG'] = 'master'
+ os.environ['CI_LOOP'] = 'weekly'
+
+ def test_get_unset_unknown_env(self):
+ del os.environ['FOO']
+ self.assertEqual(env.get('FOO'), None)
+ # Backward compatibilty (waiting for SDNVPN and SFC)
+ reload_module(env)
+ with self.assertRaises(AttributeError):
+ getattr(env.ENV, 'FOO')
+ reload_module(constants)
+ with self.assertRaises(AttributeError):
+ getattr(constants.CONST, 'FOO')
+
+ def test_get_unknown_env(self):
+ self.assertEqual(env.get('FOO'), 'foo')
+ reload_module(env)
+ # Backward compatibilty (waiting for SDNVPN and SFC)
+ with self.assertRaises(AttributeError):
+ getattr(env.ENV, 'FOO')
+ reload_module(constants)
+ with self.assertRaises(AttributeError):
+ getattr(constants.CONST, 'FOO')
+
+ def test_get_unset_env(self):
+ del os.environ['CI_LOOP']
+ self.assertEqual(
+ env.get('CI_LOOP'), env.INPUTS['CI_LOOP'])
+ # Backward compatibilty (waiting for SDNVPN and SFC)
+ reload_module(env)
+ self.assertEqual(
+ getattr(env.ENV, 'CI_LOOP'), env.INPUTS['CI_LOOP'])
+ reload_module(constants)
+ self.assertEqual(
+ getattr(constants.CONST, 'CI_LOOP'),
+ env.INPUTS['CI_LOOP'])
+
+ def test_get_env(self):
+ self.assertEqual(
+ env.get('CI_LOOP'), 'weekly')
+ # Backward compatibilty (waiting for SDNVPN and SFC)
+ reload_module(env)
+ self.assertEqual(getattr(env.ENV, 'CI_LOOP'), 'weekly')
+ reload_module(constants)
+ self.assertEqual(getattr(constants.CONST, 'CI_LOOP'), 'weekly')
+
+ def test_get_unset_env2(self):
+ del os.environ['BUILD_TAG']
+ self.assertEqual(
+ env.get('BUILD_TAG'), env.INPUTS['BUILD_TAG'])
+ # Backward compatibilty (waiting for SDNVPN and SFC)
+ reload_module(env)
+ self.assertEqual(
+ getattr(env.ENV, 'BUILD_TAG'), env.INPUTS['BUILD_TAG'])
+ reload_module(constants)
+ self.assertEqual(
+ getattr(constants.CONST, 'BUILD_TAG'), env.INPUTS['BUILD_TAG'])
+
+ def test_get_env2(self):
+ self.assertEqual(env.get('BUILD_TAG'), 'master')
+ # Backward compatibilty (waiting for SDNVPN and SFC)
+ reload_module(env)
+ self.assertEqual(getattr(env.ENV, 'BUILD_TAG'), 'master')
+ reload_module(env)
+ self.assertEqual(getattr(constants.CONST, 'BUILD_TAG'), 'master')
+
+
+if __name__ == "__main__":
+ logging.disable(logging.CRITICAL)
+ unittest.main(verbosity=2)
self.assertEqual(functest_utils.get_resolvconf_ns(),
self.test_ip[1:])
- def _get_environ(self, var):
+ def _get_environ(self, var, *args): # pylint: disable=unused-argument
if var == 'INSTALLER_TYPE':
return self.installer
elif var == 'DEPLOY_SCENARIO':
# pylint: disable=missing-docstring
-import os
import pkg_resources
import yaml
import six
+from functest.utils import env
+
class Config(object):
def __init__(self):
patch_file = yaml.safe_load(yfile)
for key in patch_file:
- if key in os.environ.get('DEPLOY_SCENARIO', ""):
+ if key in env.get('DEPLOY_SCENARIO'):
self.functest_yaml = dict(Config._merge_dicts(
self.functest_yaml, patch_file[key]))
CONF = Config()
CONF.patch_file(pkg_resources.resource_filename(
'functest', 'ci/config_patch.yaml'))
-if os.getenv("POD_ARCH", None) and os.getenv("POD_ARCH", None) in ['aarch64']:
+if env.get("POD_ARCH") in ['aarch64']:
CONF.patch_file(pkg_resources.resource_filename(
'functest', 'ci/config_aarch64_patch.yaml'))
CONF.fill()
# pylint: disable=missing-docstring
+import pkg_resources
import six
from functest.utils import config
class Constants(object): # pylint: disable=too-few-public-methods
+ CONFIG_FUNCTEST_YAML = pkg_resources.resource_filename(
+ 'functest', 'ci/config_functest.yaml')
+
def __init__(self):
for attr_n, attr_v in six.iteritems(config.CONF.__dict__):
setattr(self, attr_n, attr_v)
#!/usr/bin/env python
+# Copyright (c) 2018 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
+# http://www.apache.org/licenses/LICENSE-2.0
+
# pylint: disable=missing-docstring
import os
-import re
-import pkg_resources
import six
+INPUTS = {
+ 'EXTERNAL_NETWORK': None,
+ 'CI_LOOP': 'daily',
+ 'DEPLOY_SCENARIO': 'os-nosdn-nofeature-noha',
+ 'INSTALLER_TYPE': None,
+ 'SDN_CONTROLLER_IP': None,
+ 'BUILD_TAG': None,
+ 'NODE_NAME': None,
+ 'POD_ARCH': None,
+ 'TEST_DB_URL': 'http://testresults.opnfv.org/test/api/v1/results',
+ 'ENERGY_RECORDER_API_URL': 'http://energy.opnfv.fr/resources',
+ 'ENERGY_RECORDER_API_USER': '',
+ 'ENERGY_RECORDER_API_PASSWORD': ''
+}
-class Environment(object): # pylint: disable=too-few-public-methods
- default_envs = {
- 'NODE_NAME': 'unknown_pod',
- 'DEPLOY_SCENARIO': 'os-nosdn-nofeature-noha',
- 'DEPLOY_TYPE': 'virt',
- 'INSTALLER_TYPE': None,
- 'BUILD_TAG': None,
- 'OS_ENDPOINT_TYPE': None,
- 'OS_AUTH_URL': None,
- 'CONFIG_FUNCTEST_YAML': pkg_resources.resource_filename(
- 'functest', 'ci/config_functest.yaml'),
- 'OS_INSECURE': '',
- 'OS_REGION_NAME': 'RegionOne'
- }
+def get(env_var):
+ if env_var not in INPUTS.keys():
+ return os.environ.get(env_var, None)
+ return os.environ.get(env_var, INPUTS[env_var])
- def __init__(self):
- for key, value in six.iteritems(os.environ):
- setattr(self, key, value)
- for key, value in six.iteritems(self.default_envs):
- if key not in os.environ:
- setattr(self, key, value)
- if 'CI_LOOP' not in os.environ:
- self._set_ci_loop()
-
- def _set_ci_loop(self):
- if (getattr(self, "BUILD_TAG") and
- re.search("daily", getattr(self, "BUILD_TAG"))):
- setattr(self, "CI_LOOP", "daily")
- else:
- setattr(self, "CI_LOOP", "weekly")
+class Environment(object): # pylint: disable=too-few-public-methods
+
+ # Backward compatibilty (waiting for SDNVPN and SFC)
+ def __init__(self):
+ for key, _ in six.iteritems(INPUTS):
+ setattr(self, key, get(key))
+# Backward compatibilty (waiting for SDNVPN and SFC)
ENV = Environment()
from __future__ import print_function
import logging
-import os
import re
import shutil
import subprocess
import yaml
from functest.utils import constants
+from functest.utils import env
LOGGER = logging.getLogger(__name__)
Get the CI env variables
"""
ci_env_var = {
- "installer": os.environ.get('INSTALLER_TYPE'),
- "scenario": os.environ.get('DEPLOY_SCENARIO')}
+ "installer": env.get('INSTALLER_TYPE'),
+ "scenario": env.get('DEPLOY_SCENARIO')}
return ci_env_var
from keystoneclient import client as keystoneclient
from neutronclient.neutron import client as neutronclient
-from functest.utils.constants import CONST
+from functest.utils import env
import functest.utils.functest_utils as ft_utils
logger = logging.getLogger(__name__)
def get_external_net(neutron_client):
- if (hasattr(CONST, 'EXTERNAL_NETWORK')):
- return CONST.__getattribute__('EXTERNAL_NETWORK')
+ if (env.get('EXTERNAL_NETWORK')):
+ return env.get('EXTERNAL_NETWORK')
for network in neutron_client.list_networks()['networks']:
if network['router:external']:
return network['name']
def get_external_net_id(neutron_client):
- if (hasattr(CONST, 'EXTERNAL_NETWORK')):
+ if (env.get('EXTERNAL_NETWORK')):
networks = neutron_client.list_networks(
- name=CONST.__getattribute__('EXTERNAL_NETWORK'))
+ name=env.get('EXTERNAL_NETWORK'))
net_id = networks['networks'][0]['id']
return net_id
for network in neutron_client.list_networks()['networks']:
Code Review / functest-xtesting.git / commitdiff