pyinfra/test/unit_tests/server/buffer_test.py

import pytest
from funcy import compose, lmapcat, compact, flatten, identity

from pyinfra.exceptions import NoBufferCapacity
from pyinfra.server.buffering.bufferize import bufferize
from pyinfra.server.nothing import Nothing


def test_buffer(buffer_size):
    def buffer_mean(xs):
        return [sum(xs) / len(xs)] if xs else []

    buffer_mean = bufferize(compose(buffer_mean, list), buffer_size=3, null_value=[])
    ys = lmapcat(buffer_mean, (*range(20), Nothing))
    assert list(ys) == [1.0, 4.0, 7.0, 10.0, 13.0, 16.0, 18.5]

    def reverse_buffer(xs):
        return reversed(list(xs))

    reverse_buffer = bufferize(reverse_buffer, buffer_size=3)
    ys = flatten(compact(map(reverse_buffer, (*range(10), Nothing))))
    assert list(ys) == [2, 1, 0, 5, 4, 3, 8, 7, 6, 9]

    def buffer_sum(xs):
        return [sum(xs)]

    buffer_sum = bufferize(buffer_sum, buffer_size=2)
    ys = flatten(compact(map(buffer_sum, (*range(10), Nothing))))
    assert list(ys) == [0, 1, 0, 5, 0, 9, 0, 13, 0, 17, 0]


def test_buffer_overflow(buffer_size):
    with pytest.raises(NoBufferCapacity):
        bufferize(identity, buffer_size=0)