RED-4653: Removed code that is (probably) not needed for pyinfra as a module

2022-07-15 11:52:57 +02:00 · 2022-07-15 11:52:57 +02:00 · 5f97b3dedb
commit 5f97b3dedb
parent 3f645484d9
35 changed files with 0 additions and 1296 deletions
--- a/19
+++ b/19
@ -1,19 +0,0 @@
 FROM python:3.8
 # Use a virtual environment.
 RUN python -m venv /app/venv
 ENV PATH="/app/venv/bin:$PATH"
 # Upgrade pip.
 RUN python -m pip install --upgrade pip
 # Make a directory for the service files and copy the service repo into the container.
 WORKDIR /app/service
 COPY . .
 # Install module & dependencies
 RUN python3 -m pip install -e .
 RUN python3 -m pip install -r requirements.txt
 # Run the service loop.
 CMD ["python", "src/serve.py"]
--- a/19
+++ b/19
@ -1,19 +0,0 @@
 ARG BASE_ROOT="nexus.iqser.com:5001/red/"
 ARG VERSION_TAG="dev"
 FROM ${BASE_ROOT}pyinfra:${VERSION_TAG}
 EXPOSE 5000
 EXPOSE 8080
 RUN python3 -m pip install coverage
 # Make a directory for the service files and copy the service repo into the container.
 WORKDIR /app/service
 COPY . .
 # Install module & dependencies
 RUN python3 -m pip install -e .
 RUN python3 -m pip install -r requirements.txt
 CMD coverage run -m pytest test/ -x && coverage report -m && coverage xml
--- a/banner.txt
+++ b/banner.txt
@ -1,6 +0,0 @@
             ___    _  _    ___               __
    o O O   | _ \  | || |  |_ _|   _ _       / _|    _ _   __ _
   o        |  _/   \_, |   | |   | ' \     |  _|   | '_| / _` |
  TS__[O]  _|_|_   _|__/   |___|  |_||_|   _|_|_   _|_|_  \__,_|
 {======|_| ``` |_| ````|_|`````|_|`````|_|`````|_|`````|_|`````|
 ./o--000' `-0-0-' `-0-0-' `-0-0-' `-0-0-' `-0-0-' `-0-0-' `-0-0-'
--- a/config.yaml
+++ b/config.yaml
@ -1,35 +0,0 @@
 service:
  logging_level: $LOGGING_LEVEL_ROOT|DEBUG  # Logging level for service logger
 probing_webserver:
  host: $PROBING_WEBSERVER_HOST|"0.0.0.0"  # Probe webserver address
  port: $PROBING_WEBSERVER_PORT|8080  # Probe webserver port
  mode: $PROBING_WEBSERVER_MODE|production  # webserver mode: {development, production}
 rabbitmq:
  host: $RABBITMQ_HOST|localhost  # RabbitMQ host address
  port: $RABBITMQ_PORT|5672  # RabbitMQ host port
  user: $RABBITMQ_USERNAME|user  # RabbitMQ username
  password: $RABBITMQ_PASSWORD|bitnami  # RabbitMQ password
  heartbeat: $RABBITMQ_HEARTBEAT|7200  # Controls AMQP heartbeat timeout in seconds
  queues:
    input: $REQUEST_QUEUE|request_queue  # Requests to service
    output: $RESPONSE_QUEUE|response_queue  # Responses by service
    dead_letter: $DEAD_LETTER_QUEUE|dead_letter_queue  # Messages that failed to process
  callback:
    analysis_endpoint: $ANALYSIS_ENDPOINT|"http://127.0.0.1:5000"
 storage:
  backend: $STORAGE_BACKEND|s3  # The type of storage to use {s3, azure}
  bucket: "STORAGE_BUCKET_NAME|STORAGE_AZURECONTAINERNAME|pyinfra-test-bucket"  # The bucket / container to pull files specified in queue requests from
  s3:
    endpoint: $STORAGE_ENDPOINT|"http://127.0.0.1:9000"
    access_key: $STORAGE_KEY|root
    secret_key: $STORAGE_SECRET|password
    region: $STORAGE_REGION|"eu-west-1"
  azure:
    connection_string: $STORAGE_AZURECONNECTIONSTRING|"DefaultEndpointsProtocol=https;AccountName=iqserdevelopment;AccountKey=4imAbV9PYXaztSOMpIyAClg88bAZCXuXMGJG0GA1eIBpdh2PlnFGoRBnKqLy2YZUSTmZ3wJfC7tzfHtuC6FEhQ==;EndpointSuffix=core.windows.net"
--- a/docker-compose.yml
+++ b/docker-compose.yml
@ -1,32 +0,0 @@
 version: '2'
 services:
  minio:
    image: minio/minio
    ports:
      - "9000:9000"
    environment:
      - MINIO_ROOT_PASSWORD=password
      - MINIO_ROOT_USER=root
    volumes:
      - ./data/minio_store:/data
    command: server /data
    network_mode: "bridge" 
  rabbitmq:
    image: docker.io/bitnami/rabbitmq:3.9
    ports:
      - '4369:4369'
      - '5551:5551'
      - '5552:5552'
      - '5672:5672'
      - '25672:25672'
      - '15672:15672'
    environment:
      - RABBITMQ_SECURE_PASSWORD=yes
      - RABBITMQ_VM_MEMORY_HIGH_WATERMARK=100%
      - RABBITMQ_DISK_FREE_ABSOLUTE_LIMIT=20Gi
    network_mode: "bridge"
    volumes:
      - /opt/bitnami/rabbitmq/.rabbitmq/:/data/bitnami
 volumes:
  mdata:
