test_utils.py 文件源码

def test_write_csv_to_s3_no_header(generate_data):
    bucket = 'test-bucket'
    key = 'test.csv'

    conn = boto3.resource('s3', region_name='us-west-2')
    conn.create_bucket(Bucket=bucket)

    utils.write_csv_to_s3(generate_data(), bucket, key, header=False)

    body = (
        conn
        .Object(bucket, key)
        .get()['Body']
        .read().decode('utf-8')
    )

    assert len(body.rstrip().split('\n')) == 1