def test_local_path(self):
with mock.patch('builtins.open', mock.mock_open()):
cli.client.create_upload = mock.MagicMock(return_value={'upload': {'url': 'https://test', 'arn': 'arn'}})
requests.put = mock.MagicMock()
result = runner.invoke(cli.upload, ['-n', 'test_upload', '-p', 'test_project', '-t', 'ANDROID_APP',
'-f', 'app.apk'])
self.assertEqual(result.exit_code, 0)
python类mock_open()的实例源码
def test_download_url_default(self, mock_url):
with mock.patch("six.moves.builtins.open", mock.mock_open()) as m, \
mock.patch('functest.utils.functest_utils.shutil.copyfileobj')\
as mock_sh:
name = self.url.rsplit('/')[-1]
dest = self.dest_path + "/" + name
self.assertTrue(functest_utils.download_url(self.url,
self.dest_path))
m.assert_called_once_with(dest, 'wb')
self.assertTrue(mock_sh.called)
def test_execute_command_args_present_with_error(self, mock_logger_info,
mock_logger_error):
with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
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)
mock_obj2 = mock.Mock()
attrs = {'stdout': mock_obj, '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)
self.assertEqual(resp, 1)
msg_exec = ("Executing command: '%s'" % self.cmd)
mock_logger_info.assert_called_once_with(msg_exec)
mopen.assert_called_once_with(self.output_file, "w")
mock_logger_error.assert_called_once_with(self.error_msg)
def test_execute_command_args_present_with_success(self, mock_logger_info,
):
with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
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)
mock_obj2 = mock.Mock()
attrs = {'stdout': mock_obj, '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)
self.assertEqual(resp, 0)
msg_exec = ("Executing command: '%s'" % self.cmd)
mock_logger_info.assert_called_once_with(msg_exec)
mopen.assert_called_once_with(self.output_file, "w")
def test_get_dict_by_test(self, mock_logger_error):
with mock.patch('six.moves.builtins.open', mock.mock_open()), \
mock.patch('functest.utils.functest_utils.yaml.safe_load') \
as mock_yaml:
mock_obj = mock.Mock()
attrs = {'get.return_value': [{'testcases': [self.testcase_dict]}]}
mock_obj.configure_mock(**attrs)
mock_yaml.return_value = mock_obj
self.assertDictEqual(functest_utils.
get_dict_by_test(self.testname),
self.testcase_dict)
def test_get_parameter_from_yaml_failed(self):
self.file_yaml['general'] = None
with mock.patch('six.moves.builtins.open', mock.mock_open()), \
mock.patch('functest.utils.functest_utils.yaml.safe_load') \
as mock_yaml, \
self.assertRaises(ValueError) as excep:
mock_yaml.return_value = self.file_yaml
functest_utils.get_parameter_from_yaml(self.parameter,
self.test_file)
self.assertTrue(("The parameter %s is not"
" defined in config_functest.yaml" %
self.parameter) in excep.exception)
def test_get_functest_yaml(self):
with mock.patch('six.moves.builtins.open', mock.mock_open()), \
mock.patch('functest.utils.functest_utils.yaml.safe_load') \
as mock_yaml:
mock_yaml.return_value = self.file_yaml
resp = functest_utils.get_functest_yaml()
self.assertEqual(resp, self.file_yaml)
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_create_glance_image_default(self, mock_logger_info):
with mock.patch('functest.utils.openstack_utils.'
'os.path.isfile',
return_value=True), \
mock.patch('functest.utils.openstack_utils.get_image_id',
return_value=''), \
mock.patch('six.moves.builtins.open',
mock.mock_open(read_data='1')) as m:
self.assertEqual(openstack_utils.
create_glance_image(self.glance_client,
'test_image',
'file_path'),
'image_id')
m.assert_called_once_with('file_path')
self.assertTrue(mock_logger_info.called)
def _test_dump(self, *args):
CONST.__setattr__('results_test_db_url', URL)
with mock.patch.object(decorators, 'open', mock.mock_open(),
create=True) as mock_open:
self.assertTrue(functest_utils.push_results_to_db(
self._project_name, self._case_name, self._start_time,
self._stop_time, self._result, {}))
mock_open.assert_called_once_with(FILE, 'a')
handle = mock_open()
call_args, _ = handle.write.call_args
self.assertIn('POST', call_args[0])
self.assertIn(self._get_json(), call_args[0])
args[0].assert_called_once_with()
def setUp(self):
self.dependency = {'installer': 'test_installer',
'scenario': 'test_scenario'}
self.testcase = {'dependencies': self.dependency,
'enabled': 'true',
'case_name': 'test_name',
'criteria': 'test_criteria',
'blocking': 'test_blocking',
'description': 'test_desc',
'project_name': 'project_name'}
self.dic_tier = {'name': 'test_tier',
'order': 'test_order',
'ci_loop': 'test_ci_loop',
'description': 'test_desc',
'testcases': [self.testcase]}
self.mock_yaml = mock.Mock()
attrs = {'get.return_value': [self.dic_tier]}
self.mock_yaml.configure_mock(**attrs)
with mock.patch('functest.ci.tier_builder.yaml.safe_load',
return_value=self.mock_yaml), \
mock.patch('six.moves.builtins.open', mock.mock_open()):
self.tierbuilder = tier_builder.TierBuilder('test_installer',
'test_scenario',
'testcases_file')
self.tier_obj = self.tierbuilder.tier_objects[0]
def test_check_rc_missing_os_auth(self):
with mock.patch('six.moves.builtins.open',
mock.mock_open(read_data='test')), \
self.assertRaises(Exception) as context:
msg = 'OS_AUTH_URL not defined in {}.'.format(self.rc_file)
self.assertTrue(msg in context)
def test_configure_tempest_defcore_default(self):
with mock.patch('functest.opnfv_tests.openstack.tempest.'
'conf_utils.configure_verifier',
return_value='test_conf_file'), \
mock.patch('functest.opnfv_tests.openstack.tempest.'
'conf_utils.configure_tempest_update_params'), \
mock.patch('functest.opnfv_tests.openstack.tempest.'
'conf_utils.ConfigParser.RawConfigParser.'
'set') as mset, \
mock.patch('functest.opnfv_tests.openstack.tempest.'
'conf_utils.ConfigParser.RawConfigParser.'
'read') as mread, \
mock.patch('functest.opnfv_tests.openstack.tempest.'
'conf_utils.ConfigParser.RawConfigParser.'
'write') as mwrite, \
mock.patch('__builtin__.open', mock.mock_open()), \
mock.patch('functest.opnfv_tests.openstack.tempest.'
'conf_utils.generate_test_accounts_file'), \
mock.patch('functest.opnfv_tests.openstack.tempest.'
'conf_utils.shutil.copyfile'):
conf_utils.configure_tempest_defcore(
'test_dep_dir', 'test_image_id', 'test_flavor_id',
'test_image_alt_id', 'test_flavor_alt_id', 'test_tenant_id')
mset.assert_any_call('compute', 'image_ref', 'test_image_id')
mset.assert_any_call('compute', 'image_ref_alt',
'test_image_alt_id')
mset.assert_any_call('compute', 'flavor_ref', 'test_flavor_id')
mset.assert_any_call('compute', 'flavor_ref_alt',
'test_flavor_alt_id')
self.assertTrue(mread.called)
self.assertTrue(mwrite.called)
def test_generate_test_accounts_file_default(self):
with mock.patch("__builtin__.open", mock.mock_open()), \
mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
'yaml.dump') as mock_dump:
conf_utils.generate_test_accounts_file('test_tenant_id')
self.assertTrue(mock_dump.called)
def test_apply_tempest_blacklist_no_blacklist(self):
with mock.patch('__builtin__.open', mock.mock_open()) as m, \
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')
self.tempestcommon.apply_tempest_blacklist()
obj = m()
obj.write.assert_any_call('test1\n')
obj.write.assert_any_call('test2\n')
def test_run_verifier_tests_default(self, mock_logger_info):
with mock.patch('__builtin__.open', mock.mock_open()), \
mock.patch('__builtin__.iter', return_value=['\} tempest\.']), \
mock.patch('functest.opnfv_tests.openstack.tempest.tempest.'
'subprocess.Popen'):
conf_utils.TEMPEST_LIST = 'test_tempest_list'
cmd_line = ("rally verify start --load-list "
"test_tempest_list --detailed")
self.tempestcommon.run_verifier_tests()
mock_logger_info. \
assert_any_call("Starting Tempest test suite: '%s'."
% cmd_line)
def test_excl_scenario_exception(self, mock_open):
self.assertEqual(self.rally_base.excl_scenario(), [])
mock_open.assert_called()
def test_excl_func_exception(self, mock_open):
self.assertEqual(self.rally_base.excl_func(), [])
mock_open.assert_called()