--- a/pyinfra/exceptions.py
+++ b/pyinfra/exceptions.py
@ -1,34 +0,0 @@
 class AnalysisFailure(Exception):
    pass
 class DataLoadingFailure(Exception):
    pass
 class ProcessingFailure(Exception):
    pass
 class UnknownStorageBackend(ValueError):
    pass
 class InvalidEndpoint(ValueError):
    pass
 class UnknownClient(ValueError):
    pass
 class ConsumerError(Exception):
    pass
 class NoSuchContainer(KeyError):
    pass
 class IntentionalTestException(RuntimeError):
    pass
--- a/pyinfra/flask.py
+++ b/pyinfra/flask.py
@ -1,64 +0,0 @@
 import logging
 import requests
 from flask import Flask, jsonify
 from waitress import serve
 from pyinfra.config import CONFIG
 logger = logging.getLogger(__file__)
 logger.setLevel(CONFIG.service.logging_level)
 def run_probing_webserver(app, host=None, port=None, mode=None):
    if not host:
        host = CONFIG.probing_webserver.host
    if not port:
        port = CONFIG.probing_webserver.port
    if not mode:
        mode = CONFIG.probing_webserver.mode
    if mode == "development":
        app.run(host=host, port=port, debug=True)
    elif mode == "production":
        serve(app, host=host, port=port)
 def set_up_probing_webserver():
    # TODO: implement meaningful checks
    app = Flask(__name__)
    informed_about_missing_prometheus_endpoint = False
    @app.route("/ready", methods=["GET"])
    def ready():
        resp = jsonify("OK")
        resp.status_code = 200
        return resp
    @app.route("/health", methods=["GET"])
    def healthy():
        resp = jsonify("OK")
        resp.status_code = 200
        return resp
    @app.route("/prometheus", methods=["GET"])
    def get_metrics_from_analysis_endpoint():
        nonlocal informed_about_missing_prometheus_endpoint
        try:
            resp = requests.get(f"{CONFIG.rabbitmq.callback.analysis_endpoint}/prometheus")
            resp.raise_for_status()
        except ConnectionError:
            return ""
        except requests.exceptions.HTTPError as err:
            if resp.status_code == 404:
                if not informed_about_missing_prometheus_endpoint:
                    logger.warning(f"Got no metrics from analysis prometheus endpoint: {err}")
                    informed_about_missing_prometheus_endpoint = True
            else:
                logging.warning(f"Caught {err}")
        return resp.text
    return app
--- a/pyinfra/queue/consumer.py
+++ b/pyinfra/queue/consumer.py
@ -1,16 +0,0 @@
 from pyinfra.queue.queue_manager.queue_manager import QueueManager
 class Consumer:
    def __init__(self, callback, queue_manager: QueueManager):
        self.queue_manager = queue_manager
        self.callback = callback
    def consume_and_publish(self):
        self.queue_manager.consume_and_publish(self.callback)
    def basic_consume_and_publish(self):
        self.queue_manager.basic_consume_and_publish(self.callback)
    def consume(self, **kwargs):
        return self.queue_manager.consume(**kwargs)
--- a/pyinfra/queue/queue_manager/pika_queue_manager.py
+++ b/pyinfra/queue/queue_manager/pika_queue_manager.py
--- a/pyinfra/queue/queue_manager/init.py
+++ b/pyinfra/queue/queue_manager/init.py
--- a/pyinfra/queue/queue_manager/queue_manager.py
+++ b/pyinfra/queue/queue_manager/queue_manager.py
@ -1,51 +0,0 @@
 import abc
 class QueueHandle:
    def empty(self) -> bool:
        raise NotImplementedError
    def to_list(self) -> list:
        raise NotImplementedError
 class QueueManager(abc.ABC):
    def __init__(self, input_queue, output_queue):
        self._input_queue = input_queue
        self._output_queue = output_queue
    @abc.abstractmethod
    def publish_request(self, request):
        raise NotImplementedError
    @abc.abstractmethod
    def publish_response(self, response, callback):
        raise NotImplementedError
    @abc.abstractmethod
    def pull_request(self):
        raise NotImplementedError
    @abc.abstractmethod
    def consume(self, **kwargs):
        raise NotImplementedError
    @abc.abstractmethod
    def clear(self):
        raise NotImplementedError
    @abc.abstractmethod
    def input_queue(self) -> QueueHandle:
        raise NotImplementedError
    @abc.abstractmethod
    def output_queue(self) -> QueueHandle:
        raise NotImplementedError
    @abc.abstractmethod
    def consume_and_publish(self, callback):
        raise NotImplementedError
    @abc.abstractmethod
    def basic_consume_and_publish(self, callback):
        raise NotImplementedError
--- a/pyinfra/utils/banner.py
+++ b/pyinfra/utils/banner.py
@ -1,21 +0,0 @@
 import logging
 from pyinfra.locations import BANNER_FILE
 def show_banner():
    with open(BANNER_FILE) as f:
        banner = "\n" + "".join(f.readlines()) + "\n"
    logger = logging.getLogger(__name__)
    logger.propagate = False
    handler = logging.StreamHandler()
    handler.setLevel(logging.INFO)
    formatter = logging.Formatter("")
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    logger.info(banner)
--- a/pyinfra/visitor.py
+++ b/pyinfra/visitor.py
@ -1,91 +0,0 @@
 import abc
 import gzip
 import json
 import logging
 from operator import itemgetter
 from typing import Callable
 from pyinfra.config import CONFIG, parse_disjunction_string
 from pyinfra.exceptions import DataLoadingFailure
 from pyinfra.storage.storage import Storage
 def get_object_name(body):
    dossier_id, file_id, target_file_extension = itemgetter("dossierId", "fileId", "targetFileExtension")(body)
    object_name = f"{dossier_id}/{file_id}.{target_file_extension}"
    return object_name
 def get_response_object_name(body):
    dossier_id, file_id, response_file_extension = itemgetter("dossierId", "fileId", "responseFileExtension")(body)
    object_name = f"{dossier_id}/{file_id}.{response_file_extension}"
    return object_name
 def get_object_descriptor(body):
    return {"bucket_name": parse_disjunction_string(CONFIG.storage.bucket), "object_name": get_object_name(body)}
 def get_response_object_descriptor(body):
    return {
        "bucket_name": parse_disjunction_string(CONFIG.storage.bucket),
        "object_name": get_response_object_name(body),
    }
 class ResponseStrategy(abc.ABC):
    @abc.abstractmethod
    def handle_response(self, body):
        pass
    def __call__(self, body):
        return self.handle_response(body)
 class StorageStrategy(ResponseStrategy):
    def __init__(self, storage):
        self.storage = storage
    def handle_response(self, body):
        self.storage.put_object(**get_response_object_descriptor(body), data=gzip.compress(json.dumps(body).encode()))
        body.pop("data")
        return body
 class ForwardingStrategy(ResponseStrategy):
    def handle_response(self, body):
        return body
 class QueueVisitor:
    def __init__(self, storage: Storage, callback: Callable, response_strategy):
        self.storage = storage
        self.callback = callback
        self.response_strategy = response_strategy
    def load_data(self, body):
        def download():
            logging.debug(f"Downloading {object_descriptor}...")
            data = self.storage.get_object(**object_descriptor)
            logging.debug(f"Downloaded {object_descriptor}.")
            return data
        object_descriptor = get_object_descriptor(body)
        try:
            return gzip.decompress(download())
        except Exception as err:
            logging.warning(f"Loading data from storage failed for {object_descriptor}.")
            raise DataLoadingFailure from err
    def process_data(self, data, body):
        return self.callback({**body, "data": data})
    def load_and_process(self, body):
        data = self.process_data(self.load_data(body), body)
        result_body = {**body, "data": data}
        return result_body
    def __call__(self, body):
        result_body = self.load_and_process(body)
        return self.response_strategy(result_body)
--- a/requirements.txt
+++ b/requirements.txt
@ -1,15 +1,8 @@
 pika==1.2.0
 retry==0.9.2
 envyaml==1.10.211231
 minio==7.1.3
 Flask==2.1.1
 waitress==2.0.0
 azure-core==1.22.1
 azure-storage-blob==12.9.0
 requests==2.27.1
 testcontainers==3.4.2
 docker-compose==1.29.2
 tqdm==4.62.3
 pytest~=7.0.1
 funcy==1.17
 fpdf==1.7.2
--- a/run_tests.sh
+++ b/run_tests.sh
@ -1,9 +0,0 @@
 echo "${bamboo_nexus_password}" | docker login --username "${bamboo_nexus_user}" --password-stdin nexus.iqser.com:5001
 docker build -f Dockerfile_tests -t pyinfra-tests .
 rnd=$(date +"%s")
 name=pyinfra-tests-${rnd}
 echo "running tests container"
 docker run --rm --net=host --name $name -v $PWD:$PWD -w $PWD -v /var/run/docker.sock:/var/run/docker.sock pyinfra-tests
--- a/scripts/manage_minio.py
+++ b/scripts/manage_minio.py
@ -1,72 +0,0 @@
 import argparse
 import gzip
 import os
 from pathlib import Path
 from tqdm import tqdm
 from pyinfra.config import CONFIG, parse_disjunction_string
 from pyinfra.storage.storages import get_s3_storage
 def parse_args():
    parser = argparse.ArgumentParser()
    subparsers = parser.add_subparsers(help="sub-command help", dest="command")
    parser_add = subparsers.add_parser("add", help="Add file(s) to the MinIO store")
    parser_add.add_argument("dossier_id")
    add_group = parser_add.add_mutually_exclusive_group(required=True)
    add_group.add_argument("--file", "-f")
    add_group.add_argument("--directory", "-d")
    subparsers.add_parser("purge", help="Delete all files and buckets in the MinIO store")
    args = parser.parse_args()
    return args
 def combine_dossier_id_and_file_id_and_extension(dossier_id, file_id, extension):
    return f"{dossier_id}/{file_id}{extension}"
 def upload_compressed_response(storage, bucket_name, dossier_id, file_id, result) -> None:
    data = gzip.compress(result.encode())
    path_gz = combine_dossier_id_and_file_id_and_extension(dossier_id, file_id, CONFIG.service.response.extension)
    storage.put_object(bucket_name, path_gz, data)
 def add_file_compressed(storage, bucket_name, dossier_id, path) -> None:
    if Path(path).suffix == ".pdf":
        suffix_gz = ".ORIGIN.pdf.gz"
    if Path(path).suffix == ".json":
        suffix_gz = ".TEXT.json.gz"
    path_gz = combine_dossier_id_and_file_id_and_extension(dossier_id, Path(path).stem, suffix_gz)
    with open(path, "rb") as f:
        data = gzip.compress(f.read())
        storage.put_object(bucket_name, path_gz, data)
 if __name__ == "__main__":
    storage = get_s3_storage()
    bucket_name = parse_disjunction_string(CONFIG.storage.bucket)
    if not storage.has_bucket(bucket_name):
        storage.make_bucket(bucket_name)
    args = parse_args()
    if args.command == "add":
        if args.file:
            add_file_compressed(storage, bucket_name, args.dossier_id, args.file)
        elif args.directory:
            for fname in tqdm([*os.listdir(args.directory)], desc="Adding files"):
                path = Path(args.directory) / fname
                add_file_compressed(storage, bucket_name, args.dossier_id, path)
    elif args.command == "purge":
        storage.clear_bucket(bucket_name)
