Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_run_test_matrix(do_tests, runner, continue_on_fail, result, expected):
dt = do_tests.DoTests({}, {}, continue_on_fail=continue_on_fail)
with mock.patch.multiple(do_tests, pytest_runner=mock.DEFAULT, shell_runner=mock.DEFAULT) as mock_rs:
mock_r = mock_rs[runner + '_runner']
mock_r.runner.return_value = result
assert dt.run_test(sentinel.ssh, {runner: sentinel.path}, sentinel.config, sentinel.var) is expected
assert mock_r.runner.called
def test_retry(self, mock_exc_to_code, mock_time):
mock_exc_to_code.side_effect = lambda e: e.code
to_attempt = 3
retry = RetryOptions(
[_FAKE_STATUS_CODE_1],
BackoffSettings(0, 0, 0, 0, 0, 0, 1))
# Succeeds on the to_attempt'th call, and never again afterward
mock_call = mock.Mock()
mock_call.side_effect = ([CustomException('', _FAKE_STATUS_CODE_1)] *
(to_attempt - 1) + [mock.DEFAULT])
mock_call.return_value = 1729
mock_time.return_value = 0
settings = _CallSettings(timeout=0, retry=retry)
my_callable = api_callable.create_api_call(mock_call, settings)
self.assertEqual(my_callable(None), 1729)
self.assertEqual(mock_call.call_count, to_attempt)
os_environ = {
'OS_AUTH_URL': 'https://keystone.example.com/v2.0/',
'OS_TENANT_NAME': 'demo',
'OS_USERNAME': 'demo',
'OS_PASSWORD': 'admin',
}
os_environ.update(pre_auth_env)
os_environ['OS_AUTH_TOKEN'] = 'expired'
fake_conn = self.fake_http_connection(401, 200)
fake_keystone = fake_get_auth_keystone(storage_url=url + '_not_used',
token=token + '_new')
with mock.patch.multiple('swiftclient.client',
http_connection=fake_conn,
get_auth_keystone=fake_keystone,
sleep=mock.DEFAULT):
with mock.patch.dict(os.environ, os_environ):
argv = ['', 'stat']
swiftclient.shell.main(argv)
self.assertRequests([
('HEAD', url, '', {
'x-auth-token': 'expired',
}),
('HEAD', url, '', {
'x-auth-token': token + '_new',
}),
def test_exceptions(self):
"""Test there are no default implementations in ManagedInstance
Doesn't do much but get us test coverage.
"""
self.assertRaises(NotImplementedError,
lambda: ManagedInstance('foo1'))
with mock.patch.multiple(ManagedInstance,
_start_process=mock.DEFAULT,
_start_log_watcher=mock.DEFAULT) as mocks:
instance = ManagedInstance('foo2')
instance._process = mock.MagicMock()
self.assertRaises(NotImplementedError, lambda: instance.flush())
open(instance.logfile, 'w').write('ohai') # so can delete
instance.terminate()
def validateBody(blogId=None, body=None, isDraft=True):
assert body["title"] == "t"
assert body["content"] == "c"
assert body["labels"] == ["l"]
assert isDraft
return DEFAULT
def validateBody(blogId, postId, body, revert, publish):
assert blogId == "1234"
assert postId == "4321"
assert body["title"] == "t"
assert body["content"] == "c"
assert body["labels"] == ["l"]
assert revert
assert not publish
return DEFAULT
@patch.multiple(Dataset, register_file=mock.DEFAULT, compute_md5=mock.DEFAULT)
def test_datasets_add_bulk(self, register_file, compute_md5):
filename1 = "added1.txt"
upload_file1 = SimpleUploadedFile(filename1, b"Hello, World!")
filename2 = "added2.txt"
upload_file2 = SimpleUploadedFile(filename2, b"Goodbye, Town!")
response = self.client.post(
reverse('datasets_add_bulk'),
data=dict(dataset_files=[upload_file1, upload_file2]))
self.assertEqual(200, response.status_code)
self.assertEqual(2, response.context['num_files_added'])
self.assertEqual(2, register_file.call_count)
self.assertEqual(filename1,
register_file.call_args_list[0][1]['file_handle'].name)
self.assertEqual(filename2,
register_file.call_args_list[1][1]['file_handle'].name)
cli_args = []
# Make sure last command is 'exit'
if prompt_commands is None:
prompt_commands = ['exit']
else:
prompt_commands += ['exit']
# Fool cli() so that it believes we're running from CLI instead of pytest.
# We will restore it at the end of the function.
orig_argv = sys.argv
sys.argv = ['http-prompt'] + cli_args
try:
with patch.multiple('http_prompt.cli',
prompt=DEFAULT, execute=DEFAULT) as mocks:
mocks['execute'].side_effect = execute
# prompt() is mocked to return the command in 'prompt_commands' in
# sequence, i.e., prompt() returns prompt_commands[i-1] when it is
# called for the ith time
mocks['prompt'].side_effect = prompt_commands
result = CliRunner().invoke(cli, cli_args)
context = mocks['execute'].call_args[0][1]
return result, context
finally:
sys.argv = orig_argv
command = get_command('setaddr')
expect(command).to.be.a('guerrillamail.SetAddressCommand')
def test_get_list_command_should_return_get_list_command_instance(self):
command = get_command('list')
expect(command).to.be.a('guerrillamail.ListEmailCommand')
def test_get_get_command_should_return_get_email_command_instance(self):
command = get_command('get')
expect(command).to.be.a('guerrillamail.GetEmailCommand')
def test_get_unknown_command_should_raise_exception(self):
expect(get_command).when.called_with('cheese').to.throw(ValueError)
@patch.multiple('guerrillamail', load_settings=DEFAULT, save_settings=DEFAULT, GuerrillaMailSession=DEFAULT,
parse_args=DEFAULT, get_command=DEFAULT)
class GuerrillaMailCliTest(TestCase):
def setup_mocks(self, GuerrillaMailSession, load_settings, parse_args, get_command, **kwargs):
load_settings.return_value = {}
self.mock_session = Mock()
self.mock_args = Mock()
self.mock_command = Mock()
GuerrillaMailSession.return_value = self.mock_session
parse_args.return_value = self.mock_args
get_command.return_value = self.mock_command
def test_cli_should_create_session_using_settings(self, GuerrillaMailSession, load_settings, **kwargs):
self.setup_mocks(GuerrillaMailSession=GuerrillaMailSession, load_settings=load_settings, **kwargs)
load_settings.return_value = {'arg1': 1, 'arg2': 'cheese'}
cli()
GuerrillaMailSession.assert_called_with(arg1=1, arg2='cheese')
def validateBody(blogId=None, body=None, isDraft=True):
assert body["title"] == "t"
assert body["content"] == ""
assert len(body["labels"]) == 3
return DEFAULT
self.posts.insert.side_effect = validateBody