added test for sending pdf plus metadata in the same request as one json dict

pyinfra/exceptions.py

@@ -28,3 +28,7 @@ class ConsumerError(Exception):
 
 class NoSuchContainer(KeyError):
     pass
+
+
+class IntentionalTestException(RuntimeError):
+    pass

pyinfra/test/exploration_tests/request_test.py

@@ -0,0 +1,79 @@
+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")

requirements.txt

@@ -2,7 +2,7 @@ pika==1.2.0
 retry==0.9.2
 envyaml==1.10.211231
 minio==7.1.3
-Flask==2.0.3
+Flask==2.1.1
 waitress==2.0.0
 azure-core==1.22.1
 azure-storage-blob==12.9.0
@@ -12,3 +12,4 @@ docker-compose==1.29.2
 tqdm==4.62.3
 pytest~=7.0.1
 funcy==1.17
+fpdf==1.7.2