--- a/scripts/mock_client.py
+++ b/scripts/mock_client.py
@ -1,88 +0,0 @@
 import argparse
 import json
 import pika
 from pyinfra.config import CONFIG, parse_disjunction_string
 from pyinfra.storage.storages import get_s3_storage
 def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument("--bucket_name", "-b", required=True)
    parser.add_argument("--analysis_container", "-a", choices=["detr", "ner", "image", "dl_error"], required=True)
    args = parser.parse_args()
    return args
 def read_connection_params():
    credentials = pika.PlainCredentials(CONFIG.rabbitmq.user, CONFIG.rabbitmq.password)
    parameters = pika.ConnectionParameters(
        host=CONFIG.rabbitmq.host,
        port=CONFIG.rabbitmq.port,
        heartbeat=CONFIG.rabbitmq.heartbeat,
        credentials=credentials,
    )
    return parameters
 def make_channel(connection) -> pika.adapters.blocking_connection.BlockingChannel:
    channel = connection.channel()
    channel.basic_qos(prefetch_count=1)
    return channel
 def declare_queue(channel, queue: str):
    args = {"x-dead-letter-exchange": "", "x-dead-letter-routing-key": CONFIG.rabbitmq.queues.dead_letter}
    return channel.queue_declare(queue=queue, auto_delete=False, durable=True, arguments=args)
 def make_connection() -> pika.BlockingConnection:
    parameters = read_connection_params()
    connection = pika.BlockingConnection(parameters)
    return connection
 def build_message_bodies(analyse_container_type, bucket_name):
    def update_message(message_dict):
        if analyse_container_type == "detr" or analyse_container_type == "image":
            message_dict.update({"targetFileExtension": "ORIGIN.pdf.gz", "responseFileExtension": "IMAGE_INFO.json.gz"})
        if analyse_container_type == "dl_error":
            message_dict.update({"targetFileExtension": "no_such_file", "responseFileExtension": "IMAGE_INFO.json.gz"})
        if analyse_container_type == "ner":
            message_dict.update(
                {"targetFileExtension": "TEXT.json.gz", "responseFileExtension": "NER_ENTITIES.json.gz"}
            )
        return message_dict
    storage = get_s3_storage()
    for bucket_name, pdf_name in storage.get_all_object_names(bucket_name):
        if "pdf" not in pdf_name:
            continue
        file_id = pdf_name.split(".")[0]
        dossier_id, file_id = file_id.split("/")
        message_dict = {"dossierId": dossier_id, "fileId": file_id}
        update_message(message_dict)
        yield json.dumps(message_dict).encode()
 def main(args):
    connection = make_connection()
    channel = make_channel(connection)
    declare_queue(channel, CONFIG.rabbitmq.queues.input)
    declare_queue(channel, CONFIG.rabbitmq.queues.output)
    for body in build_message_bodies(args.analysis_container, args.bucket_name):
        channel.basic_publish("", CONFIG.rabbitmq.queues.input, body)
        print(f"Put {body} on {CONFIG.rabbitmq.queues.input}")
    for method_frame, _, body in channel.consume(queue=CONFIG.rabbitmq.queues.output, inactivity_timeout=1):
        if not body:
            break
        print(f"Received {json.loads(body)}")
        channel.basic_ack(method_frame.delivery_tag)
    channel.close()
 if __name__ == "__main__":
    main(parse_args())
--- a/src/serve.py
+++ b/src/serve.py
@ -1,84 +0,0 @@
 import logging
 from multiprocessing import Process
 import requests
 from retry import retry
 from pyinfra.config import CONFIG
 from pyinfra.exceptions import AnalysisFailure, ConsumerError
 from pyinfra.flask import run_probing_webserver, set_up_probing_webserver
 from pyinfra.queue.consumer import Consumer
 from pyinfra.queue.queue_manager.pika_queue_manager import PikaQueueManager
 from pyinfra.storage.storages import get_storage
 from pyinfra.utils.banner import show_banner
 from pyinfra.visitor import QueueVisitor, StorageStrategy
 def make_callback(analysis_endpoint):
    def callback(message):
        def perform_operation(operation):
            endpoint = f"{analysis_endpoint}/{operation}"
            try:
                logging.debug(f"Requesting analysis from {endpoint}...")
                analysis_response = requests.post(endpoint, data=message["data"])
                analysis_response.raise_for_status()
                analysis_response = analysis_response.json()
                logging.debug(f"Received response.")
                return analysis_response
            except Exception as err:
                logging.warning(f"Exception caught when calling analysis endpoint {endpoint}.")
                raise AnalysisFailure() from err
        operations = message.get("operations", ["/"])
        results = map(perform_operation, operations)
        result = dict(zip(operations, results))
        if list(result.keys()) == ["/"]:
            result = list(result.values())[0]
        return result
    return callback
 def main():
    logger = logging.getLogger("main")
    show_banner()
    webserver = Process(target=run_probing_webserver, args=(set_up_probing_webserver(),))
    logging.info("Starting webserver...")
    webserver.start()
    callback = make_callback(CONFIG.rabbitmq.callback.analysis_endpoint)
    storage = get_storage(CONFIG.storage.backend)
    response_strategy = StorageStrategy(storage)
    visitor = QueueVisitor(storage, callback, response_strategy)
    @retry(ConsumerError, tries=3, delay=5, jitter=(1, 3))
    def consume():
        try:  # RED-4049 queue manager needs to be in try scope to eventually throw Exception after connection loss.
            queue_manager = PikaQueueManager(CONFIG.rabbitmq.queues.input, CONFIG.rabbitmq.queues.output)
            consumer = Consumer(visitor, queue_manager)
            consumer.basic_consume_and_publish()
        except Exception as err:
            logger.exception(err)
            raise ConsumerError from err
    try:
        consume()
    except KeyboardInterrupt:
        pass
    except ConsumerError:
        webserver.terminate()
        raise
    webserver.join()
 if __name__ == "__main__":
    logging_level = CONFIG.service.logging_level
    logging.basicConfig(level=logging_level)
    logging.getLogger("pika").setLevel(logging.ERROR)
    logging.getLogger("flask").setLevel(logging.ERROR)
    logging.getLogger("urllib3").setLevel(logging.ERROR)
    main()
