From 00ea224379e249887384468b0a738f4cac1af12d Mon Sep 17 00:00:00 2001
From: Matthias Bisping <matthias.bisping@iqser.com>
Date: Wed, 4 May 2022 16:41:29 +0200
Subject: [PATCH] refactoring; packer test; sender test

---
 pyinfra/server/packer/__init__.py         |  0
 pyinfra/server/packer/packer.py           |  8 ++++++++
 pyinfra/server/packer/packers/__init__.py |  0
 pyinfra/server/packer/packers/identity.py | 14 ++++++++++++++
 pyinfra/server/packer/packers/rest.py     | 19 +++++++++++++++++++
 pyinfra/server/rest.py                    | 13 ++++++-------
 pyinfra/server/server.py                  |  8 --------
 pyinfra/server/utils.py                   |  0
 test/unit_tests/rest/sender_test.py       |  8 ++++++++
 9 files changed, 55 insertions(+), 15 deletions(-)
 create mode 100644 pyinfra/server/packer/__init__.py
 create mode 100644 pyinfra/server/packer/packer.py
 create mode 100644 pyinfra/server/packer/packers/__init__.py
 create mode 100644 pyinfra/server/packer/packers/identity.py
 create mode 100644 pyinfra/server/packer/packers/rest.py
 create mode 100644 pyinfra/server/utils.py
 create mode 100644 test/unit_tests/rest/sender_test.py

diff --git a/pyinfra/server/packer/__init__.py b/pyinfra/server/packer/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/pyinfra/server/packer/packer.py b/pyinfra/server/packer/packer.py
new file mode 100644
index 0000000..e15a104
--- /dev/null
+++ b/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
diff --git a/pyinfra/server/packer/packers/__init__.py b/pyinfra/server/packer/packers/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/pyinfra/server/packer/packers/identity.py b/pyinfra/server/packer/packers/identity.py
new file mode 100644
index 0000000..0b3e7f3
--- /dev/null
+++ b/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))
diff --git a/pyinfra/server/packer/packers/rest.py b/pyinfra/server/packer/packers/rest.py
new file mode 100644
index 0000000..eccd38e
--- /dev/null
+++ b/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)
diff --git a/pyinfra/server/rest.py b/pyinfra/server/rest.py
index 5d7ba46..cc8a3cb 100644
--- a/pyinfra/server/rest.py
+++ b/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)
 
diff --git a/pyinfra/server/server.py b/pyinfra/server/server.py
index 694044c..eda6545 100644
--- a/pyinfra/server/server.py
+++ b/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 = []
diff --git a/pyinfra/server/utils.py b/pyinfra/server/utils.py
new file mode 100644
index 0000000..e69de29
diff --git a/test/unit_tests/rest/sender_test.py b/test/unit_tests/rest/sender_test.py
new file mode 100644
index 0000000..2fa0d3d
--- /dev/null
+++ b/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))