Refactor wsgi_utils.py

tests/wsgi_utils.py

@@ -9,37 +9,31 @@ WSGIApp = Callable[[WSGIEnv, StartResponse], Iterable[bytes]]
 def create_wsgi_env(
     path: str,
     method: str,
-    body: Union[str, bytes] = b"",
-    queries: Optional[Dict[str, str]] = None,
-    headers: Optional[Dict[str, str]] = None,
-    content_type: str = "text/plain; charset=utf-8",
+    content_type: str,
+    body: bytes,
+    queries: Dict[str, str],
+    headers: Dict[str, str],
 ) -> WSGIEnv:
-    request_method = method.upper()
-    bytes_body = body if isinstance(body, bytes) else body.encode("utf-8")
-    wsgi_input = io.BytesIO(bytes_body)
-    content_length = len(body)
-
     # 'key1=value1&key2=value2'
-    query_string = "&".join([f"{k}={v}" for k, v in queries.items()]) if queries else ""
+    query_string = "&".join([f"{k}={v}" for k, v in queries.items()])
 
     # See https://www.python.org/dev/peps/pep-3333/#environ-variables
     env = {
         "PATH_INFO": path,
-        "REQUEST_METHOD": request_method,
+        "REQUEST_METHOD": method.upper(),
         "SCRIPT_NAME": "",
         "QUERY_STRING": query_string,
         "CONTENT_TYPE": content_type,
-        "CONTENT_LENGTH": content_length,
+        "CONTENT_LENGTH": len(body),
         "SERVER_PROTOCOL": "http",
         "SERVER_NAME": "localhost",
-        "wsgi.input": wsgi_input,
+        "wsgi.input": io.BytesIO(body),
         "wsgi.version": (1, 0),
         "wsgi.errors": io.StringIO(""),
         "wsgi.multithread": True,
         "wsgi.multitprocess": True,
         "wsgi.run_once": False,
     }
-    headers = headers or {}
     for k, v in headers.items():
         env[f"HTTP_{k.upper()}"] = v
     return env
@@ -62,7 +56,10 @@ def send_request(
         status = status_
         response_headers = headers_
 
-    env = create_wsgi_env(path, method, body=body, queries=queries, headers=headers, content_type=content_type)
+    bytes_body = body if isinstance(body, bytes) else body.encode("utf-8")
+    headers = headers or {}
+    queries = queries or {}
+    env = create_wsgi_env(path, method, content_type, bytes_body, queries, headers)
     body = b""
     iterable_body = app(env, start_response)
     for b in iterable_body: