65 lines
1.9 KiB
Python
65 lines
1.9 KiB
Python
import logging
|
|
from functools import partial
|
|
from itertools import chain, repeat
|
|
from operator import methodcaller
|
|
from typing import Iterable
|
|
|
|
import pytest
|
|
import requests
|
|
from funcy import rcompose, compose, rpartial, identity, lmap, ilen, first, flatten
|
|
|
|
from pyinfra.rest import pack, unpack, bundle, inspect
|
|
from pyinfra.utils.func import lift, starlift, parallel_map, star, lstarlift
|
|
|
|
logger = logging.getLogger("PIL.PngImagePlugin")
|
|
logger.setLevel(logging.WARNING)
|
|
|
|
|
|
def dispatch_methods(input_data):
|
|
return *repeat(requests.patch, ilen(input_data) - 1), requests.post
|
|
|
|
|
|
def post_partial(url, input_data: Iterable[bytes], metadata):
|
|
def send(method, data):
|
|
response = method(url, json=data)
|
|
response.raise_for_status()
|
|
return response
|
|
|
|
pack_data_and_metadata_for_rest_transfer = lift(rpartial(pack, metadata))
|
|
dispatch_http_method_left_and_forward_data_right = parallel_map(dispatch_methods, lift(identity))
|
|
send_data_with_method_to_analyzer = starlift(send)
|
|
extract_payload_from_responses = compose(flatten, lift(methodcaller("json")))
|
|
flatten_buffered_payloads = flatten
|
|
|
|
input_data_to_result_data = rcompose(
|
|
pack_data_and_metadata_for_rest_transfer,
|
|
dispatch_http_method_left_and_forward_data_right,
|
|
send_data_with_method_to_analyzer,
|
|
extract_payload_from_responses,
|
|
flatten_buffered_payloads,
|
|
)
|
|
|
|
return input_data_to_result_data(input_data)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"item_type",
|
|
[
|
|
# "string",
|
|
# "image",
|
|
"pdf"
|
|
],
|
|
)
|
|
def test_sending_partial_request(url, input_data_items, metadata, operation, target_data_items, server_process):
|
|
|
|
print()
|
|
print("exp")
|
|
print(target_data_items)
|
|
|
|
output = list(post_partial(f"{url}/process", input_data_items, metadata))
|
|
|
|
print()
|
|
print("out")
|
|
print(output)
|
|
assert output == target_data_items
|