from functest.core import tenantnetwork
from functest.utils import config
from functest.utils import env
+from functest.utils import functest_utils
class VmReady1(tenantnetwork.TenantNetwork1):
extra_properties = self.extra_properties.copy()
if env.get('IMAGE_PROPERTIES'):
extra_properties.update(
- dict((k.strip(), v.strip()) for k, v in (
- item.split(': ') for item in env.get(
- 'IMAGE_PROPERTIES').split(','))))
+ functest_utils.convert_ini_to_dict(
+ env.get('IMAGE_PROPERTIES')))
extra_properties.update(
getattr(config.CONF, '{}_extra_properties'.format(
self.case_name), {}))
extra_alt_properties = self.extra_alt_properties.copy()
if env.get('IMAGE_PROPERTIES'):
extra_alt_properties.update(
- dict((k.strip(), v.strip()) for k, v in (
- item.split(': ') for item in env.get(
- 'IMAGE_PROPERTIES').split(','))))
+ functest_utils.convert_ini_to_dict(
+ env.get('IMAGE_PROPERTIES')))
extra_alt_properties.update(
getattr(config.CONF, '{}_extra_alt_properties'.format(
self.case_name), {}))
import subprocess
import pkg_resources
-import six
from six.moves import configparser
import yaml
from functest.utils import config
from functest.utils import env
+from functest.utils import functest_utils
RALLY_CONF_PATH = "/etc/rally/rally.conf"
'ServerGroupAntiAffinityFilter', 'ServerGroupAffinityFilter']
rconfig.set(
'compute-feature-enabled', 'scheduler_available_filters',
- convert_list_to_ini(filters))
+ functest_utils.convert_list_to_ini(filters))
if os.environ.get('OS_REGION_NAME'):
rconfig.set('identity', 'region', os.environ.get('OS_REGION_NAME'))
if env.get("NEW_USER_ROLE").lower() != "member":
rconfig.set(
'auth', 'tempest_roles',
- convert_list_to_ini([env.get("NEW_USER_ROLE")]))
+ functest_utils.convert_list_to_ini([env.get("NEW_USER_ROLE")]))
if not json.loads(env.get("USE_DYNAMIC_CREDENTIALS").lower()):
rconfig.set('auth', 'use_dynamic_credentials', False)
account_file = os.path.join(
tempest_conf_file)
return None
return tempest_conf_file
-
-
-def convert_dict_to_ini(value):
- "Convert dict to oslo.conf input"
- assert isinstance(value, dict)
- return ",".join("{}:{}".format(
- key, val) for (key, val) in six.iteritems(value))
-
-
-def convert_list_to_ini(value):
- "Convert list to oslo.conf input"
- assert isinstance(value, list)
- return ",".join("{}".format(val) for val in value)
from functest.opnfv_tests.openstack.tempest import conf_utils
from functest.utils import config
from functest.utils import env
+from functest.utils import functest_utils
LOGGER = logging.getLogger(__name__)
extra_properties = self.extra_properties.copy()
if env.get('IMAGE_PROPERTIES'):
extra_properties.update(
- dict((k.strip(), v.strip()) for k, v in (
- item.split(': ') for item in env.get(
- 'IMAGE_PROPERTIES').split(','))))
+ functest_utils.convert_ini_to_dict(
+ env.get('IMAGE_PROPERTIES')))
extra_properties.update(
getattr(config.CONF, '{}_extra_properties'.format(
self.case_name), {}))
rconfig.set(
'scenario', 'img_properties',
- conf_utils.convert_dict_to_ini(extra_properties))
+ functest_utils.convert_dict_to_ini(extra_properties))
with open(self.conf_file, 'wb') as config_file:
rconfig.write(config_file)
['rally', 'verify', 'configure-verifier', '--reconfigure',
'--id', str(getattr(config.CONF, 'tempest_verifier_name'))])
- def test_convert_dict_to_ini(self):
- self.assertEqual(
- conf_utils.convert_dict_to_ini({}), "")
- self.assertEqual(
- conf_utils.convert_dict_to_ini({"a": "b"}), "a:b")
- value = conf_utils.convert_dict_to_ini({"a": "b", "c": "d"})
- self.assertTrue(value == "a:b,c:d" or value == "c:d,a:b")
- with self.assertRaises(AssertionError):
- conf_utils.convert_list_to_ini("")
-
- def test_convert_list_to_ini(self):
- self.assertEqual(
- conf_utils.convert_list_to_ini([]), "")
- self.assertEqual(
- conf_utils.convert_list_to_ini(["a"]), "a")
- self.assertEqual(
- conf_utils.convert_list_to_ini(["a", "b"]), "a,b")
- with self.assertRaises(AssertionError):
- conf_utils.convert_list_to_ini("")
-
if __name__ == "__main__":
logging.disable(logging.CRITICAL)
functest_utils.get_openstack_version(cloud), "Unknown")
args[0].assert_called_once_with(cloud)
+ def test_convert_dict_to_ini(self):
+ self.assertEqual(
+ functest_utils.convert_dict_to_ini({}), "")
+ self.assertEqual(
+ functest_utils.convert_dict_to_ini({"a": "b"}), "a:b")
+ value = functest_utils.convert_dict_to_ini({"a": "b", "c": "d"})
+ self.assertTrue(value == "a:b,c:d" or value == "c:d,a:b")
+ with self.assertRaises(AssertionError):
+ functest_utils.convert_list_to_ini("")
+
+ def test_convert_list_to_ini(self):
+ self.assertEqual(
+ functest_utils.convert_list_to_ini([]), "")
+ self.assertEqual(
+ functest_utils.convert_list_to_ini(["a"]), "a")
+ self.assertEqual(
+ functest_utils.convert_list_to_ini(["a", "b"]), "a,b")
+ with self.assertRaises(AssertionError):
+ functest_utils.convert_list_to_ini("")
+
+ def test_convert_ini_to_dict(self):
+ self.assertEqual(
+ functest_utils.convert_ini_to_dict(""), {})
+ self.assertEqual(
+ functest_utils.convert_ini_to_dict("a:b"), {"a": "b"})
+ self.assertEqual(
+ functest_utils.convert_ini_to_dict(
+ "a:b,c:d"), {"a": "b", "c": "d"})
+ with self.assertRaises(AssertionError):
+ functest_utils.convert_list_to_ini({})
+
+ def test_convert_ini_to_list(self):
+ self.assertEqual(
+ functest_utils.convert_ini_to_list(""), [])
+ self.assertEqual(
+ functest_utils.convert_ini_to_list("a"), ["a"])
+ self.assertEqual(
+ functest_utils.convert_ini_to_list("a,b"), ["a", "b"])
+ with self.assertRaises(AssertionError):
+ functest_utils.convert_ini_to_list([])
+
if __name__ == "__main__":
logging.disable(logging.CRITICAL)
import sys
import yaml
+import six
+
LOGGER = logging.getLogger(__name__)
except AssertionError:
LOGGER.exception("Cannot detect OpenStack version")
return "Unknown"
+
+
+def convert_dict_to_ini(value):
+ "Convert dict to oslo.conf input"
+ assert isinstance(value, dict)
+ return ",".join("{}:{}".format(
+ key, val) for (key, val) in six.iteritems(value))
+
+
+def convert_list_to_ini(value):
+ "Convert list to oslo.conf input"
+ assert isinstance(value, list)
+ return ",".join("{}".format(val) for val in value)
+
+
+def convert_ini_to_dict(value):
+ "Convert oslo.conf input to dict"
+ assert isinstance(value, str)
+ try:
+ return {k: v for k, v in (x.split(':') for x in value.split(','))}
+ except ValueError:
+ return {}
+
+
+def convert_ini_to_list(value):
+ "Convert list to oslo.conf input"
+ assert isinstance(value, str)
+ if not value:
+ return []
+ return [x for x in value.split(',')]