test_io_handling.py 文件源码

def test_single_io_write_stream_encode_output(
        tmpdir,
        patch_aws_encryption_sdk_stream,
        patch_json_ready_header,
        patch_json_ready_header_auth
):
    patch_aws_encryption_sdk_stream.return_value = io.BytesIO(DATA)
    patch_aws_encryption_sdk_stream.return_value.header = MagicMock(encryption_context=sentinel.encryption_context)
    target_file = tmpdir.join('target')
    mock_source = MagicMock()
    kwargs = GOOD_IOHANDLER_KWARGS.copy()
    kwargs['encode_output'] = True
    handler = io_handling.IOHandler(**kwargs)
    with open(str(target_file), 'wb') as destination_writer:
        handler._single_io_write(
            stream_args={
                'mode': 'encrypt',
                'a': sentinel.a,
                'b': sentinel.b
            },
            source=mock_source,
            destination_writer=destination_writer
        )

    assert target_file.read('rb') == base64.b64encode(DATA)