Fix TestClient passing broken request objects to endpoints

ninja/testing/client.py

@@ -2,25 +2,15 @@
 from json import loads as json_loads
 from typing import Any, Callable, Dict, List, Optional, Tuple, Union
 from unittest.mock import Mock
-from urllib.parse import urljoin
 
 from django.http import QueryDict, StreamingHttpResponse
-from django.http.request import HttpHeaders, HttpRequest
+from django.http.request import HttpRequest
 
 from ninja import NinjaAPI, Router
 from ninja.responses import NinjaJSONEncoder
 from ninja.responses import Response as HttpResponse
 
 
-def build_absolute_uri(location: Optional[str] = None) -> str:
-    base = "http://testlocation/"
-
-    if location:
-        base = urljoin(base, location)
-
-    return base
-
-
 # TODO: this should be changed
 # maybe add here urlconf object and add urls from here
 class NinjaClientBase:
@@ -117,8 +107,8 @@ def urls(self) -> List:
         return self._urls_cache
 
     def _resolve(
-        self, method: str, path: str, data: Dict, request_params: Any
-    ) -> Tuple[Callable, Mock, Dict]:
+        self, method: str, path: str, data: Dict, request_params: Dict[str, Any]
+    ) -> Tuple[Callable, HttpRequest, Dict]:
         url_path = path.split("?")[0].lstrip("/")
         for url in self.urls:
             match = url.resolve(url_path)
@@ -128,74 +118,83 @@ def _resolve(
         raise Exception(f'Cannot resolve "{path}"')
 
     def _build_request(
-        self, method: str, path: str, data: Dict, request_params: Any
-    ) -> Mock:
-        request = Mock(spec=HttpRequest)
+        self, method: str, path: str, data: Dict, request_params: Dict[str, Any]
+    ) -> HttpRequest:
+        request = HttpRequest()
         request.method = method
-        request.path = path
-        request.body = ""
-        request.COOKIES = {}
+        body = request_params.pop("body", b"")
+        request._body = body.encode() if isinstance(body, str) else body
         request._dont_enforce_csrf_checks = True
-        request.is_secure.return_value = False
-        request.build_absolute_uri = build_absolute_uri
 
         request.auth = None
-        request.user = Mock()
         if "user" not in request_params:
+            request.user = Mock()
             request.user.is_authenticated = False
             request.user.is_staff = False
             request.user.is_superuser = False
 
-        request.META = request_params.pop("META", {"REMOTE_ADDR": "127.0.0.1"})
-        request.FILES = request_params.pop("FILES", {})
-
-        request.META.update({
-            f"HTTP_{k.replace('-', '_')}": v
-            for k, v in request_params.pop("headers", {}).items()
-        })
-
-        request.headers = HttpHeaders(request.META)
+        files = request_params.pop("FILES", None)
+        if files is not None:
+            request.FILES = files
 
         if isinstance(data, QueryDict):
             request.POST = data
-        else:
-            request.POST = QueryDict(mutable=True)
-
-            if isinstance(data, (str, bytes)):
-                request_params["body"] = data
-            elif data:
-                for k, v in data.items():
-                    request.POST[k] = v
+        elif isinstance(data, (str, bytes)):
+            request._body = data.encode() if isinstance(data, str) else data
+        elif data:
+            for k, v in data.items():
+                request.POST[k] = v
 
+        query_params = request_params.pop("query_params", None)
         if "?" in path:
-            request.GET = QueryDict(path.split("?")[1])
-        else:
-            query_params = request_params.pop("query_params", None)
-            if query_params:
-                query_dict = QueryDict(mutable=True)
-                for k, v in query_params.items():
-                    if isinstance(v, list):
-                        for item in v:
-                            query_dict.appendlist(k, item)
-                    else:
-                        query_dict[k] = v
-                request.GET = query_dict
-            else:
-                request.GET = QueryDict()
+            path, query_string = path.split("?", maxsplit=1)
+            request.GET = QueryDict(query_string)
+        elif query_params is not None:
+            for k, v in query_params.items():
+                if isinstance(v, list):
+                    for item in v:
+                        request.GET.appendlist(k, item)
+                else:
+                    request.GET[k] = v
+        request.path = path
+        # If "settings.FORCE_SCRIPT_NAME" is set, "request.path_info" ought
+        # to respect it, but this class skips the Django URL resolver,
+        # so don't bother
+        request.path_info = path
+
+        request.META = request_params.pop(
+            "META",
+            {
+                "REQUEST_METHOD": request.method,
+                "SCRIPT_NAME": "",
+                "PATH_INFO": request.path_info,
+                "QUERY_STRING": request.GET.urlencode(),
+                "SERVER_NAME": "testserver",
+                "SERVER_PORT": "80",
+                "SERVER_PROTOCOL": "HTTP/1.1",
+                "REMOTE_ADDR": "127.0.0.1",
+            },
+        )
+        request.META.update({
+            f"HTTP_{k.replace('-', '_')}": v
+            for k, v in request_params.pop("headers", {}).items()
+        })
 
         for k, v in request_params.items():
             setattr(request, k, v)
         return request
 
 
 class TestClient(NinjaClientBase):
-    def _call(self, func: Callable, request: Mock, kwargs: Dict) -> "NinjaResponse":
+    def _call(
+        self, func: Callable, request: HttpRequest, kwargs: Dict
+    ) -> "NinjaResponse":
         return NinjaResponse(func(request, **kwargs))
 
 
 class TestAsyncClient(NinjaClientBase):
     async def _call(
-        self, func: Callable, request: Mock, kwargs: Dict
+        self, func: Callable, request: HttpRequest, kwargs: Dict
     ) -> "NinjaResponse":
         return NinjaResponse(await func(request, **kwargs))
 

tests/test_pagination.py

@@ -507,8 +507,8 @@ def test_case9():
     response = client.get("/items_9?skip=5").json()
     assert response == {
         "items": [5, 6, 7, 8, 9],
-        "next": "http://testlocation/?skip=10",
-        "prev": "http://testlocation/?skip=0",
+        "next": "http://testserver/items_9?skip=10",
+        "prev": "http://testserver/items_9?skip=0",
     }
 
 

tests/test_parser.py

@@ -12,7 +12,7 @@ class MyParser(Parser):
 
     def parse_body(self, request: HttpRequest):
         "just splitting body to lines"
-        return request.body.encode().splitlines()
+        return request.body.splitlines()
 
     def parse_querydict(
         self, data: QueryDict, list_fields: List[str], request: HttpRequest

tests/test_test_client.py

@@ -19,7 +19,7 @@ def request_build_absolute_uri(request):
 
 @router.get("/request/build_absolute_uri/location")
 def request_build_absolute_uri_location(request):
-    return request.build_absolute_uri("location")
+    return request.build_absolute_uri("/different-location")
 
 
 @router.get("/test")
@@ -37,17 +37,36 @@ def get_cookies(request):
     return dict(request.COOKIES)
 
 
+@router.get("/test-host")
+def get_host(request):
+    return {
+        "host": request.get_host(),
+    }
+
+
+@router.get("/test-path")
+def get_path(request):
+    return {
+        "path": request.path,
+        "full_path": request.get_full_path(),
+    }
+
+
 client = TestClient(router)
 
 
 @pytest.mark.parametrize(
     "path,expected_status,expected_response",
     [
-        ("/request/build_absolute_uri", HTTPStatus.OK, "http://testlocation/"),
+        (
+            "/request/build_absolute_uri",
+            HTTPStatus.OK,
+            "http://testserver/request/build_absolute_uri",
+        ),
         (
             "/request/build_absolute_uri/location",
             HTTPStatus.OK,
-            "http://testlocation/location",
+            "http://testserver/different-location",
         ),
     ],
 )
@@ -124,3 +143,20 @@ def test_headered_client_request_with_default_cookies():
 def test_headered_client_request_with_overwritten_and_additional_cookies():
     r = cookied_client.get("/test-cookies", COOKIES={"A": "na", "C": "nc"})
     assert r.json() == {"A": "na", "B": "b", "C": "nc"}
+
+
+def test_client_host():
+    r = client.get("/test-host")
+    assert r.json() == {
+        "host": "testserver",
+    }
+
+
+def test_client_path_query_params():
+    r = client.get("/test-path", query_params={"foo": "bar"})
+    assert r.json() == {"path": "/test-path", "full_path": "/test-path?foo=bar"}
+
+
+def test_client_path_query_string():
+    r = client.get("/test-path?foo=bar")
+    assert r.json() == {"path": "/test-path", "full_path": "/test-path?foo=bar"}