consumer with visitor test WIP

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}

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):

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)

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)

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()