if not rconfig.has_section('image-feature-enabled'):
rconfig.add_section('image-feature-enabled')
rconfig.set('image-feature-enabled', 'api_v1', False)
- with open(self.conf_file, 'wb') as config_file:
+ with open(self.conf_file, 'w') as config_file:
rconfig.write(config_file)
self.backup_tempest_config(self.conf_file, self.res_dir)
rconfig.add_section('rbac')
rconfig.set('rbac', 'enable_rbac', True)
rconfig.set('rbac', 'rbac_test_role', kwargs.get('role', 'admin'))
- with open(self.conf_file, 'wb') as config_file:
+ with open(self.conf_file, 'w') as config_file:
rconfig.write(config_file)
self.backup_tempest_config(self.conf_file, self.res_dir)
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT)
deployment_uuid = proc.stdout.readline().rstrip()
- return deployment_uuid
+ return deployment_uuid.decode()
@staticmethod
def create_rally_deployment(environ=None):
rconfig.add_section('openstack')
rconfig.set(
'openstack', 'keystone_default_role', env.get("NEW_USER_ROLE"))
- with open(rally_conf, 'wb') as config_file:
+ with open(rally_conf, 'w') as config_file:
rconfig.write(config_file)
@staticmethod
rconfig.read(rally_conf)
if rconfig.has_option('openstack', 'keystone_default_role'):
rconfig.remove_option('openstack', 'keystone_default_role')
- with open(rally_conf, 'wb') as config_file:
+ with open(rally_conf, 'w') as config_file:
rconfig.write(config_file)
@staticmethod
taskid_re = re.compile('^Task +(.*): started$')
for line in cmd_raw.splitlines(True):
line = line.strip()
- match = taskid_re.match(line)
+ match = taskid_re.match(line.decode())
if match:
return match.group(1)
return None
"{}/{}.xml".format(self.results_dir, self.case_name),
export_type="junit-xml")
res = testcase.TestCase.EX_OK
- except Exception as exc: # pylint: disable=broad-except
- LOGGER.error('Error with run: %s', exc)
+ except Exception: # pylint: disable=broad-except
+ LOGGER.exception('Error with run:')
self.result = 0
res = testcase.TestCase.EX_RUN_ERROR
self.stop_time = time.time()
yaml_data2 = ""
for line in output.splitlines():
try:
- grp = re.search(r'^([^\[]*)(\[.*\])\n*$', line)
+ grp = re.search(r'^([^\[]*)(\[.*\])\n*$', line.decode())
yaml_data2 = "{}\n{}: {}".format(
yaml_data2, grp.group(1), grp.group(2))
except Exception: # pylint: disable=broad-except
stderr=subprocess.STDOUT)
for line in proc.stdout:
LOGGER.info(line.rstrip())
- new_line = line.replace(' ', '').split('|')
+ new_line = line.decode().replace(' ', '').split('|')
if 'Tests' in new_line:
break
if 'Testscount' in new_line:
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT)
verifier_uuid = proc.stdout.readline().rstrip()
- return verifier_uuid
+ return verifier_uuid.decode()
@staticmethod
def get_verifier_repo_dir(verifier_id):
for key, value in sub_conf.items():
rconfig.set(section, key, value)
- with open(conf_file, 'wb') as config_file:
+ with open(conf_file, 'w') as config_file:
rconfig.write(config_file)
@staticmethod
with proc.stdout:
for line in iter(proc.stdout.readline, b''):
- if re.search(r"\} tempest\.", line):
+ if re.search(r"\} tempest\.", line.decode()):
LOGGER.info(line.rstrip())
- elif re.search(r'(?=\(UUID=(.*)\))', line):
+ elif re.search(r'(?=\(UUID=(.*)\))', line.decode()):
self.verification_id = re.search(
- r'(?=\(UUID=(.*)\))', line).group(1)
- f_stdout.write(line)
+ r'(?=\(UUID=(.*)\))', line.decode()).group(1)
+ f_stdout.write(line.decode())
proc.wait()
f_stdout.close()
rconfig.add_section('openstack')
rconfig.set('openstack', 'img_name_regex', '^{}$'.format(
self.image.name))
- with open(rally_conf, 'wb') as config_file:
+ with open(rally_conf, 'w') as config_file:
rconfig.write(config_file)
def update_default_role(self, rally_conf='/etc/rally/rally.conf'):
if not rconfig.has_section('openstack'):
rconfig.add_section('openstack')
rconfig.set('openstack', 'swift_operator_role', role.name)
- with open(rally_conf, 'wb') as config_file:
+ with open(rally_conf, 'w') as config_file:
rconfig.write(config_file)
def update_rally_logs(self, rally_conf='/etc/rally/rally.conf'):
rconfig.set('DEFAULT', 'use_stderr', False)
rconfig.set('DEFAULT', 'log-file', 'rally.log')
rconfig.set('DEFAULT', 'log_dir', self.res_dir)
- with open(rally_conf, 'wb') as config_file:
+ with open(rally_conf, 'w') as config_file:
rconfig.write(config_file)
@staticmethod
rconfig.remove_option('DEFAULT', 'log-file')
if rconfig.has_option('DEFAULT', 'log_dir'):
rconfig.remove_option('DEFAULT', 'log_dir')
- with open(rally_conf, 'wb') as config_file:
+ with open(rally_conf, 'w') as config_file:
rconfig.write(config_file)
def update_network_section(self):
rconfig.add_section('network')
rconfig.set('network', 'public_network_id', self.ext_net.id)
rconfig.set('network', 'floating_network_name', self.ext_net.name)
- with open(self.conf_file, 'wb') as config_file:
+ with open(self.conf_file, 'w') as config_file:
rconfig.write(config_file)
def update_compute_section(self):
if not rconfig.has_section('compute'):
rconfig.add_section('compute')
rconfig.set('compute', 'fixed_network_name', self.network.name)
- with open(self.conf_file, 'wb') as config_file:
+ with open(self.conf_file, 'w') as config_file:
rconfig.write(config_file)
def update_scenario_section(self):
rconfig.set(
'scenario', 'img_properties',
functest_utils.convert_dict_to_ini(extra_properties))
- with open(self.conf_file, 'wb') as config_file:
+ with open(self.conf_file, 'w') as config_file:
rconfig.write(config_file)
def configure(self, **kwargs): # pylint: disable=unused-argument
for i in range(10):
output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
self.__logger.info("%s\n%s", " ".join(cmd), output)
- ret = re.search(r'(?=workload:({})\))'.format(status), output)
+ ret = re.search(
+ r'(?=workload:({})\))'.format(status), output.decode())
if ret:
self.__logger.info("%s workload is %s", name, status)
break
cmd)
break
- res_buff += res
+ res_buff += res.decode()
self.logger.debug("Response : '%s'", res_buff)
return res_buff
mock_os_makedirs.assert_called()
def test_get_task_id_default(self):
- cmd_raw = 'Task 1: started'
+ cmd_raw = b'Task 1: started'
self.assertEqual(self.rally_base.get_task_id(cmd_raw),
'1')
def test_get_task_id_missing_id(self):
- cmd_raw = ''
+ cmd_raw = b''
self.assertEqual(self.rally_base.get_task_id(cmd_raw),
None)
with mock.patch('functest.opnfv_tests.openstack.tempest.'
'tempest.subprocess.Popen') as mock_popen:
mock_stdout = mock.Mock()
- attrs = {'stdout.readline.return_value': 'test_deploy_id'}
+ attrs = {'stdout.readline.return_value': b'test_deploy_id'}
mock_stdout.configure_mock(**attrs)
mock_popen.return_value = mock_stdout
with mock.patch('functest.opnfv_tests.openstack.tempest.'
'tempest.subprocess.Popen') as mock_popen:
mock_stdout = mock.Mock()
- attrs = {'stdout.readline.return_value': 'test_deploy_id'}
+ attrs = {'stdout.readline.return_value': b'test_deploy_id'}
mock_stdout.configure_mock(**attrs)
mock_popen.return_value = mock_stdout
import mock
import pkg_resources
+import six
from functest.utils import functest_utils
as mock_subproc_open, \
mock.patch('six.moves.builtins.open',
mock.mock_open()) as mopen:
-
- FunctestUtilsTesting.readline = 0
-
- mock_obj = mock.Mock()
- attrs = {'readline.side_effect': self.cmd_readline()}
- mock_obj.configure_mock(**attrs)
-
+ stream = six.BytesIO()
+ stream.write(self.cmd_readline().encode())
mock_obj2 = mock.Mock()
- attrs = {'stdout': mock_obj, 'wait.return_value': 1}
+ attrs = {'stdout': stream, 'wait.return_value': 1}
mock_obj2.configure_mock(**attrs)
-
mock_subproc_open.return_value = mock_obj2
-
- resp = functest_utils.execute_command(self.cmd, info=True,
- error_msg=self.error_msg,
- verbose=True,
- output_file=self.output_file)
+ resp = functest_utils.execute_command(
+ self.cmd, info=True, error_msg=self.error_msg, verbose=True,
+ output_file=self.output_file)
self.assertEqual(resp, 1)
msg_exec = ("Executing command: '%s'" % self.cmd)
mock_logger_info.assert_called_once_with(msg_exec)
as mock_subproc_open, \
mock.patch('six.moves.builtins.open',
mock.mock_open()) as mopen:
-
- FunctestUtilsTesting.readline = 0
-
- mock_obj = mock.Mock()
- attrs = {'readline.side_effect': self.cmd_readline()}
- mock_obj.configure_mock(**attrs)
-
+ stream = six.BytesIO()
+ stream.write(self.cmd_readline().encode())
mock_obj2 = mock.Mock()
- attrs = {'stdout': mock_obj, 'wait.return_value': 0}
+ attrs = {'stdout': stream, 'wait.return_value': 0}
mock_obj2.configure_mock(**attrs)
-
mock_subproc_open.return_value = mock_obj2
-
- resp = functest_utils.execute_command(self.cmd, info=True,
- error_msg=self.error_msg,
- verbose=True,
- output_file=self.output_file)
+ resp = functest_utils.execute_command(
+ self.cmd, info=True, error_msg=self.error_msg, verbose=True,
+ output_file=self.output_file)
self.assertEqual(resp, 0)
msg_exec = ("Executing command: '%s'" % self.cmd)
mock_logger_info.assert_called_once_with(msg_exec)
# pylint: disable=unused-argument
with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
as mock_subproc_open:
-
- FunctestUtilsTesting.readline = 2
-
- mock_obj = mock.Mock()
- attrs = {'readline.side_effect': self.cmd_readline()}
- mock_obj.configure_mock(**attrs)
-
+ stream = six.BytesIO()
+ stream.write(self.cmd_readline().encode())
mock_obj2 = mock.Mock()
- attrs = {'stdout': mock_obj, 'wait.return_value': 0}
+ attrs = {'stdout': stream, 'wait.return_value': 0}
mock_obj2.configure_mock(**attrs)
-
mock_subproc_open.return_value = mock_obj2
-
- resp = functest_utils.execute_command(self.cmd, info=False,
- error_msg="",
- verbose=False,
- output_file=None)
+ resp = functest_utils.execute_command(
+ self.cmd, info=False, error_msg="", verbose=False,
+ output_file=None)
self.assertEqual(resp, 0)
@mock.patch('sys.stdout')
# pylint: disable=unused-argument
with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
as mock_subproc_open:
-
- FunctestUtilsTesting.readline = 2
- mock_obj = mock.Mock()
- attrs = {'readline.side_effect': self.cmd_readline()}
- mock_obj.configure_mock(**attrs)
-
+ stream = six.BytesIO()
+ stream.write(self.cmd_readline().encode())
mock_obj2 = mock.Mock()
- attrs = {'stdout': mock_obj, 'wait.return_value': 1}
+ attrs = {'stdout': stream, 'wait.return_value': 1}
mock_obj2.configure_mock(**attrs)
-
mock_subproc_open.return_value = mock_obj2
-
- resp = functest_utils.execute_command(self.cmd, info=False,
- error_msg="",
- verbose=False,
- output_file=None)
+ resp = functest_utils.execute_command(
+ self.cmd, info=False, error_msg="", verbose=False,
+ output_file=None)
self.assertEqual(resp, 1)
def test_get_param_from_yaml_failed(self):
ofd = open(output_file, "w")
for line in iter(popen.stdout.readline, b''):
if output_file:
- ofd.write(line)
+ ofd.write(line.decode())
else:
- line = line.replace('\n', '')
+ line = line.decode().replace('\n', '')
print(line)
sys.stdout.flush()
if output_file:
Code Review / functest.git / commitdiff