28 lines
945 B
Python
28 lines
945 B
Python
from funcy import compose, lmapcat, compact, flatten
|
|
|
|
from pyinfra.server.dispatcher.dispatcher import Nothing
|
|
from pyinfra.server.bufferizer.buffer import bufferize
|
|
|
|
|
|
def test_buffer():
|
|
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]
|