Marmelator/pyinfra
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 36 Wiki Activity

blacky

Browse Source
This commit is contained in:
Julius Unverfehrt 2022-03-15 13:30:41 +01:00
parent 38fb073291
commit 48eb2bb792
5 changed files with 12 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
pyinfra/test/unit_tests/conftest.py
View File

@ -96,6 +96,7 @@ def queue_manager(queue_manager_name):
def callback(): def callback():
def inner(request): def inner(request):
return request["data"].decode() * 2 return request["data"].decode() * 2
return inner return inner

5
pyinfra/test/unit_tests/consumer_test.py
View File

@ -22,7 +22,6 @@ def access_callback():
@pytest.fixture() @pytest.fixture()
def items(): def items():
def inner(): def inner():
for i in range(3): for i in range(3):
body = { body = {
@ -53,6 +52,7 @@ class TestConsumer:
def mock_visitor(callback): def mock_visitor(callback):
def inner(data): def inner(data):
return callback({"data": data.encode()}) return callback({"data": data.encode()})
return inner return inner
callback = mock_visitor(callback) callback = mock_visitor(callback)
@ -94,9 +94,8 @@ class TestConsumer:
@pytest.mark.parametrize("queue_manager_name", ["pika"], scope="session") @pytest.mark.parametrize("queue_manager_name", ["pika"], scope="session")
def test_message_is_republished_when_callback_raises_processing_failure_exception( def test_message_is_republished_when_callback_raises_processing_failure_exception(
self, consumer, queue_manager, bucket_name, items self, consumer, queue_manager, bucket_name, items
): ):
class DebugError(Exception): class DebugError(Exception):
pass pass

4
scripts/manage_minio.py
View File

@ -29,9 +29,7 @@ def parse_args():
def add_file_compressed(storage, bucket_name, dossier_id, path) -> None: def add_file_compressed(storage, bucket_name, dossier_id, path) -> None:
path_gz = combine_dossier_id_and_file_id_and_extension( path_gz = combine_dossier_id_and_file_id_and_extension(dossier_id, Path(path).stem, ".ORIGIN.pdf.gz")
dossier_id, Path(path).stem, ".ORIGIN.pdf.gz"
)
with open(path, "rb") as f: with open(path, "rb") as f:
data = gzip.compress(f.read()) data = gzip.compress(f.read())

9
scripts/mock_client.py
View File

@ -10,7 +10,14 @@ def build_message_bodies():
for bucket_name, pdf_name in storage.get_all_object_names(CONFIG.storage.bucket): for bucket_name, pdf_name in storage.get_all_object_names(CONFIG.storage.bucket):
file_id = pdf_name.split(".")[0] file_id = pdf_name.split(".")[0]
dossier_id, file_id = file_id.split("/") dossier_id, file_id = file_id.split("/")
yield json.dumps({"dossierId": dossier_id, "fileId": file_id, "targetFileExtension": "ORIGIN.pdf.gz", "responseFileExtension": "detr.json.gz"}).encode() yield json.dumps(
{
"dossierId": dossier_id,
"fileId": file_id,
"targetFileExtension": "ORIGIN.pdf.gz",
"responseFileExtension": "detr.json.gz",
}
).encode()
if __name__ == "__main__": if __name__ == "__main__":

1
src/serve.py
View File

@ -16,7 +16,6 @@ from pyinfra.visitor import QueueVisitor, StorageStrategy
def make_callback(analysis_endpoint): def make_callback(analysis_endpoint):
def callback(message): def callback(message):
def perform_operation(operation): def perform_operation(operation):
endpoint = f"{analysis_endpoint}/{operation}" endpoint = f"{analysis_endpoint}/{operation}"
try: try: