Refactor test_api.py

tests/test_api.py

@@ -1,23 +1,9 @@
 import json
-from typing import Dict, Optional, Any
 from unittest import TestCase
 
 import optuna
-from .wsgi_utils import create_wsgi_env, send_request, WSGIEnv
 from optuna_dashboard.app import create_app
-
-
-def create_json_api_wsgi_env(
-    path: str,
-    method: str,
-    json_body: Optional[Dict[str, Any]] = None,
-    content_type: str = "application/json",
-    headers: Optional[Dict[str, str]] = None,
-) -> WSGIEnv:
-    body = json.dumps(json_body) if json_body else ""
-    return create_wsgi_env(
-        path, method, body, content_type=content_type, headers=headers
-    )
+from .wsgi_utils import send_request
 
 
 class APITestCase(TestCase):
@@ -27,11 +13,12 @@ class APITestCase(TestCase):
         storage.create_new_study("foo2")
 
         app = create_app(storage)
-        env = create_json_api_wsgi_env(
+        status, _, body = send_request(
+            app,
             "/api/studies/",
             "GET",
+            content_type="application/json",
         )
-        status, _, body = send_request(app, env)
         self.assertEqual(status, "200 OK")
         study_summaries = json.loads(body)["study_summaries"]
         self.assertEqual(len(study_summaries), 2)
@@ -41,15 +28,17 @@ class APITestCase(TestCase):
         self.assertEqual(len(storage.get_all_study_summaries()), 0)
 
         app = create_app(storage)
-        env = create_json_api_wsgi_env(
+        request_body = {
+            "study_name": "foo",
+            "direction": "minimize",
+        }
+        status, _, _ = send_request(
+            app,
             "/api/studies",
             "POST",
-            json_body={
-                "study_name": "foo",
-                "direction": "minimize",
-            },
+            content_type="application/json",
+            body=json.dumps(request_body),
         )
-        status, _, _ = send_request(app, env)
         self.assertEqual(status, "201 Created")
         self.assertEqual(len(storage.get_all_study_summaries()), 1)
 
@@ -59,15 +48,17 @@ class APITestCase(TestCase):
         self.assertEqual(len(storage.get_all_study_summaries()), 1)
 
         app = create_app(storage)
-        env = create_json_api_wsgi_env(
+        request_body = {
+            "study_name": "foo",
+            "direction": "minimize",
+        }
+        status, _, _ = send_request(
+            app,
             "/api/studies",
             "POST",
-            json_body={
-                "study_name": "foo",
-                "direction": "minimize",
-            },
+            content_type="application/json",
+            body=json.dumps(request_body),
         )
-        status, _, _ = send_request(app, env)
         self.assertEqual(status, "400 Bad Request")
         self.assertEqual(len(storage.get_all_study_summaries()), 1)
 
@@ -78,20 +69,22 @@ class APITestCase(TestCase):
         self.assertEqual(len(storage.get_all_study_summaries()), 2)
 
         app = create_app(storage)
-        env = create_json_api_wsgi_env(
+        status, _, _ = send_request(
+            app,
             "/api/studies/1",
             "DELETE",
+            content_type="application/json",
         )
-        status, _, _ = send_request(app, env)
         self.assertEqual(status, "204 No Content")
         self.assertEqual(len(storage.get_all_study_summaries()), 1)
 
     def test_delete_study_not_found(self) -> None:
         storage = optuna.storages.InMemoryStorage()
         app = create_app(storage)
-        env = create_json_api_wsgi_env(
+        status, _, _ = send_request(
+            app,
             "/api/studies/1",
             "DELETE",
+            content_type="application/json",
         )
-        status, _, _ = send_request(app, env)
         self.assertEqual(status, "404 Not Found")

tests/wsgi_utils.py

@@ -46,19 +46,26 @@ def create_wsgi_env(
 
 
 def send_request(
-    app: WSGIApp, env: WSGIEnv
+    app: WSGIApp,
+    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",
 ) -> Tuple[str, List[Tuple[str, str]], bytes]:
     status: str = ""
-    headers: List[Tuple[str, str]] = []
+    response_headers: List[Tuple[str, str]] = []
 
     def start_response(status_: str, headers_: List[Tuple[str, str]]) -> None:
-        nonlocal status, headers
+        nonlocal status, response_headers
         status = status_
-        headers = headers_
+        response_headers = headers_
 
+    env = create_wsgi_env(path, method, body=body, queries=queries, headers=headers, content_type=content_type)
     body = b""
     iterable_body = app(env, start_response)
     for b in iterable_body:
         body += b
 
-    return status, headers, body
+    return status, response_headers, body