--- a/test/config.py
+++ b/test/config.py
@ -1,5 +0,0 @@
 from pyinfra.config import Config
 from pyinfra.locations import TEST_CONFIG_FILE
 CONFIG = Config(TEST_CONFIG_FILE)
--- a/test/config.yaml
+++ b/test/config.yaml
@ -1,25 +0,0 @@
 storage:
  minio:
    endpoint: "http://127.0.0.1:9000"
    access_key: root
    secret_key: password
    region: null
  aws:
    endpoint: https://s3.amazonaws.com
    access_key: AKIA4QVP6D4LCDAGYGN2
    secret_key: 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
    region: $STORAGE_REGION|"eu-west-1"
  azure:
    connection_string: "DefaultEndpointsProtocol=https;AccountName=iqserdevelopment;AccountKey=4imAbV9PYXaztSOMpIyAClg88bAZCXuXMGJG0GA1eIBpdh2PlnFGoRBnKqLy2YZUSTmZ3wJfC7tzfHtuC6FEhQ==;EndpointSuffix=core.windows.net"
  bucket: "pyinfra-test-bucket"
 webserver:
  host: $SERVER_HOST|"127.0.0.1"  # webserver address
  port: $SERVER_PORT|5000  # webserver port
  mode: $SERVER_MODE|production  # webserver mode: {development, production}
 mock_analysis_endpoint: "http://127.0.0.1:5000"
--- a/test/exploration_tests/init.py
+++ b/test/exploration_tests/init.py
--- a/test/exploration_tests/data_json_request_test.py
+++ b/test/exploration_tests/data_json_request_test.py
@ -1,79 +0,0 @@
 import json
 from operator import itemgetter
 import pytest
 from flask import Flask, request, jsonify
 import fpdf
 def set_up_processing_server():
    app = Flask(__name__)
    @app.route("/ready", methods=["GET"])
    def ready():
        resp = jsonify("OK")
        resp.status_code = 200
        return resp
    @app.route("/process", methods=["POST"])
    def process():
        payload = json.loads(request.json)
        data = payload["data"].encode()
        metadata = payload["metadata"]
        response_payload = {"metadata_type": str(type(metadata)), "data_type": str(type(data))}
        return jsonify(response_payload)
    return app
@pytest.fixture
 def server():
    server = set_up_processing_server()
    server.config.update({"TESTING": True})
    return server
@pytest.fixture
 def client(server):
    return server.test_client()
 def test_server_ready_check(client):
    response = client.get("/ready")
    assert response.status_code == 200
    assert response.json == "OK"
@pytest.mark.parametrize("data_type", ["pdf", "bytestring"])
 def test_sending_bytes_through_json(client, data):
    payload = {"data": data.decode("latin1"), "metadata": {"A": 1, "B": [2, 3]}}
    response = client.post("/process", json=json.dumps(payload))
    response_payload = response.json
    data_type, metadata_type = itemgetter("data_type", "metadata_type")(response_payload)
    assert data_type == "<class 'bytes'>"
    assert metadata_type == "<class 'dict'>"
@pytest.fixture
 def pdf():
    pdf = fpdf.FPDF(unit="pt")
    pdf.add_page()
    return pdf_stream(pdf)
 def pdf_stream(pdf: fpdf.fpdf.FPDF):
    return pdf.output(dest="S").encode("latin1")
@pytest.fixture
 def data(data_type, pdf):
    if data_type == "pdf":
        return pdf
    elif data_type == "bytestring":
        return "content".encode("latin1")
--- a/test/queue/init.py
+++ b/test/queue/init.py
--- a/test/queue/queue_manager_mock.py
+++ b/test/queue/queue_manager_mock.py
@ -1,46 +0,0 @@
 from pyinfra.queue.queue_manager.queue_manager import QueueManager, QueueHandle
 from test.queue.queue_mock import QueueMock
 def monkey_patch_queue_handle(queue) -> QueueHandle:
    queue_handle = QueueHandle()
    queue_handle.empty = lambda: not queue
    queue_handle.to_list = lambda: list(queue)
    return queue_handle
 class QueueManagerMock(QueueManager):
    def __init__(self, input_queue, output_queue):
        super().__init__(QueueMock(), QueueMock())
    def publish_request(self, request):
        self._input_queue.append(request)
    def publish_response(self, message, callback):
        self._output_queue.append(callback(message))
    def pull_request(self):
        return self._input_queue.popleft()
    def consume(self, **kwargs):
        while self._input_queue:
            yield self.pull_request()
    def consume_and_publish(self, callback):
        for message in self.consume():
            self.publish_response(message, callback)
    def basic_consume_and_publish(self, callback):
        raise NotImplementedError
    def clear(self):
        self._input_queue.clear()
        self._output_queue.clear()
    @property
    def input_queue(self) -> QueueHandle:
        return monkey_patch_queue_handle(self._input_queue)
    @property
    def output_queue(self) -> QueueHandle:
        return monkey_patch_queue_handle(self._output_queue)
