From 91701929e58a47920110d3589639975244cf57d1 Mon Sep 17 00:00:00 2001 From: Matthias Bisping Date: Tue, 7 Jun 2022 15:08:58 +0200 Subject: [PATCH] adjusted stream buffer test for core-operations taking tuples now --- pyinfra/server/buffering/stream.py | 2 +- test/conftest.py | 2 ++ test/integration_tests/serve_test.py | 8 ++++++-- test/unit_tests/server/stream_buffer_test.py | 8 ++++---- 4 files changed, 13 insertions(+), 7 deletions(-) diff --git a/pyinfra/server/buffering/stream.py b/pyinfra/server/buffering/stream.py index 6a53c61..e1cd669 100644 --- a/pyinfra/server/buffering/stream.py +++ b/pyinfra/server/buffering/stream.py @@ -38,7 +38,7 @@ class StreamBuffer: try: yield from self.fn(item) except TypeError as err: - raise TypeError("Function failed with type error. Is it mappable?") from err + raise TypeError("Function failed with type-error. Is it mappable?") from err def pop(self): return first(chain(self.result_stream, [Nothing])) diff --git a/test/conftest.py b/test/conftest.py index ad3db72..fb51e7d 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -26,11 +26,13 @@ logging.basicConfig() logger = logging.getLogger() +# TODO: refactor all fixtures into cleanly separated modules pytest_plugins = [ "test.fixtures.consumer", "test.fixtures.input", "test.fixtures.pdf", "test.fixtures.server", + "test.integration_tests.serve_test", ] diff --git a/test/integration_tests/serve_test.py b/test/integration_tests/serve_test.py index df6f103..458b23a 100644 --- a/test/integration_tests/serve_test.py +++ b/test/integration_tests/serve_test.py @@ -81,8 +81,12 @@ def test_serving(server_process, bucket_name, components, targets, data_message_ @pytest.fixture -def data_message_pairs(input_data_items, metadata): - data_metadata_packs = starmap(compose(lambda s: s.encode(), json.dumps, pack), zip(input_data_items, metadata)) +def data_metadata_packs(input_data_items, metadata): + return list(starmap(compose(lambda s: s.encode(), json.dumps, pack), zip(input_data_items, metadata))) + + +@pytest.fixture +def data_message_pairs(data_metadata_packs): data_message_pairs = pair_data_with_queue_message(data_metadata_packs) return data_message_pairs diff --git a/test/unit_tests/server/stream_buffer_test.py b/test/unit_tests/server/stream_buffer_test.py index 962257b..beac69b 100644 --- a/test/unit_tests/server/stream_buffer_test.py +++ b/test/unit_tests/server/stream_buffer_test.py @@ -4,7 +4,7 @@ from funcy import repeatedly, takewhile, notnone, lmap, lmapcat, lflatten from pyinfra.server.buffering.stream import FlatStreamBuffer, StreamBuffer from pyinfra.server.dispatcher.dispatcher import Nothing from pyinfra.server.stream.queued_stream_function import QueuedStreamFunction -from pyinfra.utils.func import lift, foreach +from pyinfra.utils.func import lift, foreach, starlift @pytest.fixture @@ -40,15 +40,15 @@ def test_flat_stream_buffer(func, inputs, outputs, buffer_size): def test_flat_stream_buffer_on_different_data( - core_operation, input_data_items, target_data_items, buffer_size, item_type, one_to_many + core_operation, input_data_items, metadata, target_data_items, buffer_size, item_type, one_to_many ): if core_operation is Nothing: pytest.skip(f"No operation defined for parameter combination: {item_type=}, {one_to_many=}") - flat_stream_buffer = FlatStreamBuffer(lift(core_operation), buffer_size=buffer_size) + flat_stream_buffer = FlatStreamBuffer(starlift(core_operation), buffer_size=buffer_size) - assert list(flat_stream_buffer(input_data_items)) == target_data_items + assert list(flat_stream_buffer(zip(input_data_items, metadata))) == target_data_items assert list(flat_stream_buffer([])) == []