test_tailchaser.py 文件源码

def test_backfill(log_handler=RotatingWithDelayFileHandler, consumer=None, tail_to_dir=None, vargs=None):
    tail_from_dir = tempfile.mkdtemp(prefix='tail-test_backfill-tail_from_dir')
    six.print_('generating log files', tail_from_dir)
    log_handler.generate(os.path.join(tail_from_dir, 'test.log'), BACKFILL_EMITS)
    if not tail_to_dir:
        tail_to_dir = tempfile.mkdtemp(prefix='tail-test_backfill-tail_to_dir')

    if not consumer:
        def consumer_gen():
            while True:
                record = yield ()
                open(os.path.join(tail_to_dir, record[1][0]), 'ab').write(record[2])

        consumer = consumer_gen()
        consumer.send(None)
    six.print_('start tailer', tail_to_dir)
    source_pattern = os.path.join(tail_from_dir, '*')
    if not vargs:
        vargs = [__name__, '--only-backfill', '--clear-checkpoint']
    vargs.append(source_pattern)
    main(vargs, consumer)
    cmp_files(source_pattern, tail_to_dir, lambda x: Tailer.make_sig(x))
    six.print_('all done', tail_to_dir)
    # for src_file_path in glob.glob(source_pattern):
    #     dst_file_path = os.path.join(tail_to_dir, Tailer.make_sig(src_file_path))
    #     six.print_("testing:", src_file_path, dst_file_path)
    #     assert (Tailer.file_opener(src_file_path).read() == Tailer.file_opener(dst_file_path).read())