--- a/test/queue/queue_mock.py
+++ b/test/queue/queue_mock.py
@ -1,5 +0,0 @@
 from collections import deque
 class QueueMock(deque):
    pass
--- a/test/storage/init.py
+++ b/test/storage/init.py
--- a/test/storage/adapter_mock.py
+++ b/test/storage/adapter_mock.py
@ -1,30 +0,0 @@
 from pyinfra.storage.adapters.adapter import StorageAdapter
 from test.storage.client_mock import StorageClientMock
 class StorageAdapterMock(StorageAdapter):
    def __init__(self, client: StorageClientMock):
        assert isinstance(client, StorageClientMock)
        super().__init__(client=client)
        self.__client = self._StorageAdapter__client
    def make_bucket(self, bucket_name):
        self.__client.make_bucket(bucket_name)
    def has_bucket(self, bucket_name):
        return self.__client.has_bucket(bucket_name)
    def put_object(self, bucket_name, object_name, data):
        return self.__client.put_object(bucket_name, object_name, data)
    def get_object(self, bucket_name, object_name):
        return self.__client.get_object(bucket_name, object_name)
    def get_all_objects(self, bucket_name):
        return self.__client.get_all_objects(bucket_name)
    def clear_bucket(self, bucket_name):
        return self.__client.clear_bucket(bucket_name)
    def get_all_object_names(self, bucket_name):
        return self.__client.get_all_object_names(bucket_name)
--- a/test/storage/client_mock.py
+++ b/test/storage/client_mock.py
@ -1,27 +0,0 @@
 from itertools import repeat
 class StorageClientMock:
    def __init__(self):
        self.__data = {}
    def make_bucket(self, bucket_name):
        self.__data[bucket_name] = {}
    def has_bucket(self, bucket_name):
        return bucket_name in self.__data
    def put_object(self, bucket_name, object_name, data):
        self.__data[bucket_name][object_name] = data
    def get_object(self, bucket_name, object_name):
        return self.__data[bucket_name][object_name]
    def get_all_objects(self, bucket_name):
        return self.__data[bucket_name].values()
    def clear_bucket(self, bucket_name):
        self.__data[bucket_name] = {}
    def get_all_object_names(self, bucket_name):
        return zip(repeat(bucket_name), self.__data[bucket_name])
--- a/test/unit_tests/init.py
+++ b/test/unit_tests/init.py
--- a/test/unit_tests/azure_adapter_test.py
+++ b/test/unit_tests/azure_adapter_test.py
@ -1,10 +0,0 @@
 import pytest
 from pyinfra.storage.adapters.azure import AzureStorageAdapter
 from test.storage.client_mock import StorageClientMock
@pytest.fixture
 def adapter():
    adapter = AzureStorageAdapter(StorageClientMock())
    return adapter
--- a/test/unit_tests/config_test.py
+++ b/test/unit_tests/config_test.py
@ -1,45 +0,0 @@
 import os
 import tempfile
 import pytest
 import yaml
 from pyinfra.config import Config, parse_disjunction_string
@pytest.fixture
 def config_file_content():
    return {"A": [{"B": [1, 2]}, {"C": 3}, 4], "D": {"E": {"F": True}}}
@pytest.fixture
 def config(config_file_content):
    with tempfile.NamedTemporaryFile(suffix=".yaml", mode="w") as f:
        yaml.dump(config_file_content, f, default_flow_style=False)
        yield Config(f.name)
 def test_dot_access_key_exists(config):
    assert config.A == [{"B": [1, 2]}, {"C": 3}, 4]
    assert config.D.E["F"]
 def test_access_key_exists(config):
    assert config["A"] == [{"B": [1, 2]}, {"C": 3}, 4]
    assert config["A"][0] == {"B": [1, 2]}
    assert config["A"][0]["B"] == [1, 2]
    assert config["A"][0]["B"][0] == 1
 def test_dot_access_key_does_not_exists(config):
    assert config.B is None
 def test_access_key_does_not_exists(config):
    assert config["B"] is None
 def test_parse_disjunction_string():
    assert parse_disjunction_string("A|Bb|c") == "c"
    os.environ["Bb"] = "d"
    assert parse_disjunction_string("A|Bb|c") == "d"
--- a/test/unit_tests/conftest.py
+++ b/test/unit_tests/conftest.py
@ -1,160 +0,0 @@
 import json
 import logging
 import time
 from unittest.mock import Mock
 import pika
 import pytest
 import testcontainers.compose
 from pyinfra.exceptions import UnknownClient
 from pyinfra.locations import TEST_DIR, COMPOSE_PATH
 from pyinfra.queue.queue_manager.pika_queue_manager import PikaQueueManager, get_connection_params
 from pyinfra.queue.queue_manager.queue_manager import QueueManager
 from pyinfra.storage.adapters.azure import AzureStorageAdapter
 from pyinfra.storage.adapters.s3 import S3StorageAdapter
 from pyinfra.storage.clients.azure import get_azure_client
 from pyinfra.storage.clients.s3 import get_s3_client
 from pyinfra.storage.storage import Storage
 from test.config import CONFIG
 from test.queue.queue_manager_mock import QueueManagerMock
 from test.storage.adapter_mock import StorageAdapterMock
 from test.storage.client_mock import StorageClientMock
 from pyinfra.visitor import StorageStrategy, ForwardingStrategy, QueueVisitor
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.DEBUG)
@pytest.fixture(scope="session")
 def bucket_name():
    return "pyinfra-test-bucket"
