test_datasets_job_postings.py 文件源码

def test_job_postings():
    s3_conn = boto.connect_s3()
    bucket_name = 'test-bucket'
    path = 'postings'
    quarter = '2014Q1'
    bucket = s3_conn.create_bucket(bucket_name)
    for i in range(0, 2):
        key = boto.s3.key.Key(
            bucket=bucket,
            name='{}/{}/{}'.format(path, quarter, i)
        )
        key.set_contents_from_string('test')

    # both variants of job postings getter should have identical results
    for func in [job_postings, job_postings_highmem]:
        postings = [posting for posting in func(
            s3_conn,
            quarter,
            '{}/{}'.format(bucket_name, path)
        )]
        assert postings == ['test'] * 2