From ccad97992d9f6c50318677101280a5a4f0db6137 Mon Sep 17 00:00:00 2001 From: Matthias Bisping Date: Mon, 14 Mar 2022 13:53:37 +0100 Subject: [PATCH] consumer with visitor test WIP --- config.yaml | 2 +- .../queue_manager/pika_queue_manager.py | 8 +- pyinfra/test/unit_tests/conftest.py | 7 +- pyinfra/test/unit_tests/consumer_test.py | 83 +++++++++++++++++++ pyinfra/test/unit_tests/queue_visitor_test.py | 7 +- 5 files changed, 97 insertions(+), 10 deletions(-) diff --git a/config.yaml b/config.yaml index ebbd7a5..cfc6aed 100755 --- a/config.yaml +++ b/config.yaml @@ -1,5 +1,5 @@ service: - logging_level: $LOGGING_LEVEL_ROOT|INFO # Logging level for service logger + logging_level: $LOGGING_LEVEL_ROOT|DEBUG # Logging level for service logger response: type: $RESPONSE_TYPE|"file" # Whether the analysis response is stored as file on storage or sent as stream # extension: $RESPONSE_FILE_EXTENSION|"IMAGE_INFO.json.gz" # {.IMAGE_INFO.json.gz | .NER_ENTITIES.json.gz} diff --git a/pyinfra/pyampq/queue_manager/pika_queue_manager.py b/pyinfra/pyampq/queue_manager/pika_queue_manager.py index 43fa077..a9ee187 100644 --- a/pyinfra/pyampq/queue_manager/pika_queue_manager.py +++ b/pyinfra/pyampq/queue_manager/pika_queue_manager.py @@ -62,12 +62,15 @@ class PikaQueueManager(QueueManager): self.channel.queue_declare(output_queue) def publish_request(self, request): - self.channel.basic_publish("", self._input_queue, json.dumps(request).encode()) + logger.debug(f"Publishing {request}") + self.channel.basic_publish("", self._input_queue, json.dumps(request)) def publish_response(self, message, callback): + logger.debug(f"Publishing response for {message}.") + frame, properties, body = message - response = json.dumps(callback(json.loads(body))).encode() + response = json.dumps(callback(json.loads(body))) self.channel.basic_publish("", self._output_queue, response) self.channel.basic_ack(frame.delivery_tag) @@ -76,6 +79,7 @@ class PikaQueueManager(QueueManager): return self.channel.basic_get(self._input_queue) def consume(self): + logger.debug("Consuming") return self.channel.consume(self._input_queue) def consume_and_publish(self, callback): diff --git a/pyinfra/test/unit_tests/conftest.py b/pyinfra/test/unit_tests/conftest.py index f39785a..32c685a 100644 --- a/pyinfra/test/unit_tests/conftest.py +++ b/pyinfra/test/unit_tests/conftest.py @@ -17,7 +17,7 @@ from pyinfra.test.config import CONFIG from pyinfra.test.queue_manager_mock import QueueManagerMock from pyinfra.test.storage.adapter_mock import StorageAdapterMock from pyinfra.test.storage.client_mock import StorageClientMock -from pyinfra.visitor import StorageStrategy, ForwardingStrategy +from pyinfra.visitor import StorageStrategy, ForwardingStrategy, QueueVisitor logger = logging.getLogger(__name__) logger.setLevel(logging.DEBUG) @@ -107,3 +107,8 @@ def response_strategy(response_strategy_name): return StorageStrategy() if response_strategy_name == "forwarding": return ForwardingStrategy() + + +@pytest.fixture() +def visitor(storage, analysis_callback, response_strategy): + return QueueVisitor(storage, analysis_callback, response_strategy) diff --git a/pyinfra/test/unit_tests/consumer_test.py b/pyinfra/test/unit_tests/consumer_test.py index 66e2bb3..f8793ae 100644 --- a/pyinfra/test/unit_tests/consumer_test.py +++ b/pyinfra/test/unit_tests/consumer_test.py @@ -1,6 +1,12 @@ +import gzip +import json +import logging +from operator import itemgetter + import pytest from pyinfra.pyampq.consumer import Consumer +from pyinfra.visitor import get_object_descriptor, ForwardingStrategy @pytest.fixture @@ -8,6 +14,11 @@ def consumer(queue_manager, callback): return Consumer(callback, queue_manager) +@pytest.fixture(scope="session") +def access_callback(): + return itemgetter("fileId") + + @pytest.mark.parametrize("queue_manager_name", ["mock", "pika"], scope="session") class TestConsumer: @@ -34,3 +45,75 @@ class TestConsumer: queue_manager.publish_response(r, callback) assert 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, storage, bucket_name + # ): + # + # visitor.response_strategy = ForwardingStrategy() + # + # 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"]) + # 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)) + # queue_manager.publish_request(message) + # + # print(11111111111111111111111111) + # + # requests = consumer.consume() + # + # for _, r in zip(produce_items(), requests): + # queue_manager.publish_response(r, access_callback) + # + # assert queue_manager.output_queue.to_list() == ["file0", "file1", "file2"] + + +logger = logging.getLogger("minio") +logger.setLevel(logging.DEBUG) diff --git a/pyinfra/test/unit_tests/queue_visitor_test.py b/pyinfra/test/unit_tests/queue_visitor_test.py index 8a7e107..d610bdd 100644 --- a/pyinfra/test/unit_tests/queue_visitor_test.py +++ b/pyinfra/test/unit_tests/queue_visitor_test.py @@ -3,12 +3,7 @@ import json import pytest -from pyinfra.visitor import QueueVisitor, get_object_descriptor, StorageStrategy, get_response_object_descriptor - - -@pytest.fixture() -def visitor(storage, analysis_callback, response_strategy): - return QueueVisitor(storage, analysis_callback, response_strategy) +from pyinfra.visitor import get_object_descriptor, StorageStrategy, get_response_object_descriptor @pytest.fixture()