From 07a51f85ca6ce5768b0982bc6f78dbe7c9a19ea2 Mon Sep 17 00:00:00 2001
From: c-bata <contact@c-bata.link>
Date: Mon, 26 Oct 2020 15:46:19 +0900
Subject: [PATCH] Add unittests for json api

---
 tests/__init__.py   |  0
 tests/test_api.py   | 90 +++++++++++++++++++++++++++++++++++++++++++++
 tests/wsgi_utils.py | 33 +++++++++++++++++
 3 files changed, 123 insertions(+)
 create mode 100644 tests/__init__.py
 create mode 100644 tests/test_api.py
 create mode 100644 tests/wsgi_utils.py

diff --git a/tests/__init__.py b/tests/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/tests/test_api.py b/tests/test_api.py
new file mode 100644
index 00000000..726992bc
--- /dev/null
+++ b/tests/test_api.py
@@ -0,0 +1,90 @@
+import json
+from typing import Dict, Optional, Tuple, List, Any
+from unittest import TestCase
+
+import optuna
+from .wsgi_utils import create_wsgi_env
+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,
+):
+    body = json.dumps(json_body) if json_body else ""
+    return create_wsgi_env(
+        path, method, body, content_type=content_type, headers=headers
+    )
+
+
+class APITestCase(TestCase):
+    def setUp(self):
+        self.status: Optional[str] = None
+        self.headers: Optional[List[Tuple[str, str]]] = None
+
+    def _start_response(self, status, headers):
+        self.status = status
+        self.headers = headers
+
+    def test_create_study(self):
+        storage = optuna.storages.InMemoryStorage()
+        self.assertEqual(len(storage.get_all_study_summaries()), 0)
+
+        app = create_app(storage)
+        env = create_json_api_wsgi_env(
+            "/api/studies",
+            "POST",
+            json_body={
+                "study_name": "foo",
+                "direction": "minimize",
+            },
+        )
+        _ = app(env, self._start_response)
+        self.assertEqual(self.status, "201 Created")
+        self.assertEqual(len(storage.get_all_study_summaries()), 1)
+
+    def test_create_study_duplicated(self):
+        storage = optuna.storages.InMemoryStorage()
+        storage.create_new_study("foo")
+        self.assertEqual(len(storage.get_all_study_summaries()), 1)
+
+        app = create_app(storage)
+        env = create_json_api_wsgi_env(
+            "/api/studies",
+            "POST",
+            json_body={
+                "study_name": "foo",
+                "direction": "minimize",
+            },
+        )
+        _ = app(env, self._start_response)
+        self.assertEqual(self.status, "400 Bad Request")
+        self.assertEqual(len(storage.get_all_study_summaries()), 1)
+
+    def test_delete_study(self):
+        storage = optuna.storages.InMemoryStorage()
+        storage.create_new_study("foo1")
+        storage.create_new_study("foo2")
+        self.assertEqual(len(storage.get_all_study_summaries()), 2)
+
+        app = create_app(storage)
+        env = create_json_api_wsgi_env(
+            "/api/studies/1",
+            "DELETE",
+        )
+        _ = app(env, self._start_response)
+        self.assertEqual(self.status, "204 No Content")
+        self.assertEqual(len(storage.get_all_study_summaries()), 1)
+
+    def test_delete_study_not_found(self):
+        storage = optuna.storages.InMemoryStorage()
+        app = create_app(storage)
+        env = create_json_api_wsgi_env(
+            "/api/studies/1",
+            "DELETE",
+        )
+        _ = app(env, self._start_response)
+        self.assertEqual(self.status, "404 Not Found")
diff --git a/tests/wsgi_utils.py b/tests/wsgi_utils.py
new file mode 100644
index 00000000..8e738443
--- /dev/null
+++ b/tests/wsgi_utils.py
@@ -0,0 +1,33 @@
+import io
+from typing import Dict, Optional, Union
+
+
+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",
+):
+    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 ""
+    env = {
+        "HTTP_X_FORWARDED_PROTO": "http",
+        "HTTP_X_FORWARDED_HOST": "localhost",
+        "PATH_INFO": path,
+        "REQUEST_METHOD": request_method,
+        "QUERY_STRING": query_string,
+        "wsgi.input": wsgi_input,
+        "CONTENT_TYPE": content_type,
+        "CONTENT_LENGTH": content_length,
+    }
+    headers = headers or {}
+    for k, v in headers.items():
+        env[f"HTTP_{k.upper()}"] = v
+    return env