image_file_name: cirros-0.3.5-x86_64-lxc.tar.gz
image_disk_format: raw
- healthcheck:
- disk_image: /home/opnfv/functest/data/cirros-0.3.5-x86_64-lxc.tar.gz
- disk_format: raw
fdio:
general:
flavor_extra_specs: {'hw:mem_page_size':'large'}
import prettytable
import six
-import yaml
import functest.ci.tier_builder as tb
import functest.core.testcase as testcase
CONFIG_FUNCTEST_PATH = pkg_resources.resource_filename(
'functest', 'ci/config_functest.yaml')
-CONFIG_PATCH_PATH = pkg_resources.resource_filename(
- 'functest', 'ci/config_patch.yaml')
-CONFIG_AARCH64_PATCH_PATH = pkg_resources.resource_filename(
- 'functest', 'ci/config_aarch64_patch.yaml')
-# set the architecture to default
-pod_arch = os.getenv("POD_ARCH", None)
-arch_filter = ['aarch64']
class Result(enum.Enum):
CONST.__getattribute__('DEPLOY_SCENARIO'),
pkg_resources.resource_filename('functest', 'ci/testcases.yaml'))
- @staticmethod
- def update_config_file():
- Runner.patch_file(CONFIG_PATCH_PATH)
-
- if pod_arch and pod_arch in arch_filter:
- Runner.patch_file(CONFIG_AARCH64_PATCH_PATH)
-
- @staticmethod
- def patch_file(patch_file_path):
- logger.debug('Updating file: %s', patch_file_path)
- with open(patch_file_path) as f:
- patch_file = yaml.safe_load(f)
-
- updated = False
- for key in patch_file:
- if key in CONST.__getattribute__('DEPLOY_SCENARIO'):
- new_functest_yaml = dict(ft_utils.merge_dicts(
- ft_utils.get_functest_yaml(), patch_file[key]))
- updated = True
-
- if updated:
- os.remove(CONFIG_FUNCTEST_PATH)
- with open(CONFIG_FUNCTEST_PATH, "w") as f:
- f.write(yaml.dump(new_functest_yaml, default_style='"'))
-
@staticmethod
def source_rc_file():
rc_file = CONST.__getattribute__('openstack_creds')
self.run_tier(tier)
def main(self, **kwargs):
- Runner.update_config_file()
-
if 'noclean' in kwargs:
self.clean_flag = not kwargs['noclean']
if 'report' in kwargs:
self.run_tests_parser = run_tests.RunTestsParser()
- self.config_file_yaml = {'general': {
- 'openstack': {
- 'image_name': 'test_image_name'}},
- 'results': {
- 'test_db_url': 'url1'}}
- self.config_file_patch_yaml = {'fdio': {'general': {
- 'openstack': {
- 'image_name':
- 'test_image_name_2'}}}}
- self.config_file_aarcg64_patch_yaml = {'os': {'general': {
- 'openstack': {'image_name': 'test_image_name_3'}}}}
-
- @mock.patch('functest.ci.run_tests.Runner.patch_file')
- def test_update_config_file_default(self, mock_patch):
- self.runner.update_config_file()
- mock_patch.assert_called()
-
- def test_patch_file_missing_file(self):
- patch_file_path = "unexisting_file"
- with self.assertRaises(IOError):
- self.runner.patch_file(patch_file_path)
-
- @mock.patch('functest.ci.run_tests.ft_utils.merge_dicts')
- @mock.patch('functest.ci.run_tests.ft_utils.get_functest_yaml')
- def test_patch_file_default(self, *mock_methods):
- CONST.__setattr__('DEPLOY_SCENARIO', 'os-nosdn-nofeature-noha')
- with mock.patch(
- 'six.moves.builtins.open', mock.mock_open()), mock.patch(
- 'functest.ci.run_tests.yaml.safe_load') as yaml1, mock.patch(
- 'functest.ci.run_tests.ft_utils.get_functest_yaml') as yaml2:
- yaml1.return_value = self.config_file_patch_yaml
- yaml2.return_value = self.config_file_yaml
- self.runner.patch_file(yaml1)
- mock_methods[1].assert_not_called()
- mock_methods[0].assert_not_called()
-
- @mock.patch('functest.ci.run_tests.ft_utils.merge_dicts')
- @mock.patch('functest.ci.run_tests.os.remove')
- def test_patch_file_match_scenario(self, *mock_methods):
- CONST.__setattr__('DEPLOY_SCENARIO', 'os-nosdn-fdio-noha')
- with mock.patch(
- 'six.moves.builtins.open', mock.mock_open()), mock.patch(
- 'functest.ci.run_tests.yaml.safe_load') as yaml1, mock.patch(
- 'functest.ci.run_tests.ft_utils.get_functest_yaml') as yaml2:
- yaml1.return_value = self.config_file_patch_yaml
- yaml2.return_value = self.config_file_yaml
- self.runner.patch_file(yaml2)
- mock_methods[1].assert_called()
- mock_methods[0].assert_called()
-
@mock.patch('functest.ci.run_tests.logger.error')
def test_source_rc_file_missing_file(self, mock_logger_error):
with mock.patch('functest.ci.run_tests.os.path.isfile',
#!/usr/bin/env python
+# pylint: disable=missing-docstring
+
+import os
+import pkg_resources
import yaml
import six
-from functest.utils import env
-
class Config(object):
def __init__(self):
try:
- with open(env.ENV.CONFIG_FUNCTEST_YAML) as f:
- self.functest_yaml = yaml.safe_load(f)
- self._parse(None, self.functest_yaml)
+ with open(pkg_resources.resource_filename(
+ 'functest', 'ci/config_functest.yaml')) as yfile:
+ self.functest_yaml = yaml.safe_load(yfile)
except Exception as error:
raise Exception('Parse config failed: {}'.format(str(error)))
+ @staticmethod
+ def _merge_dicts(dict1, dict2):
+ for k in set(dict1.keys()).union(dict2.keys()):
+ if k in dict1 and k in dict2:
+ if isinstance(dict1[k], dict) and isinstance(dict2[k], dict):
+ yield (k, dict(Config._merge_dicts(dict1[k], dict2[k])))
+ else:
+ yield (k, dict2[k])
+ elif k in dict1:
+ yield (k, dict1[k])
+ else:
+ yield (k, dict2[k])
+
+ def patch_file(self, patch_file_path):
+ with open(patch_file_path) as yfile:
+ patch_file = yaml.safe_load(yfile)
+
+ for key in patch_file:
+ if key in os.environ.get('DEPLOY_SCENARIO', ""):
+ self.functest_yaml = dict(Config._merge_dicts(
+ self.functest_yaml, patch_file[key]))
+
def _parse(self, attr_now, left_parametes):
for param_n, param_v in six.iteritems(left_parametes):
attr_further = self._get_attr_further(attr_now, param_n)
if isinstance(param_v, dict):
self._parse(attr_further, param_v)
- def _get_attr_further(self, attr_now, next):
+ @staticmethod
+ def _get_attr_further(attr_now, next): # pylint: disable=redefined-builtin
return attr_now if next == 'general' else (
'{}_{}'.format(attr_now, next) if attr_now else next)
+ def fill(self):
+ try:
+ self._parse(None, self.functest_yaml)
+ except Exception as error:
+ raise Exception('Parse config failed: {}'.format(str(error)))
+
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']:
+ CONF.patch_file(pkg_resources.resource_filename(
+ 'functest', 'ci/config_aarch64_patch.yaml'))
+CONF.fill()
return get_parameter_from_yaml(parameter, yaml_)
-def merge_dicts(dict1, dict2):
- for k in set(dict1.keys()).union(dict2.keys()):
- if k in dict1 and k in dict2:
- if isinstance(dict1[k], dict) and isinstance(dict2[k], dict):
- yield (k, dict(merge_dicts(dict1[k], dict2[k])))
- else:
- yield (k, dict2[k])
- elif k in dict1:
- yield (k, dict1[k])
- else:
- yield (k, dict2[k])
-
-
def get_functest_yaml():
with open(constants.CONST.__getattribute__('CONFIG_FUNCTEST_YAML')) as f:
functest_yaml = yaml.safe_load(f)
Code Review / functest-xtesting.git / commitdiff