diff --git a/pyinfra/file_descriptor_builder.py b/pyinfra/file_descriptor_builder.py index 3965802..5330b06 100644 --- a/pyinfra/file_descriptor_builder.py +++ b/pyinfra/file_descriptor_builder.py @@ -42,9 +42,11 @@ class RedFileDescriptorBuilder(FileDescriptorBuilder): file_pattern = self.operation2file_patterns[operation][end] + print(f"{end=}") + file_descriptor = { **project(queue_item_body, ["dossierId", "fileId", "pages"]), - "pages": queue_item_body.get("pages", []), + "pages": [queue_item_body["id"]] if end == "output" else [], "extension": file_pattern["extension"], "subdir": file_pattern["subdir"], } @@ -59,21 +61,25 @@ class RedFileDescriptorBuilder(FileDescriptorBuilder): "dossierId", "fileId", "subdir", "pages", "extension" )(file_descriptor) - matcher = os.path.join(dossier_id, make_filename(file_id, subdir, self.__build_page_regex(pages) + extension)) + matcher = os.path.join( + dossier_id, make_filename(file_id, subdir, self.__build_page_regex(pages, subdir) + extension) + ) return matcher @staticmethod - def __build_page_regex(pages): + def __build_page_regex(pages, subdir): n_pages = len(pages) + print("pages:", pages) if n_pages > 1: page_re = "id:(" + "|".join(map(str, pages)) + ")." elif n_pages == 1: page_re = f"id:{pages[0]}." - else: # no pages - page_re = r"id:\d+." + else: # no pages specified -> either all pages or no pages, depending on whether a subdir is specified + print("subdir", subdir) + page_re = r"id:\d+." if subdir else "" return page_re diff --git a/pyinfra/file_descriptor_manager.py b/pyinfra/file_descriptor_manager.py index 8c1072a..546402e 100644 --- a/pyinfra/file_descriptor_manager.py +++ b/pyinfra/file_descriptor_manager.py @@ -8,7 +8,8 @@ class FileDescriptorManager: TODO: This is supposed to be fully decoupled from the concrete file descriptor builder implementation, however some bad coupling is still left. - """ + """ + def __init__(self, bucket_name, file_descriptor_builder: FileDescriptorBuilder): self.bucket_name = bucket_name self.operation_file_descriptor_builder = file_descriptor_builder @@ -20,10 +21,7 @@ class FileDescriptorManager: return self.get_object_name(queue_item_body, end="output") def get_object_name(self, queue_item_body: dict, end): - file_descriptor = self.build_file_descriptor(queue_item_body, end=end) - # FIXME: bad coupling - file_descriptor["pages"] = [queue_item_body.get("id", 0)] object_name = self.__build_matcher(file_descriptor) return object_name @@ -48,7 +46,9 @@ class FileDescriptorManager: return self.get_object_descriptor(queue_item_body, end="input") def get_output_object_descriptor(self, storage_upload_info): - return self.get_object_descriptor(storage_upload_info, end="output") + r = self.get_object_descriptor(storage_upload_info, end="output") + print(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>", r) + return r def get_object_descriptor(self, queue_item_body, end): # TODO: this is complected with the Storage class API diff --git a/pyinfra/storage/adapters/azure.py b/pyinfra/storage/adapters/azure.py index 074a9d6..6e6edcf 100644 --- a/pyinfra/storage/adapters/azure.py +++ b/pyinfra/storage/adapters/azure.py @@ -5,7 +5,7 @@ from azure.storage.blob import ContainerClient, BlobServiceClient from pyinfra.storage.adapters.adapter import StorageAdapter -logger = logging.getLogger() +logger = logging.getLogger(__name__) logging.getLogger("azure").setLevel(logging.WARNING) logging.getLogger("urllib3").setLevel(logging.WARNING) diff --git a/pyinfra/storage/adapters/s3.py b/pyinfra/storage/adapters/s3.py index a7729df..15c17ce 100644 --- a/pyinfra/storage/adapters/s3.py +++ b/pyinfra/storage/adapters/s3.py @@ -8,7 +8,7 @@ from minio import Minio from pyinfra.exceptions import DataLoadingFailure from pyinfra.storage.adapters.adapter import StorageAdapter -logger = logging.getLogger() +logger = logging.getLogger(__name__) class S3StorageAdapter(StorageAdapter): @@ -27,6 +27,7 @@ class S3StorageAdapter(StorageAdapter): logger.debug(f"Uploading '{object_name}'...") data = io.BytesIO(data) self.__client.put_object(bucket_name, object_name, data, length=data.getbuffer().nbytes) + # input("uploaded") def get_object(self, bucket_name, object_name): logger.debug(f"Downloading '{object_name}'...") diff --git a/scripts/show_compressed_json.py b/scripts/show_compressed_json.py index 7fa2859..86ea339 100644 --- a/scripts/show_compressed_json.py +++ b/scripts/show_compressed_json.py @@ -2,6 +2,10 @@ import argparse import gzip import json +from funcy import lmap + +from pyinfra.server.packing import string_to_bytes + def parse_args(): parser = argparse.ArgumentParser() @@ -9,14 +13,20 @@ def parse_args(): return parser.parse_args() +def interpret(parsed): + return {**parsed, "data": str(string_to_bytes(parsed["data"]))} + + def main(fp): with open(fp, "rb") as f: compressed_json_path = f.read() json_str = gzip.decompress(compressed_json_path) - json_dict = json.loads(json_str) + parsed = json.loads(json_str) + parsed = [parsed] if isinstance(parsed, dict) else parsed + parsed = lmap(interpret, parsed) - print(json.dumps(json_dict, indent=2)) + print(json.dumps(parsed, indent=2)) if __name__ == "__main__": diff --git a/test/config.yaml b/test/config.yaml index fc5ea30..5b17f66 100644 --- a/test/config.yaml +++ b/test/config.yaml @@ -1,19 +1,56 @@ service: response_formatter: identity operations: - multi_inp_op: +# multi_inp_op: +# input: +# subdir: op_inp_files +# extension: IN.gz +# output: +# subdir: op_outp_files +# extension: OUT.gz + upper: input: - subdir: op_inp_files - extension: IN.gz + subdir: "" + extension: up_in.gz output: - subdir: op_outp_files - extension: OUT.gz + subdir: "" + extension: up_out.gz + extract: + input: + subdir: "" + extension: extr_in.gz + output: + subdir: "extractions" + extension: gz + rotate: + input: + subdir: "" + extension: rot_in.gz + output: + subdir: "" + extension: rot_out.gz + classify: + input: + subdir: "" + extension: cls_in.gz + output: + subdir: "" + extension: cls_out.gz + stream_pages: + input: + subdir: "" + extension: pgs_in.gz + output: + subdir: "pages" + extension: pgs_out.gz + + default: input: subdir: "" extension: IN.gz output: - subdir: op_outp_files + subdir: "" extension: OUT.gz storage: @@ -41,5 +78,5 @@ webserver: mock_analysis_endpoint: "http://127.0.0.1:5000" -use_docker_fixture: 1 -logging: 0 \ No newline at end of file +use_docker_fixture: 0 +logging: 1 \ No newline at end of file diff --git a/test/conftest.py b/test/conftest.py index b5aa614..d10b12c 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -89,8 +89,8 @@ def mock_make_load_data(): return load_data -# def pytest_make_parametrize_id(config, val, argname): -# return f"\n\t{argname}={val}\n" +def pytest_make_parametrize_id(config, val, argname): + return f"\n\t{argname}={val}\n" @pytest.fixture diff --git a/test/fixtures/consumer.py b/test/fixtures/consumer.py index d4e3220..7f95699 100644 --- a/test/fixtures/consumer.py +++ b/test/fixtures/consumer.py @@ -26,8 +26,8 @@ def pair_data_with_queue_message(data: Iterable[bytes]): def inner(): for i, d in enumerate(data): body = { - "dossierId": "folder", - "fileId": f"file{i}", + "dossierId": "dossier_id", + "fileId": f"file_id_{i}", "targetFileExtension": "in.gz", "responseFileExtension": "out.gz", "pages": [0, 2, 3], diff --git a/test/fixtures/input.py b/test/fixtures/input.py index 6ef1365..7fd7f68 100644 --- a/test/fixtures/input.py +++ b/test/fixtures/input.py @@ -6,6 +6,7 @@ import pytest from PIL import Image from funcy import lmap, compose, flatten, lflatten, omit, second, first, lzip, merge +from pyinfra.server.debugging import inspect from pyinfra.server.dispatcher.dispatcher import Nothing from pyinfra.server.normalization import normalize_item from pyinfra.server.packing import pack, unpack @@ -80,7 +81,7 @@ def targets(data_message_pairs, input_data_items, operation, metadata, server_si if operation is Nothing: return Nothing - op = compose(lift(star(pack)), normalize_item, operation) + op = compose(lift(star(pack)), normalize_item, inspect("A"), operation) try: response_data, response_metadata = zip(*map(unpack, flatten(starmap(op, zip(input_data_items, metadata))))) @@ -147,15 +148,19 @@ def metadata(n_items, many_to_n): @pytest.fixture def queue_message_metadata(n_items, operation_name): + print() + print("n_items:", n_items) def metadata(i): - return merge( + r = merge( { - "dossierId": "folder", - "fileId": f"file{i}", - "pages": [0, 2, 3], + "dossierId": "dossier_id", + "fileId": f"file_id_{i}", }, ({"operation": operation_name} if operation_name else {}), + ({"pages": [0, 2, 3]} if n_items > 1 else {}), ) + print("metadata:", r) + return r return lmap(metadata, range(n_items)) diff --git a/test/fixtures/server.py b/test/fixtures/server.py index ebac679..dcc428f 100644 --- a/test/fixtures/server.py +++ b/test/fixtures/server.py @@ -2,6 +2,7 @@ import io import logging import socket from collections import Counter +from itertools import count from multiprocessing import Process from operator import itemgetter from typing import Generator @@ -51,8 +52,10 @@ def server(server_stream_function, buffer_size, operation_name): @pytest.fixture -def operation_name(many_to_n): - return "multi_inp_op" if many_to_n else "" +def operation_name(core_operation): + operation_name = core_operation.__name__ + print("OPERATION NAME", operation_name) + return operation_name @pytest.fixture @@ -105,11 +108,18 @@ def core_operation(item_type, one_to_many, analysis_task): yield upper(string, metadata), metadata def upper(string: bytes, metadata): - return string.decode().upper().encode(), metadata + print("upper>>>>>>>>>>>", metadata) + print("???????????????????????????????", string) + r = string.decode().upper().encode(), metadata + print("!!!!!!!!!!!!!!!!!!!!!", r) + return r def extract(string: bytes, metadata): - for i, c in project(dict(enumerate(string.decode())), metadata["pages"]).items(): + print("extract>>>>>>>>>>>", metadata) + print(string.decode()) + for i, c in project(dict(enumerate(string.decode())), metadata.get("pages", range(len(string.decode())))).items(): metadata["id"] = i + print(i) yield c.encode(), metadata def rotate(im: bytes, metadata): diff --git a/test/integration_tests/serve_test.py b/test/integration_tests/serve_test.py index 6ce0647..5d5b63b 100644 --- a/test/integration_tests/serve_test.py +++ b/test/integration_tests/serve_test.py @@ -37,7 +37,7 @@ from test.config import CONFIG as TEST_CONFIG @pytest.mark.parametrize( "n_items", [ - 1, + # 1, 3, ], ) @@ -78,11 +78,15 @@ from test.config import CONFIG as TEST_CONFIG @pytest.mark.parametrize( "many_to_n", [ - True, - # False, + # True, + False, ], ) -def test_serving(server_process, bucket_name, components, targets, data_message_pairs, n_items, many_to_n): +def test_serving(operation_name, server_process, bucket_name, components, targets, data_message_pairs, n_items, many_to_n): + + print(f"{operation_name=}") + print() + print() storage, queue_manager, consumer, file_descriptor_manager = components @@ -108,6 +112,12 @@ def test_serving(server_process, bucket_name, components, targets, data_message_ # TODO: correctness of target should be validated as well, since production has become non-trivial assert sorted(outputs) == sorted(targets) + print(sorted(targets)) + print(sorted(outputs)) + + # sleep(2) + # print("---") + # input() @pytest.fixture @@ -146,8 +156,9 @@ def upload_data_to_folder_in_storage_and_publish_single_request_to_queue( pages = ref_message["pages"] for data, page in zip(map(first, data_message_pairs), pages): - object_descriptor = file_descriptor_manager.get_input_object_descriptor(ref_message) + object_descriptor = file_descriptor_manager.get_input_object_descriptor({**ref_message, "id": page}) object_descriptor["object_name"] = build_filepath(object_descriptor, page) + print(f"{object_descriptor=}") storage.put_object(**object_descriptor, data=compress(data)) diff --git a/test/unit_tests/queue_visitor_test.py b/test/unit_tests/queue_visitor_test.py index 4ad298c..0fd1f01 100644 --- a/test/unit_tests/queue_visitor_test.py +++ b/test/unit_tests/queue_visitor_test.py @@ -9,7 +9,7 @@ from test.utils.storage import pack_for_upload @pytest.fixture() def body(): - return {"dossierId": "folder", "fileId": "file"} + return {"dossierId": "dossier_id", "fileId": "file_id"} @pytest.mark.parametrize("client_name", ["mock", "azure", "s3"], scope="session")