test_field_values.py 文件源码

def test_field_value_counter():
    counter = FieldValueCounter(quarter='2014Q1', field_values=['jobtitle', 'jobdesc'])
    counter.track(
        input_document={'jobtitle': 'test', 'jobdesc': 'test'},
    )
    counter.track(
        input_document={'jobtitle': 'test', 'jobdesc': '', 'extra': 'test'},
    )
    assert counter.accumulator['jobtitle']['test'] == 2
    assert counter.accumulator['jobdesc']['test'] == 1
    assert counter.accumulator['jobdesc'][''] == 1

    with moto.mock_s3():
        s3_conn = boto.connect_s3()
        s3_conn.create_bucket('test-bucket')
        counter.save(s3_conn, 'test-bucket/stats')

        key = s3_conn.get_bucket('test-bucket')\
            .get_key('stats/field_values/2014Q1/jobtitle.csv')
        expected_count = 'test,2'
        assert key.get_contents_as_string().decode('utf-8').rstrip() == expected_count