Tests with asgi/wsgi

aikido_zen/sources/django/run_init_stage.py

@@ -38,10 +38,14 @@ def run_init_stage(request):
     # In a separate try-catch we set the context :
     try:
         context = None
-        if hasattr(request, "scope"):  # This request is an ASGI request
+        if (
+            hasattr(request, "scope") and request.scope is not None
+        ):  # This request is an ASGI request
             context = Context(req=request.scope, body=body, source="django_async")
-        elif hasattr(request, "META"):  # WSGI request
+        elif hasattr(request, "META") and request.META is not None:  # WSGI request
             context = Context(req=request.META, body=body, source="django")
+        else:
+            return
         context.set_as_current_context()
 
         # Init stage needs to be run with context already set :

aikido_zen/sources/django/run_init_stage_test.py

@@ -3,6 +3,30 @@
 from .run_init_stage import run_init_stage
 from ...context import Context, get_current_context, current_context
 
+wsgi_request = {
+    "REQUEST_METHOD": "GET",
+    "HTTP_HEADER_1": "header 1 value",
+    "HTTP_HEADER_2": "Header 2 value",
+    "RANDOM_VALUE": "Random value",
+    "HTTP_COOKIE": "sessionId=abc123xyz456;",
+    "wsgi.url_scheme": "http",
+    "HTTP_HOST": "localhost:8080",
+    "PATH_INFO": "/hello",
+    "QUERY_STRING": "user=JohnDoe&age=30&age=35",
+    "CONTENT_TYPE": "application/json",
+    "REMOTE_ADDR": "198.51.100.23",
+}
+asgi_scope = {
+    "method": "PUT",
+    "headers": [(b"COOKIE", b"a=b; c=d"), (b"header1_test-2", b"testValue2198&")],
+    "query_string": b"a=b&b=d",
+    "client": ["1.1.1.1"],
+    "server": ["192.168.0.1", 443],
+    "scheme": "https",
+    "root_path": "192.168.0.1",
+    "path": "192.168.0.1/a/b/c/d",
+}
+
 
 @pytest.fixture
 def mock_request():
@@ -11,16 +35,19 @@ def mock_request():
     request.POST.dict.return_value = {}
     request.content_type = "application/json"
     request.body = '{"key": "value"}'  # Example JSON body
-    request.META = {}
+    request.META = wsgi_request
+    request.scope = None
     return request
 
+
 @pytest.fixture(autouse=True)
 def run_around_tests():
     yield
     # Make sure to reset context after every test so it does not
     # interfere with other tests
     current_context.set(None)
 
+
 def test_run_init_stage_with_json(mock_request):
     """Test run_init_stage with a JSON request."""
     run_init_stage(mock_request)
@@ -101,3 +128,18 @@ def test_run_init_stage_with_xml(mock_request):
     # Assertions
     context: Context = get_current_context()
     assert context.body == "<root><key>value</key></root>"
+
+
+def test_uses_wsgi(mock_request):
+    run_init_stage(mock_request)
+    # Assertions
+    context: Context = get_current_context()
+    assert "/hello" == context.route
+
+
+def test_uses_asgi_prio(mock_request):
+    mock_request.scope = asgi_scope
+    run_init_stage(mock_request)
+    # Assertions
+    context: Context = get_current_context()
+    assert "/a/b/c/d" == context.route