Update api discovery logic to use new Headers object

Author: bitterpanda63

aikido_zen/api_discovery/get_api_info_test.py

@@ -1,5 +1,6 @@
 import pytest
 from .get_api_info import get_api_info
+from ..helpers.headers import Headers
 
 
 class Context:
@@ -16,7 +17,8 @@ def __init__(
         self.path = path
         self.body = body
         self.xml = xml
-        self.headers = {"CONTENT_TYPE": content_type}
+        self.headers = Headers()
+        self.headers.store_header("CONTENT_TYPE", content_type)
         self.query = query
         self.cookies = {}
 
@@ -154,7 +156,7 @@ def test_auth_get_api_info(monkeypatch):
         },
         content_type="application/json",
     )
-    context1.headers["AUTHORIZATION"] = "Bearer token"
+    context1.headers.store_header("AUTHORIZATION", "Bearer token")
     api_info = get_api_info(context1)
     assert api_info == {
         "body": {

aikido_zen/api_discovery/get_auth_type_test.py

@@ -1,10 +1,13 @@
 import pytest
 from .get_auth_types import get_auth_types
+from ..helpers.headers import Headers
 
 
 class Context:
     def __init__(self, headers={}, cookies={}):
-        self.headers = headers
+        self.headers = Headers()
+        for k, v in headers.items():
+            self.headers.store_header(k, v)
         self.cookies = cookies
 
 

aikido_zen/api_discovery/get_auth_types.py

@@ -1,5 +1,6 @@
 """Usefull export : get_auth_types"""
 
+from aikido_zen.helpers.headers import Headers
 from aikido_zen.helpers.is_http_auth_scheme import is_http_auth_scheme
 
 common_api_key_header_names = [
@@ -26,13 +27,13 @@
 
 def get_auth_types(context):
     """Get the authentication type of the API request."""
-    if not isinstance(context.headers, dict):
+    if not isinstance(context.headers, Headers):
         return None
 
     result = []
 
     # Check the Authorization header
-    auth_header = context.headers.get("AUTHORIZATION")
+    auth_header = context.headers.get_header("AUTHORIZATION")
     if isinstance(auth_header, str):
         auth_header_type = get_authorization_header_type(auth_header)
         if auth_header_type:

aikido_zen/api_discovery/get_body_data_type.py

@@ -1,5 +1,7 @@
 """Exports get_body_data_type"""
 
+from aikido_zen.helpers.headers import Headers
+
 JSON_CONTENT_TYPES = [
     "application/json",
     "application/vnd.api+json",
@@ -8,12 +10,12 @@
 ]
 
 
-def get_body_data_type(headers):
+def get_body_data_type(headers: Headers):
     """Gets the type of body data from headers"""
-    if not isinstance(headers, dict) or headers is None:
+    if not isinstance(headers, Headers) or headers is None:
         return
 
-    content_type = headers.get("CONTENT_TYPE")
+    content_type = headers.get_header("CONTENT_TYPE")
     if not content_type:
         return
 

aikido_zen/api_discovery/get_body_data_type_test.py

@@ -1,23 +1,51 @@
 import pytest
 from .get_body_data_type import get_body_data_type
+from ..helpers.headers import Headers
 
 
 def test_get_body_data_type():
-    assert get_body_data_type({"CONTENT_TYPE": "application/json"}) == "json"
-    assert get_body_data_type({"CONTENT_TYPE": "application/vnd.api+json"}) == "json"
-    assert get_body_data_type({"CONTENT_TYPE": "application/csp-report"}) == "json"
-    assert get_body_data_type({"CONTENT_TYPE": "application/x-json"}) == "json"
-    assert (
-        get_body_data_type({"CONTENT_TYPE": "application/x-www-form-urlencoded"})
-        == "form-urlencoded"
-    )
-    assert get_body_data_type({"CONTENT_TYPE": "multipart/form-data"}) == "form-data"
-    assert get_body_data_type({"CONTENT_TYPE": "text/xml"}) == "xml"
-    assert get_body_data_type({"CONTENT_TYPE": "text/html"}) is None
-    assert (
-        get_body_data_type({"CONTENT_TYPE": ["application/json", "text/html"]})
-        == "json"
-    )
-    assert get_body_data_type({"x-test": "abc"}) is None
-    assert get_body_data_type(None) is None  # Testing invalid input
-    assert get_body_data_type({}) is None
+    headers = Headers()
+    headers.store_header("CONTENT_TYPE", "application/json")
+    assert get_body_data_type(headers) == "json"
+
+    headers = Headers()
+    headers.store_header("CONTENT_TYPE", "application/vnd.api+json")
+    assert get_body_data_type(headers) == "json"
+
+    headers = Headers()
+    headers.store_header("CONTENT_TYPE", "application/csp-report")
+    assert get_body_data_type(headers) == "json"
+
+    headers = Headers()
+    headers.store_header("CONTENT_TYPE", "application/x-json")
+    assert get_body_data_type(headers) == "json"
+
+    headers = Headers()
+    headers.store_header("CONTENT_TYPE", "application/x-www-form-urlencoded")
+    assert get_body_data_type(headers) == "form-urlencoded"
+
+    headers = Headers()
+    headers.store_header("CONTENT_TYPE", "multipart/form-data")
+    assert get_body_data_type(headers) == "form-data"
+
+    headers = Headers()
+    headers.store_header("CONTENT_TYPE", "text/xml")
+    assert get_body_data_type(headers) == "xml"
+
+    headers = Headers()
+    headers.store_header("CONTENT_TYPE", "text/html")
+    assert get_body_data_type(headers) is None
+
+    headers = Headers()
+    headers.store_header("CONTENT_TYPE", "application/json, text/html")
+    assert get_body_data_type(headers) == "json"
+
+    headers = Headers()
+    headers.store_header("x-test", "abc")
+    assert get_body_data_type(headers) is None
+
+    headers = Headers()
+    assert get_body_data_type(headers) is None  # Testing invalid input
+
+    headers = Headers()
+    assert get_body_data_type(headers) is None