From f21cbc141efc10d0e779f683dcc452e6c889dd6b Mon Sep 17 00:00:00 2001 From: Matthias Bisping Date: Mon, 14 Mar 2022 14:47:27 +0100 Subject: [PATCH] consumer with visitor test fixed (added connection.close() in case of pika queue manager) --- .../queue_manager/pika_queue_manager.py | 4 +- pyinfra/test/unit_tests/conftest.py | 9 +- pyinfra/test/unit_tests/consumer_test.py | 99 +++++++++---------- 3 files changed, 56 insertions(+), 56 deletions(-) diff --git a/pyinfra/pyampq/queue_manager/pika_queue_manager.py b/pyinfra/pyampq/queue_manager/pika_queue_manager.py index a9ee187..274caef 100644 --- a/pyinfra/pyampq/queue_manager/pika_queue_manager.py +++ b/pyinfra/pyampq/queue_manager/pika_queue_manager.py @@ -56,8 +56,8 @@ def get_connection(): class PikaQueueManager(QueueManager): def __init__(self, input_queue, output_queue): super().__init__(input_queue, output_queue) - connection = get_connection() - self.channel = connection.channel() + self.connection = get_connection() + self.channel = self.connection.channel() self.channel.queue_declare(input_queue) self.channel.queue_declare(output_queue) diff --git a/pyinfra/test/unit_tests/conftest.py b/pyinfra/test/unit_tests/conftest.py index 32c685a..90c3070 100644 --- a/pyinfra/test/unit_tests/conftest.py +++ b/pyinfra/test/unit_tests/conftest.py @@ -85,11 +85,14 @@ def get_queue_manager(queue_manager_name) -> QueueManager: @pytest.fixture(scope="session") -def queue_manager(queue_manager_name) -> QueueManager: - return get_queue_manager(queue_manager_name) +def queue_manager(queue_manager_name): + queue_manager = get_queue_manager(queue_manager_name) + yield queue_manager + if queue_manager_name == "pika": + queue_manager.connection.close() -@pytest.fixture +@pytest.fixture(scope="session") def callback(): return lambda x: x * 2 diff --git a/pyinfra/test/unit_tests/consumer_test.py b/pyinfra/test/unit_tests/consumer_test.py index b6b9ccd..1786d8b 100644 --- a/pyinfra/test/unit_tests/consumer_test.py +++ b/pyinfra/test/unit_tests/consumer_test.py @@ -9,7 +9,7 @@ from pyinfra.pyampq.consumer import Consumer from pyinfra.visitor import get_object_descriptor, ForwardingStrategy -@pytest.fixture +@pytest.fixture(scope="session") def consumer(queue_manager, callback): return Consumer(callback, queue_manager) @@ -52,67 +52,64 @@ class TestConsumer: # self, consumer, queue_manager, visitor, storage, bucket_name # ): # + # def produce_items(): + # for i in range(3): + # body = { + # "dossierId": "folder", + # "fileId": f"file{i}", + # "targetFileExtension": "in.gz", + # "responseFileExtension": "out.gz" + # } + # yield f"{i}".encode(), body + # # visitor.response_strategy = ForwardingStrategy() # + # queue_manager.clear() # storage.clear_bucket(bucket_name) - # queue_manager.clear() - # - # def produce_items(): - # for i in range(3): - # body = { - # "dossierId": "folder", - # "fileId": f"file{i}", - # "targetFileExtension": "in.gz", - # "responseFileExtension": "out.gz" - # } - # yield b"content", body - # - # items = list(produce_items()) - # - # for data, item in items: - # storage.put_object(**get_object_descriptor(item), data=gzip.compress(data)) - # queue_manager.publish_request(item) - # - # requests = list(consumer.consume()) - # print(1111111111111111111) - # - # print(requests) - # for _, r in zip(items, requests): - # print(r) - # queue_manager.publish_response(r, visitor) - # - # assert list(map(itemgetter("data"), queue_manager.output_queue.to_list())) == ["00", "11", "22"] - - # @pytest.mark.parametrize("client_name", ["mock", "s3", "azure"], scope="session") - # def test_visitor( - # self, consumer, queue_manager, access_callback, bucket_name, storage - # ): - # - # def produce_items(): - # for i in range(3): - # body = { - # "dossierId": "folder", - # "fileId": f"file{i}", - # "targetFileExtension": "in.gz", - # "responseFileExtension": "out.gz" - # } - # yield "content".encode(), body - # - # queue_manager.clear() - # # storage.clear_bucket(bucket_name) # # for data, message in produce_items(): - # # storage.put_object(**get_object_descriptor(message), data=gzip.compress(data)) + # storage.put_object(**get_object_descriptor(message), data=gzip.compress(data)) # queue_manager.publish_request(message) # - # print(11111111111111111111111111) - # # requests = consumer.consume() # # for _, r in zip(produce_items(), requests): - # queue_manager.publish_response(r, access_callback) + # queue_manager.publish_response(r, visitor) # - # assert queue_manager.output_queue.to_list() == ["file0", "file1", "file2"] + # assert list(map(itemgetter("data"), queue_manager.output_queue.to_list())) == ["00", "11", "22"] + + @pytest.mark.parametrize("client_name", ["mock", "s3", "azure"], scope="session") + @pytest.mark.parametrize("response_strategy_name", ["forwarding", "storage"], scope="session") + def test_consuming_nonempty_input_queue_with_visitor_puts_messages_on_output_queue_in_fifo_order( + self, consumer, queue_manager, visitor, + bucket_name, storage + ): + + def produce_items(): + for i in range(3): + body = { + "dossierId": "folder", + "fileId": f"file{i}", + "targetFileExtension": "in.gz", + "responseFileExtension": "out.gz" + } + yield f"{i}".encode(), body + + visitor.response_strategy = ForwardingStrategy() + + queue_manager.clear() + storage.clear_bucket(bucket_name) + + for data, message in produce_items(): + storage.put_object(**get_object_descriptor(message), data=gzip.compress(data)) + queue_manager.publish_request(message) + + requests = consumer.consume() + + for _, r in zip(produce_items(), requests): + queue_manager.publish_response(r, visitor) + + assert list(map(itemgetter("data"), queue_manager.output_queue.to_list())) == ["00", "11", "22"] logger = logging.getLogger("minio")