c9bfc767a8
Merge in RR/pyinfra from partial_responses to master
Squashed commit of the following:
commit afd67d87a6349c4b97453a12274c6ccf5e976339
Author: Matthias Bisping <matthias.bisping@iqser.com>
Date: Tue Apr 26 12:48:12 2022 +0200
updated test container dockerfile for new location of tests directory
commit 37881da08ebedf0f2d0c6c2b267bdb47818a0da1
Author: Matthias Bisping <matthias.bisping@iqser.com>
Date: Tue Apr 26 12:45:12 2022 +0200
restructuring: moved test out of module scope
80 lines
1.8 KiB
Python
80 lines
1.8 KiB
Python
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")
|