Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
)
mock_list_clusters.return_value = ["cluster1", "cluster2"]
mock_get_git_url.return_value = "fake_git_url"
mock_get_instance_config.return_value = mock_instance_config
mock_instance_config.get_deploy_group.return_value = "some_group"
mock_get_remote_refs.return_value = ["not_a_real_tag", "fake_tag"]
mock_get_latest_deployment_tag.return_value = ("not_a_real_tag", None)
mock_format_timestamp.return_value = "not_a_real_timestamp"
mock_apply_args_filters.return_value = {
"cluster1": {"fake_service": {"main1": None}},
"cluster2": {"fake_service": {"main1": None, "canary": None}},
}
mock_confirm_to_continue.return_value = True
ret = args.command(args)
c1_get_instance_config_call = mock.call(
service="fake_service",
cluster="cluster1",
instance="main1",
soa_dir="/soa/dir",
load_deployments=False,
)
c2_get_instance_config_call = mock.call(
service="fake_service",
cluster="cluster2",
instance="main1",
soa_dir="/soa/dir",
load_deployments=False,
)
c3_get_instance_config_call = mock.call(
service="fake_service",
cluster="cluster2",
def test_postgres_pkg(self, deb_packages):
postgres_pkg()
self.assertTrue(deb_packages.called)
self.assertEqual(deb_packages.call_args, call(['libpq-dev'], update=False))
mock.call('rel:1', {'foo': 'bar'}),
mock.call('rel:2', {'foo': 'bar'}),
])
relation_set.reset_mock()
conv.set_remote(data={'foo': 'bar'})
self.assertEqual(relation_set.call_args_list, [
mock.call('rel:1', {'foo': 'bar'}),
mock.call('rel:2', {'foo': 'bar'}),
])
relation_set.reset_mock()
conv.set_remote(foo='bar')
self.assertEqual(relation_set.call_args_list, [
mock.call('rel:1', {'foo': 'bar'}),
mock.call('rel:2', {'foo': 'bar'}),
])
relation_set.reset_mock()
conv.set_remote('foo', 'bof', {'foo': 'bar'}, foo='qux')
self.assertEqual(relation_set.call_args_list, [
mock.call('rel:1', {'foo': 'qux'}),
mock.call('rel:2', {'foo': 'qux'}),
])
relation_set.reset_mock()
conv.set_remote()
assert not relation_set.called
def test_stdout_logger_write(pipe_writer):
logger = StdoutLogger(0, 'utf-8')
logger.write('some-string\nother-string\nlast-string')
assert pipe_writer.write_log.called
assert pipe_writer.write_log.call_args_list[0] == mock.call(
level=20,
message='[stdout] some-string'
)
assert pipe_writer.write_log.call_args_list[1] == mock.call(
level=20,
message='[stdout] other-string'
)
assert logger.buf == 'last-string'
def test_call_tracking(self):
# only works in >= mock 0.8
manager = make_manager()
inst = ManagerMock(manager, 'foo')
inst.filter(foo='bar').select_related('baz')
calls = inst.mock_calls
self.assertGreater(len(calls), 1)
inst.assert_chain_calls(mock.call.filter(foo='bar'))
inst.assert_chain_calls(mock.call.select_related('baz'))
def test_uses_bandwidth_limiter(self):
bandwidth_limiter = mock.Mock(BandwidthLimiter)
self.stubber.add_response(
'get_object', service_response={'Body': self.stream},
expected_params={'Bucket': self.bucket, 'Key': self.key}
)
task = self.get_download_task(bandwidth_limiter=bandwidth_limiter)
task()
self.stubber.assert_no_pending_responses()
self.assertEqual(
bandwidth_limiter.get_bandwith_limited_stream.call_args_list,
[mock.call(mock.ANY, self.transfer_coordinator)]
)
backup_info_mock.WAITING_FOR_WALS = "WAITING_FOR_WALS"
backup_info_mock.return_value.status = BackupInfo.WAITING_FOR_WALS
backup_info = testing_helpers.build_test_backup_info()
backup_manager = testing_helpers.build_backup_manager()
executor = RecoveryExecutor(backup_manager)
backup_info.status = BackupInfo.WAITING_FOR_WALS
destination = tmpdir.mkdir('destination').strpath
with closing(executor):
executor.recover(backup_info, destination, standby_mode=None)
# The backup info has been read again
backup_info_mock.assert_called()
# The following two warning messages have been emitted
output_mock.warning.assert_has_calls([
mock.call(
"IMPORTANT: You have requested a recovery operation for "
"a backup that does not have yet all the WAL files that "
"are required for consistency."),
mock.call(
"IMPORTANT: The backup we have recovered IS NOT "
"VALID. Required WAL files for consistency are "
"missing. Please verify that WAL archiving is "
"working correctly or evaluate using the 'get-wal' "
"option for recovery")])
# In the following test case, the backup will be validated during
# the copy of the data files, so there is no need for the warning
# message at the end of the recovery process to be emitted again
output_mock.warning.reset_mock()
backup_info_mock.return_value.status = BackupInfo.DONE
with closing(executor):
resource_classes=resource_classes,
task_id=task_id)._consumer
admin = mock.MagicMock()
user1 = {"id": "a", "tenant_id": "uuid1"}
cache = {}
consumer(cache, (admin, user1, "res"))
mock_mgr.assert_called_once_with(
resource="res",
admin=mock__get_cached_client.return_value,
user=mock__get_cached_client.return_value,
tenant_uuid=user1["tenant_id"])
mock__get_cached_client.assert_has_calls([
mock.call(admin),
mock.call(user1)
])
mock__delete_single_resource.assert_called_once_with(
mock_mgr.return_value)
mock_mgr.reset_mock()
mock__get_cached_client.reset_mock()
mock__delete_single_resource.reset_mock()
mock_name_matches_object.reset_mock()
consumer(cache, (admin, None, "res2"))
mock_mgr.assert_called_once_with(
resource="res2",
admin=mock__get_cached_client.return_value,
user=mock__get_cached_client.return_value,
tenant_uuid=None)
aux = Mock()
mx_load_checkpoint.return_value = [sym, args, aux]
eia_context = Mock()
mx_eia.return_value = eia_context
data_name = 'foo'
data_shape = [1]
signature = json.dumps([{'name': data_name, 'shape': data_shape}])
with patch('six.moves.builtins.open', mock_open(read_data=signature)):
DefaultMXNetInferenceHandler().default_model_fn(MODEL_DIR)
mx_load_checkpoint.assert_called_with(os.path.join(MODEL_DIR, 'model'), 0)
init_call = call(symbol=sym, context=eia_context, data_names=[data_name], label_names=None)
assert init_call in mx_module.mock_calls
model = mx_module.return_value
model.bind.assert_called_with(for_training=False, data_shapes=[(data_name, data_shape)])
model.set_params.assert_called_with(args, aux, allow_missing=True)
# Write more than just a few bytes in each iteration to keep the
# test reasonably fast
content += '*' * int(file.buffer_size / 10)
file.write(content)
written_content += content
counter += 1
# Save the internal file before closing
multipart.parts.all.return_value = [
mock.MagicMock(e_tag='123', part_number=1),
mock.MagicMock(e_tag='456', part_number=2)
]
file.close()
self.assertListEqual(
multipart.Part.call_args_list,
[mock.call(1), mock.call(2)]
)
part = multipart.Part.return_value
uploaded_content = ''.join(
(args_list[1]['Body'].decode('utf-8')
for args_list in part.upload.call_args_list)
)
self.assertEqual(uploaded_content, written_content)
multipart.complete.assert_called_once_with(
MultipartUpload={'Parts': [
{'ETag': '123', 'PartNumber': 1},
{'ETag': '456', 'PartNumber': 2},
]}