[http-query] fix: add method to parse headers and key/value data (#92)

Author: savacano28

http-query/src/helpers/__init__.py

@@ -0,0 +1,42 @@
+class HTTPHelpers:
+    @staticmethod
+    def parse_headers(headers_str):
+        if isinstance(headers_str, list):
+            return headers_str
+        headers_list = []
+        for kv in headers_str.split(","):
+            if "=" in kv:
+                k, v = kv.split("=", 1)
+                headers_list.append({"key": k.strip(), "value": v.strip()})
+        return headers_list
+
+    @staticmethod
+    def parse_parts(parts_str):
+        if isinstance(parts_str, list):
+            return parts_str
+        parts_list = []
+        for kv in parts_str.split("&"):
+            if "=" in kv:
+                k, v = kv.split("=", 1)
+                parts_list.append({"key": k.strip(), "value": v.strip()})
+        return parts_list
+
+    @staticmethod
+    def request_data_parts_body(request_data):
+        parts = HTTPHelpers.parse_parts(
+            request_data["injection"]["inject_content"]["parts"]
+        )
+        keys = list(map(lambda p: p["key"], parts))
+        values = list(map(lambda p: p["value"], parts))
+        return dict(zip(keys, values))
+
+    @staticmethod
+    def response_parsing(response):
+        success = 200 <= response.status_code < 300
+        success_status = "SUCCESS" if success else "ERROR"
+        return {
+            "url": response.url,
+            "code": response.status_code,
+            "status": success_status,
+            "message": response.text,
+        }

http-query/src/helpers/helpers.py

@@ -11,6 +11,7 @@
     HTTP_RAW_PUT_CONTRACT,
     HttpContracts,
 )
+from helpers.helpers import HTTPHelpers
 from pyoaev.helpers import OpenAEVConfigHelper, OpenAEVInjectorHelper
 
 
@@ -43,24 +44,6 @@ def __init__(self):
             self.config, open("img/icon-http.png", "rb")
         )
 
-    @staticmethod
-    def _request_data_parts_body(request_data):
-        parts = request_data["injection"]["inject_content"]["parts"]
-        keys = list(map(lambda p: p["key"], parts))
-        values = list(map(lambda p: p["value"], parts))
-        return dict(zip(keys, values))
-
-    @staticmethod
-    def _response_parsing(response):
-        success = 200 <= response.status_code < 300
-        success_status = "SUCCESS" if success else "ERROR"
-        return {
-            "url": response.url,
-            "code": response.status_code,
-            "status": success_status,
-            "message": response.text,
-        }
-
     def attachments_to_files(self, request_data):
         documents = request_data["injection"].get("inject_documents", [])
         attachments = list(filter(lambda d: d["document_attached"] is True, documents))
@@ -73,7 +56,9 @@ def attachments_to_files(self, request_data):
 
     def http_execution(self, data: Dict):
         # Build headers
-        inject_headers = data["injection"]["inject_content"].get("headers", [])
+        inject_headers = HTTPHelpers.parse_headers(
+            data["injection"]["inject_content"].get("headers", [])
+        )
         headers = {}
         for header_definition in inject_headers:
             headers[header_definition["key"]] = header_definition["value"]
@@ -93,35 +78,35 @@ def http_execution(self, data: Dict):
         # Get
         if inject_contract == HTTP_GET_CONTRACT:
             response = session.get(url=url, headers=headers)
-            return self._response_parsing(response)
+            return HTTPHelpers.response_parsing(response)
         # Post
         if inject_contract == HTTP_RAW_POST_CONTRACT:
             body = data["injection"]["inject_content"]["body"]
             response = session.post(
                 url=url, headers=headers, data=body, files=http_files
             )
-            return self._response_parsing(response)
+            return HTTPHelpers.response_parsing(response)
         # Put
         if inject_contract == HTTP_RAW_PUT_CONTRACT:
             body = data["injection"]["inject_content"]["body"]
             response = session.put(
                 url=url, headers=headers, data=body, files=http_files
             )
-            return self._response_parsing(response)
+            return HTTPHelpers.response_parsing(response)
         # Form Post
         if inject_contract == HTTP_FORM_POST_CONTRACT:
-            body = self._request_data_parts_body(data)
+            body = HTTPHelpers.request_data_parts_body(data)
             response = session.post(
                 url=url, headers=headers, data=body, files=http_files
             )
-            return self._response_parsing(response)
+            return HTTPHelpers.response_parsing(response)
         # Form Put
         if inject_contract == HTTP_FORM_PUT_CONTRACT:
-            body = self._request_data_parts_body(data)
+            body = HTTPHelpers.request_data_parts_body(data)
             response = session.put(
                 url=url, headers=headers, data=body, files=http_files
             )
-            return self._response_parsing(response)
+            return HTTPHelpers.response_parsing(response)
         # Nothing supported
         return {
             "code": 400,

http-query/src/openaev_http.py

@@ -0,0 +1,33 @@
+from unittest import TestCase
+
+from src.helpers.helpers import HTTPHelpers
+
+
+class HTTPHelpersTest(TestCase):
+    def test_parse_headers_with_string(self):
+        input_str = (
+            "Content-Type=application/x-www-form-urlencoded,Accept=application/json"
+        )
+        expected = [
+            {"key": "Content-Type", "value": "application/x-www-form-urlencoded"},
+            {"key": "Accept", "value": "application/json"},
+        ]
+        result = HTTPHelpers.parse_headers(input_str)
+        self.assertEqual(result, expected)
+
+
+def test_parse_parts_with_string(self):
+    input_str = "msg=test&user=alice"
+    expected = [{"key": "msg", "value": "test"}, {"key": "user", "value": "savacano"}]
+    result = HTTPHelpers.parse_parts(input_str)
+    self.assertEqual(result, expected)
+
+
+def test_parse_headers_empty_string(self):
+    result = HTTPHelpers.parse_headers("")
+    self.assertEqual(result, [])
+
+
+def test_parse_parts_empty_string(self):
+    result = HTTPHelpers.parse_parts("")
+    self.assertEqual(result, [])