refactoring; packer test; sender test

pyinfra/server/packer/packer.py

@@ -0,0 +1,8 @@
+import abc
+from typing import Iterable
+
+
+class Packer(abc.ABC):
+    @abc.abstractmethod
+    def __call__(self, data: Iterable, metadata: Iterable):
+        pass

pyinfra/server/packer/packers/identity.py

@@ -0,0 +1,14 @@
+from itertools import starmap
+from typing import Iterable
+
+from pyinfra.server.packer.packer import Packer
+
+
+def bundle(data: bytes, metadata: dict):
+    package = {"data": data, "metadata": metadata}
+    return package
+
+
+class IdentityPacker(Packer):
+    def __call__(self, data: Iterable, metadata):
+        yield from starmap(bundle, zip(data, metadata))

pyinfra/server/packer/packers/rest.py

@@ -0,0 +1,19 @@
+from itertools import starmap
+from typing import Iterable
+
+from pyinfra.server.packer.packer import Packer
+from test.utils.server import bytes_to_string
+
+
+def pack(data: bytes, metadata: dict):
+    package = {"data": bytes_to_string(data), "metadata": metadata}
+    return package
+
+
+def pack_data_and_metadata_for_rest_transfer(data: Iterable, metadata: Iterable):
+    yield from starmap(pack, zip(data, metadata))
+
+
+class RestPacker(Packer):
+    def __call__(self, data: Iterable[bytes], metadata: dict):
+        yield from pack_data_and_metadata_for_rest_transfer(data, metadata)

pyinfra/server/rest.py

@@ -1,5 +1,5 @@
 import logging
-from itertools import repeat, chain, starmap, takewhile
+from itertools import repeat, chain, takewhile, starmap
 from operator import itemgetter, methodcaller
 from typing import Iterable, Dict, List, Callable, Union, Tuple
 
@@ -7,7 +7,9 @@ import requests
 from funcy import compose, ilen, identity, flatten, rcompose, repeatedly
 
 from pyinfra.exceptions import UnexpectedItemType
-from pyinfra.utils.func import star, lift, lstarlift, parallel_map, starlift
+from pyinfra.server.packer.packers.rest import RestPacker
+from pyinfra.server.sender.senders.rest import RestSender
+from pyinfra.utils.func import star, lift, lstarlift, parallel_map
 from test.utils.server import bytes_to_string, string_to_bytes
 
 logger = logging.getLogger("PIL.PngImagePlugin")
@@ -67,14 +69,11 @@ def pipeline(url, receiver):
 def head(endpoint):
     """Builds a function that sends post or patch requests an endpoint."""
 
-    send_data_with_method_to_analyzer = starlift(sender(endpoint))
-
     def send(data: Iterable[bytes], metadata: Iterable[dict]):
         """Sends packages of data and metadata to endpoint and returns response."""
         return rcompose(
-            pack_data_and_metadata_for_rest_transfer,
-            dispatch_http_method_left_and_forward_data_right,
-            send_data_with_method_to_analyzer,
+            RestPacker(),
+            RestSender(endpoint),
             extract_payload_from_responses,
         )(data, metadata)
 

pyinfra/server/server.py

@@ -7,14 +7,6 @@ from flask import Flask, jsonify, request
 logger = logging.getLogger()
 
 
-class Nothing:
-    pass
-
-
-def has_next(peekable_iter):
-    return peekable_iter.peek(Nothing) == Nothing
-
-
 def set_up_processing_server(process_fn):
     app = Flask(__name__)
     response_payload_iter = []

test/unit_tests/rest/sender_test.py

@@ -0,0 +1,8 @@
+def test_identity_sender(data, metadata):
+    packer = IdentityPacker()
+    assert list(packer(data, metadata)) == lstarlift(bundle)(zip(data, metadata))
+
+
+def test_rest_packer(data, metadata):
+    packer = RestPacker()
+    assert list(packer(data, metadata)) == lstarlift(pack)(zip(data, metadata))