test_transform_dataset_stats.py 文件源码

def test_total_job_postings():
    with moto.mock_s3():
        s3_conn = boto.connect_s3()
        s3_conn.create_bucket('stats-bucket')
        bucket = s3_conn.get_bucket('stats-bucket')
        key = boto.s3.key.Key(
            bucket=bucket,
            name='partner-etl/summary.json'
        )
        key.set_contents_from_string(json.dumps({
            'total': 8,
            'output_counts': {
                'title': 8,
                'description': 4
            },
            'output_percentages': {
                'title': 1.0,
                'description': 0.5

            },
            'last_updated': '2017-01-10T00:00:00',
        }))

        assert GlobalStatsAggregator(s3_conn)\
            .saved_total(config['partner_stats']['s3_path']) == 8