34 lines
1006 B
Python
34 lines
1006 B
Python
import json
|
|
from itertools import repeat, starmap, chain
|
|
|
|
import pytest
|
|
import requests
|
|
from funcy import pluck
|
|
|
|
from test.utils.server import bytes_to_string, string_to_bytes
|
|
|
|
|
|
@pytest.mark.parametrize("item_type", ["string"])
|
|
def test_sending_partial_request(url, data_items, metadata):
|
|
def pack(metadata: dict, data: bytes):
|
|
package = {"data": bytes_to_string(data), "metadata": metadata}
|
|
package = json.dumps(package).encode()
|
|
return package
|
|
|
|
def post(package):
|
|
final = str(0 if package else 1)
|
|
return requests.post(f"{url}/process", data=package, headers={"final": final})
|
|
|
|
packages = starmap(pack, zip(repeat(metadata), data_items))
|
|
|
|
responses = map(post, chain(packages, [""]))
|
|
payloads = (json.loads(r.json()) for r in responses)
|
|
data = map(string_to_bytes, chain.from_iterable(pluck("data", payloads)))
|
|
|
|
assert list(data) == [b"CONTENT"] * 7
|
|
|
|
|
|
@pytest.fixture
|
|
def metadata():
|
|
return {"idx": [1, 100, 101], "path": "asd/asd"}
|