diff --git a/Dockerfile b/Dockerfile deleted file mode 100755 index 4c06a71..0000000 --- a/Dockerfile +++ /dev/null @@ -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"] diff --git a/Dockerfile_tests b/Dockerfile_tests deleted file mode 100755 index fd1164f..0000000 --- a/Dockerfile_tests +++ /dev/null @@ -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 diff --git a/banner.txt b/banner.txt deleted file mode 100644 index fa0940f..0000000 --- a/banner.txt +++ /dev/null @@ -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-' \ No newline at end of file diff --git a/config.yaml b/config.yaml deleted file mode 100755 index a7f8efe..0000000 --- a/config.yaml +++ /dev/null @@ -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" diff --git a/docker-compose.yml b/docker-compose.yml deleted file mode 100755 index 04e5296..0000000 --- a/docker-compose.yml +++ /dev/null @@ -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: - diff --git a/pyinfra/exceptions.py b/pyinfra/exceptions.py deleted file mode 100644 index 56405d4..0000000 --- a/pyinfra/exceptions.py +++ /dev/null @@ -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 diff --git a/pyinfra/flask.py b/pyinfra/flask.py deleted file mode 100644 index 549ef0c..0000000 --- a/pyinfra/flask.py +++ /dev/null @@ -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 diff --git a/pyinfra/queue/consumer.py b/pyinfra/queue/consumer.py deleted file mode 100644 index 1072178..0000000 --- a/pyinfra/queue/consumer.py +++ /dev/null @@ -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) diff --git a/pyinfra/queue/queue_manager/pika_queue_manager.py b/pyinfra/queue/pika_queue_manager.py similarity index 100% rename from pyinfra/queue/queue_manager/pika_queue_manager.py rename to pyinfra/queue/pika_queue_manager.py diff --git a/pyinfra/queue/queue_manager/__init__.py b/pyinfra/queue/queue_manager/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/pyinfra/queue/queue_manager/queue_manager.py b/pyinfra/queue/queue_manager/queue_manager.py deleted file mode 100644 index d42bfc5..0000000 --- a/pyinfra/queue/queue_manager/queue_manager.py +++ /dev/null @@ -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 diff --git a/pyinfra/utils/banner.py b/pyinfra/utils/banner.py deleted file mode 100644 index 20d066e..0000000 --- a/pyinfra/utils/banner.py +++ /dev/null @@ -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) diff --git a/pyinfra/visitor.py b/pyinfra/visitor.py deleted file mode 100644 index 1fbdab0..0000000 --- a/pyinfra/visitor.py +++ /dev/null @@ -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) diff --git a/requirements.txt b/requirements.txt index e67c02b..1220e73 100755 --- 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 diff --git a/run_tests.sh b/run_tests.sh deleted file mode 100755 index 8a5a58f..0000000 --- a/run_tests.sh +++ /dev/null @@ -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 diff --git a/scripts/manage_minio.py b/scripts/manage_minio.py deleted file mode 100644 index 88e1127..0000000 --- a/scripts/manage_minio.py +++ /dev/null @@ -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) diff --git a/scripts/mock_client.py b/scripts/mock_client.py deleted file mode 100644 index db620d0..0000000 --- a/scripts/mock_client.py +++ /dev/null @@ -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()) diff --git a/src/serve.py b/src/serve.py deleted file mode 100644 index 435d632..0000000 --- a/src/serve.py +++ /dev/null @@ -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() diff --git a/test/config.py b/test/config.py deleted file mode 100644 index c8058f1..0000000 --- a/test/config.py +++ /dev/null @@ -1,5 +0,0 @@ -from pyinfra.config import Config -from pyinfra.locations import TEST_CONFIG_FILE - - -CONFIG = Config(TEST_CONFIG_FILE) diff --git a/test/config.yaml b/test/config.yaml deleted file mode 100644 index d404715..0000000 --- a/test/config.yaml +++ /dev/null @@ -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" \ No newline at end of file diff --git a/test/exploration_tests/__init__.py b/test/exploration_tests/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/test/exploration_tests/data_json_request_test.py b/test/exploration_tests/data_json_request_test.py deleted file mode 100644 index b53ee42..0000000 --- a/test/exploration_tests/data_json_request_test.py +++ /dev/null @@ -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 == "" - assert metadata_type == "" - - -@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") diff --git a/test/queue/__init__.py b/test/queue/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/test/queue/queue_manager_mock.py b/test/queue/queue_manager_mock.py deleted file mode 100644 index 063a68b..0000000 --- a/test/queue/queue_manager_mock.py +++ /dev/null @@ -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) diff --git a/test/queue/queue_mock.py b/test/queue/queue_mock.py deleted file mode 100644 index c138f29..0000000 --- a/test/queue/queue_mock.py +++ /dev/null @@ -1,5 +0,0 @@ -from collections import deque - - -class QueueMock(deque): - pass diff --git a/test/storage/__init__.py b/test/storage/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/test/storage/adapter_mock.py b/test/storage/adapter_mock.py deleted file mode 100644 index 5443ade..0000000 --- a/test/storage/adapter_mock.py +++ /dev/null @@ -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) diff --git a/test/storage/client_mock.py b/test/storage/client_mock.py deleted file mode 100644 index c81b02b..0000000 --- a/test/storage/client_mock.py +++ /dev/null @@ -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]) diff --git a/test/unit_tests/__init__.py b/test/unit_tests/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/test/unit_tests/azure_adapter_test.py b/test/unit_tests/azure_adapter_test.py deleted file mode 100644 index 6807ffb..0000000 --- a/test/unit_tests/azure_adapter_test.py +++ /dev/null @@ -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 diff --git a/test/unit_tests/config_test.py b/test/unit_tests/config_test.py deleted file mode 100644 index 5a5d610..0000000 --- a/test/unit_tests/config_test.py +++ /dev/null @@ -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" diff --git a/test/unit_tests/conftest.py b/test/unit_tests/conftest.py deleted file mode 100644 index 92d78cf..0000000 --- a/test/unit_tests/conftest.py +++ /dev/null @@ -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) diff --git a/test/unit_tests/consumer_test.py b/test/unit_tests/consumer_test.py deleted file mode 100644 index 5874846..0000000 --- a/test/unit_tests/consumer_test.py +++ /dev/null @@ -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) diff --git a/test/unit_tests/queue_visitor_test.py b/test/unit_tests/queue_visitor_test.py deleted file mode 100644 index c781429..0000000 --- a/test/unit_tests/queue_visitor_test.py +++ /dev/null @@ -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" diff --git a/test/unit_tests/storage_test.py b/test/unit_tests/storage_test.py deleted file mode 100644 index 90354d7..0000000 --- a/test/unit_tests/storage_test.py +++ /dev/null @@ -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")