@pytest.fixture
 def storage_data():
    with open(f"{TEST_DIR}/test_data/test_data.TEXT.json", "r") as f:
        data = json.load(f)
        return data
@pytest.fixture
 def mock_response(storage_data):
    response = Mock(status_code=200)
    response.json.return_value = storage_data
    return response
@pytest.fixture
 def mock_payload():
    return json.dumps({"dossierId": "test", "fileId": "test"})
@pytest.fixture
 def mock_make_load_data():
    def load_data(payload):
        return storage_data
    return load_data
@pytest.fixture(params=["minio", "aws"], scope="session")
 def storage(client_name, bucket_name, request):
    logger.debug("Setup for storage")
    storage = Storage(get_adapter(client_name, request.param))
    storage.make_bucket(bucket_name)
    storage.clear_bucket(bucket_name)
    yield storage
    logger.debug("Teardown for storage")
    storage.clear_bucket(bucket_name)
@pytest.fixture(scope="session", autouse=True)
 def docker_compose(sleep_seconds=30):
    logger.info(f"Starting docker containers with {COMPOSE_PATH}/docker-compose.yml...")
    compose = testcontainers.compose.DockerCompose(COMPOSE_PATH, compose_file_name="docker-compose.yml")
    compose.start()
    logger.info(f"Sleeping for {sleep_seconds} seconds to wait for containers to finish startup... ")
    time.sleep(sleep_seconds)
    yield compose
    compose.stop()
 def get_pika_connection_params():
    params = get_connection_params()
    return params
 def get_s3_params(s3_backend):
    params = CONFIG.storage[s3_backend]
    return params
 def get_adapter(client_name, s3_backend):
    if client_name == "mock":
        return StorageAdapterMock(StorageClientMock())
    if client_name == "azure":
        return AzureStorageAdapter(get_azure_client(CONFIG.storage.azure.connection_string))
    if client_name == "s3":
        return S3StorageAdapter(get_s3_client(get_s3_params(s3_backend)))
    else:
        raise UnknownClient(client_name)
 def get_queue_manager(queue_manager_name) -> QueueManager:
    if queue_manager_name == "mock":
        return QueueManagerMock("input", "output")
    if queue_manager_name == "pika":
        return PikaQueueManager("input", "output", connection_params=get_pika_connection_params())
@pytest.fixture(scope="session")
 def queue_manager(queue_manager_name):
    def close_connections():
        if queue_manager_name == "pika":
            try:
                queue_manager.connection.close()
            except (pika.exceptions.StreamLostError, pika.exceptions.ConnectionWrongStateError, ConnectionResetError):
                logger.debug("Connection was already closed when attempting to close explicitly.")
    def close_channel():
        if queue_manager_name == "pika":
            try:
                queue_manager.channel.close()
            except pika.exceptions.ChannelWrongStateError:
                logger.debug("Channel was already closed when attempting to close explicitly.")
    queue_manager = get_queue_manager(queue_manager_name)
    yield queue_manager
    close_connections()
    close_channel()
@pytest.fixture(scope="session")
 def callback():
    def inner(request):
        return request["data"].decode() * 2
    return inner
@pytest.fixture
 def analysis_callback(callback):
    def inner(request):
        return callback(request)
    return inner
@pytest.fixture
 def response_strategy(response_strategy_name, storage):
    if response_strategy_name == "storage":
        return StorageStrategy(storage)
    if response_strategy_name == "forwarding":
        return ForwardingStrategy()
@pytest.fixture()
 def visitor(storage, analysis_callback, response_strategy):
    return QueueVisitor(storage, analysis_callback, response_strategy)
--- a/test/unit_tests/consumer_test.py
+++ b/test/unit_tests/consumer_test.py
@ -1,126 +0,0 @@
 import gzip
 import json
 import logging
 from operator import itemgetter
 import pytest
 from pyinfra.exceptions import ProcessingFailure
 from pyinfra.queue.consumer import Consumer
 from pyinfra.visitor import get_object_descriptor, ForwardingStrategy
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.DEBUG)
@pytest.fixture(scope="session")
 def consumer(queue_manager, callback):
    return Consumer(callback, queue_manager)
@pytest.fixture(scope="session")
 def access_callback():
    return itemgetter("fileId")
@pytest.fixture()
 def items():
    def inner():
        for i in range(3):
            body = {
                "dossierId": "folder",
                "fileId": f"file{i}",
                "targetFileExtension": "in.gz",
                "responseFileExtension": "out.gz",
            }
            yield f"{i}".encode(), body
    return list(inner())
 class TestConsumer:
    @pytest.mark.parametrize("queue_manager_name", ["mock", "pika"], scope="session")
    def test_consuming_empty_input_queue_does_not_put_anything_on_output_queue(self, consumer, queue_manager):
        queue_manager.clear()
        consumer.consume()
        assert queue_manager.output_queue.empty()
    @pytest.mark.parametrize("queue_manager_name", ["mock", "pika"], scope="session")
    def test_consuming_nonempty_input_queue_puts_messages_on_output_queue_in_fifo_order(
        self, consumer, queue_manager, callback
    ):
        def produce_items():
            return map(str, range(3))
        def mock_visitor(callback):
            def inner(data):
                return callback({"data": data.encode()})
            return inner
        callback = mock_visitor(callback)
        queue_manager.clear()
        for item in produce_items():
            queue_manager.publish_request(item)
        requests = consumer.consume()
        for _, r in zip(produce_items(), requests):
            queue_manager.publish_response(r, callback)
        assert queue_manager.output_queue.to_list() == ["00", "11", "22"]
    @pytest.mark.parametrize("queue_manager_name", ["mock", "pika"], scope="session")
    @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, items
    ):
        visitor.response_strategy = ForwardingStrategy()
        queue_manager.clear()
        storage.clear_bucket(bucket_name)
        for data, message in items:
            storage.put_object(**get_object_descriptor(message), data=gzip.compress(data))
            queue_manager.publish_request(message)
        requests = consumer.consume(inactivity_timeout=5)
        for itm, req in zip(items, requests):
            logger.debug(f"Processing item {itm}")
            queue_manager.publish_response(req, visitor)
        assert list(map(itemgetter("data"), queue_manager.output_queue.to_list())) == ["00", "11", "22"]
    @pytest.mark.parametrize("queue_manager_name", ["pika"], scope="session")
    def test_message_is_republished_when_callback_raises_processing_failure_exception(
        self, consumer, queue_manager, bucket_name, items
    ):
        class DebugError(Exception):
            pass
        def callback(_):
            raise ProcessingFailure()
        def reject_patch(*args, **kwargs):
            raise DebugError()
        queue_manager.reject = reject_patch
        queue_manager.clear()
        for data, message in items:
            queue_manager.publish_request(message)
        requests = consumer.consume()
        logger = logging.getLogger("pyinfra.queue.queue_manager.pika_queue_manager")
        logger.addFilter(lambda record: False)
        with pytest.raises(DebugError):
            while True:
                queue_manager.publish_response(next(requests), callback)
