46 lines
1.3 KiB
Python
46 lines
1.3 KiB
Python
import json
|
|
from itertools import chain
|
|
from operator import methodcaller, itemgetter
|
|
|
|
import pytest
|
|
import requests
|
|
from funcy import curry, rcompose, compose
|
|
|
|
from test.utils.server import bytes_to_string, string_to_bytes
|
|
|
|
|
|
def lift(fn):
|
|
return curry(map)(fn)
|
|
|
|
|
|
@pytest.mark.parametrize("item_type", ["string"])
|
|
def test_sending_partial_request(url, data_items, metadata):
|
|
def pack(data: bytes):
|
|
package = {"data": bytes_to_string(data), "metadata": metadata}
|
|
package = json.dumps(package).encode()
|
|
return package
|
|
|
|
def post(data):
|
|
return requests.post(f"{url}/process", data=data)
|
|
|
|
pack_data_and_metadata_for_rest_transfer = lift(pack)
|
|
send_packages_to_analyzer_and_receive_responses = lift(post)
|
|
extract_data_from_responses = lift(compose(itemgetter("data"), methodcaller("json")))
|
|
flatten_buffered_payloads = chain.from_iterable
|
|
interpret_payloads = lift(string_to_bytes)
|
|
|
|
data = rcompose(
|
|
pack_data_and_metadata_for_rest_transfer,
|
|
send_packages_to_analyzer_and_receive_responses,
|
|
extract_data_from_responses,
|
|
flatten_buffered_payloads,
|
|
interpret_payloads
|
|
)((*data_items, b""))
|
|
|
|
assert list(data) == [b"CONTENT"] * 7
|
|
|
|
|
|
@pytest.fixture
|
|
def metadata():
|
|
return {"idx": [1, 100, 101], "path": "asd/asd"}
|