--- a/test/unit_tests/queue_visitor_test.py
+++ b/test/unit_tests/queue_visitor_test.py
@ -1,38 +0,0 @@
 import gzip
 import json
 import pytest
 from pyinfra.visitor import get_object_descriptor, get_response_object_descriptor
@pytest.fixture()
 def body():
    return {"dossierId": "folder", "fileId": "file", "targetFileExtension": "in.gz", "responseFileExtension": "out.gz"}
@pytest.mark.parametrize("client_name", ["mock", "azure", "s3"], scope="session")
 class TestVisitor:
    @pytest.mark.parametrize("response_strategy_name", ["forwarding", "storage"], scope="session")
    def test_given_a_input_queue_message_callback_pulls_the_data_from_storage(
        self, visitor, body, storage, bucket_name
    ):
        storage.clear_bucket(bucket_name)
        storage.put_object(**get_object_descriptor(body), data=gzip.compress(b"content"))
        data_received = visitor.load_data(body)
        assert b"content" == data_received
    @pytest.mark.parametrize("response_strategy_name", ["forwarding", "storage"], scope="session")
    def test_visitor_pulls_and_processes_data(self, visitor, body, storage, bucket_name):
        storage.clear_bucket(bucket_name)
        storage.put_object(**get_object_descriptor(body), data=gzip.compress("2".encode()))
        response_body = visitor.load_and_process(body)
        assert response_body["data"] == "22"
    @pytest.mark.parametrize("response_strategy_name", ["storage"], scope="session")
    def test_visitor_puts_response_on_storage(self, visitor, body, storage, bucket_name):
        storage.clear_bucket(bucket_name)
        storage.put_object(**get_object_descriptor(body), data=gzip.compress("2".encode()))
        response_body = visitor(body)
        assert "data" not in response_body
        assert json.loads(gzip.decompress(storage.get_object(**get_response_object_descriptor(body))))["data"] == "22"
--- a/test/unit_tests/storage_test.py
+++ b/test/unit_tests/storage_test.py
@ -1,52 +0,0 @@
 import logging
 import pytest
 from pyinfra.exceptions import DataLoadingFailure
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.DEBUG)
@pytest.mark.parametrize("client_name", ["mock", "azure", "s3"], scope="session")
 class TestStorage:
    def test_clearing_bucket_yields_empty_bucket(self, storage, bucket_name):
        storage.clear_bucket(bucket_name)
        data_received = storage.get_all_objects(bucket_name)
        assert not {*data_received}
    def test_getting_object_put_in_bucket_is_object(self, storage, bucket_name):
        storage.clear_bucket(bucket_name)
        storage.put_object(bucket_name, "file", b"content")
        data_received = storage.get_object(bucket_name, "file")
        assert b"content" == data_received
    def test_getting_nested_object_put_in_bucket_is_nested_object(self, storage, bucket_name):
        storage.clear_bucket(bucket_name)
        storage.put_object(bucket_name, "folder/file", b"content")
        data_received = storage.get_object(bucket_name, "folder/file")
        assert b"content" == data_received
    def test_getting_objects_put_in_bucket_are_objects(self, storage, bucket_name):
        storage.clear_bucket(bucket_name)
        storage.put_object(bucket_name, "file1", b"content 1")
        storage.put_object(bucket_name, "folder/file2", b"content 2")
        data_received = storage.get_all_objects(bucket_name)
        assert {b"content 1", b"content 2"} == {*data_received}
    def test_make_bucket_produces_bucket(self, storage, bucket_name):
        storage.clear_bucket(bucket_name)
        storage.make_bucket(bucket_name)
        assert storage.has_bucket(bucket_name)
    def test_listing_bucket_files_yields_all_files_in_bucket(self, storage, bucket_name):
        storage.clear_bucket(bucket_name)
        storage.put_object(bucket_name, "file1", b"content 1")
        storage.put_object(bucket_name, "file2", b"content 2")
        full_names_received = storage.get_all_object_names(bucket_name)
        assert {(bucket_name, "file1"), (bucket_name, "file2")} == {*full_names_received}
    def test_data_loading_failure_raised_if_object_not_present(self, storage, bucket_name):
        storage.clear_bucket(bucket_name)
        with pytest.raises(DataLoadingFailure):
            storage.get_object(bucket_name, "folder/file")