test: organize and add tests

integration_tests/conftest.py

@@ -112,3 +112,17 @@ def test_session():
     process = start_server(domain, port, is_dev=True)
     yield
     kill_process(process)
+
+
+# create robyn.env before test and delete it after test
+@pytest.fixture
+def env_file():
+    CONTENT = """ROBYN_PORT=8081
+    ROBYN_URL=127.0.0.1"""
+    path = pathlib.Path(__file__).parent
+    env_path = path / "robyn.env"
+    env_path.write_text(CONTENT)
+    yield
+    env_path.unlink()
+    os.unsetenv("ROBYN_PORT")
+    os.unsetenv("ROBYN_URL")

integration_tests/http_methods_helpers.py

@@ -0,0 +1,68 @@
+from typing import Optional
+
+import requests
+
+BASE_URL = "http://127.0.0.1:8080"
+
+
+def check_response(response: requests.Response, expected_status_code: int):
+    assert response.status_code == expected_status_code
+    assert "server" in response.headers
+    assert response.headers["server"] == "robyn"
+
+
+def get(
+    endpoint: str, expected_status_code: int = 200, headers: dict = {}
+) -> requests.Response:
+    endpoint = endpoint.strip("/")
+    response = requests.get(f"{BASE_URL}/{endpoint}", headers=headers)
+    check_response(response, expected_status_code)
+    return response
+
+
+def post(
+    endpoint: str,
+    data: Optional[dict] = None,
+    expected_status_code: int = 200,
+    headers: dict = {},
+) -> requests.Response:
+    endpoint = endpoint.strip("/")
+    response = requests.post(f"{BASE_URL}/{endpoint}", data=data, headers=headers)
+    check_response(response, expected_status_code)
+    return response
+
+
+def put(
+    endpoint: str,
+    data: Optional[dict] = None,
+    expected_status_code: int = 200,
+    headers: dict = {},
+) -> requests.Response:
+    endpoint = endpoint.strip("/")
+    response = requests.put(f"{BASE_URL}/{endpoint}", data=data, headers=headers)
+    check_response(response, expected_status_code)
+    return response
+
+
+def patch(
+    endpoint: str,
+    data: Optional[dict] = None,
+    expected_status_code: int = 200,
+    headers: dict = {},
+) -> requests.Response:
+    endpoint = endpoint.strip("/")
+    response = requests.patch(f"{BASE_URL}/{endpoint}", data=data, headers=headers)
+    check_response(response, expected_status_code)
+    return response
+
+
+def delete(
+    endpoint: str,
+    data: Optional[dict] = None,
+    expected_status_code: int = 200,
+    headers: dict = {},
+) -> requests.Response:
+    endpoint = endpoint.strip("/")
+    response = requests.delete(f"{BASE_URL}/{endpoint}", data=data, headers=headers)
+    check_response(response, expected_status_code)
+    return response

integration_tests/test_app.py

@@ -0,0 +1,33 @@
+from robyn import Robyn
+from robyn.events import Events
+from robyn.types import Header
+
+
+def test_add_request_header():
+    app = Robyn(__file__)
+    app.add_request_header("server", "robyn")
+    assert app.request_headers == [Header(key="server", val="robyn")]
+
+
+def test_lifecycle_handlers():
+    def mock_startup_handler():
+        pass
+
+    async def mock_shutdown_handler():
+        pass
+
+    app = Robyn(__file__)
+
+    app.startup_handler(mock_startup_handler)
+    assert Events.STARTUP in app.event_handlers
+    startup = app.event_handlers[Events.STARTUP]
+    assert startup.handler == mock_startup_handler
+    assert startup.is_async is False
+    assert startup.number_of_params == 0
+
+    app.shutdown_handler(mock_shutdown_handler)
+    assert Events.SHUTDOWN in app.event_handlers
+    shutdown = app.event_handlers[Events.SHUTDOWN]
+    assert shutdown.handler == mock_shutdown_handler
+    assert shutdown.is_async is True
+    assert shutdown.number_of_params == 0

integration_tests/test_base_url.py

@@ -1,4 +1,5 @@
 import os
+
 import requests
 
 

integration_tests/test_basic_routes.py

@@ -0,0 +1,62 @@
+# Test the routes with:
+# - the GET method
+# - most common return types
+# - sync and async
+
+from typing import Optional
+
+import pytest
+
+from http_methods_helpers import get
+
+
+@pytest.mark.parametrize(
+    "route,expected_text,expected_header_key,expected_header_value",
+    [
+        ("/sync/str", "sync str get", None, None),
+        ("/sync/dict", "sync dict get", "sync", "dict"),
+        ("/sync/response", "sync response get", "sync", "response"),
+        ("/sync/str/const", "sync str const get", None, None),
+        ("/sync/dict/const", "sync dict const get", "sync_const", "dict"),
+        ("/sync/response/const", "sync response const get", "sync_const", "response"),
+        ("/async/str", "async str get", None, None),
+        ("/async/dict", "async dict get", "async", "dict"),
+        ("/async/response", "async response get", "async", "response"),
+        ("/async/str/const", "async str const get", None, None),
+        ("/async/dict/const", "async dict const get", "async_const", "dict"),
+        (
+            "/async/response/const",
+            "async response const get",
+            "async_const",
+            "response",
+        ),
+    ],
+)
+def test_basic_get(
+    route: str,
+    expected_text: str,
+    expected_header_key: Optional[str],
+    expected_header_value: Optional[str],
+    session,
+):
+    res = get(route)
+    assert res.text == expected_text
+    if expected_header_key is not None:
+        assert expected_header_key in res.headers
+        assert res.headers[expected_header_key] == expected_header_value
+
+
+@pytest.mark.parametrize(
+    "route, expected_json",
+    [
+        ("/sync/json", {"sync json get": "json"}),
+        ("/async/json", {"async json get": "json"}),
+        ("/sync/json/const", {"sync json const get": "json"}),
+        ("/async/json/const", {"async json const get": "json"}),
+    ],
+)
+def test_json_get(route: str, expected_json: dict, session):
+    res = get(route)
+    for key in expected_json.keys():
+        assert key in res.json()
+        assert res.json()[key] == expected_json[key]

integration_tests/test_delete_requests.py

@@ -1,15 +1,16 @@
-import requests
+import pytest
+from http_methods_helpers import delete
 
-BASE_URL = "http://127.0.0.1:8080"
 
+@pytest.mark.parametrize("function_type", ["sync", "async"])
+def test_delete(function_type: str, session):
+    res = delete(f"/{function_type}/dict")
+    assert res.text == f"{function_type} dict delete"
+    assert function_type in res.headers
+    assert res.headers[function_type] == "dict"
 
-def test_delete(session):
-    res = requests.delete(f"{BASE_URL}/delete")
-    assert res.status_code == 200
-    assert res.text == "DELETE Request"
 
-
-def test_delete_with_param(session):
-    res = requests.delete(f"{BASE_URL}/delete_with_body", data={"hello": "world"})
-    assert res.status_code == 200
+@pytest.mark.parametrize("function_type", ["sync", "async"])
+def test_delete_with_param(function_type: str, session):
+    res = delete(f"/{function_type}/body", data={"hello": "world"})
     assert res.text == "hello=world"

integration_tests/test_env_populator.py

@@ -1,29 +1,12 @@
-# from integration_tests.conftest import test_session
 import os
 import pathlib
 
-import pytest
-
 from robyn.env_populator import load_vars, parser
 
-path = pathlib.Path(__file__).parent
-
-
-# create robyn.env before test and delete it after test
-@pytest.fixture
-def env_file():
-    CONTENT = """ROBYN_PORT=8081
-    ROBYN_URL=127.0.0.1"""
-    env_path = path / "robyn.env"
-    env_path.write_text(CONTENT)
-    yield
-    env_path.unlink()
-    os.unsetenv("ROBYN_PORT")
-    os.unsetenv("ROBYN_URL")
-
 
 # this tests if a connection can be made to the server with the correct port imported from the env file
 def test_env_population(test_session, env_file):
+    path = pathlib.Path(__file__).parent
     env_path = path / "robyn.env"
     load_vars(variables=parser(config_path=env_path))
     PORT = os.environ["ROBYN_PORT"]

integration_tests/test_file_download.py

@@ -1,17 +1,9 @@
-import requests
+import pytest
+from http_methods_helpers import get
 
-BASE_URL = "http://127.0.0.1:8080"
 
-
-def test_file_download_sync(session):
-    r = requests.get(f"{BASE_URL}/file_download_sync")
-    assert r.status_code == 200
-    assert r.headers["Content-Disposition"] == "attachment"
-    assert r.text == "This is a test file for the downloading purpose\n"
-
-
-def test_file_download_async(session):
-    r = requests.get(f"{BASE_URL}/file_download_async")
-    assert r.status_code == 200
+@pytest.mark.parametrize("function_type", ["sync", "async"])
+def test_file_download(function_type: str, session):
+    r = get(f"/{function_type}/file/download")
     assert r.headers["Content-Disposition"] == "attachment"
     assert r.text == "This is a test file for the downloading purpose\n"

integration_tests/test_get_requests.py

@@ -1,96 +1,40 @@
-import requests
+import pytest
+from requests import Response
 
-BASE_URL = "http://127.0.0.1:8080"
+from http_methods_helpers import get
 
 
-def test_index_request(session):
-    res = requests.get(f"{BASE_URL}")
-    assert res.status_code == 200
+@pytest.mark.parametrize("function_type", ["sync", "async"])
+def test_param(function_type: str, session):
+    r = get(f"/{function_type}/param/1")
+    assert r.text == "1"
+    r = get(f"/{function_type}/param/12345")
+    assert r.text == "12345"
 
 
-def test_jsonify(session):
-    r = requests.get(f"{BASE_URL}/jsonify")
-    assert r.json() == {"hello": "world"}
-    assert r.status_code == 200
+@pytest.mark.parametrize("function_type", ["sync", "async"])
+def test_serve_html(function_type: str, session):
+    def check_response(r: Response):
+        assert r.text.startswith("<!DOCTYPE html>")
+        assert "Hello world. How are you?" in r.text
 
+    check_response(get(f"/{function_type}/serve/html"))
 
-def test_html(session):
-    r = requests.get(f"{BASE_URL}/test/123")
-    assert "Hello world. How are you?" in r.text
 
+@pytest.mark.parametrize("function_type", ["sync", "async"])
+def test_template(function_type: str, session):
+    def check_response(r: Response):
+        assert r.text.startswith("\n\n<!DOCTYPE html>")
+        assert "Jinja2" in r.text
+        assert "Robyn" in r.text
 
-def test_jinja_template(session):
-    r = requests.get(f"{BASE_URL}/template_render")
-    assert "Jinja2" in r.text
-    assert "Robyn" in r.text
+    check_response(get(f"/{function_type}/template"))
 
 
-def test_queries(session):
-    r = requests.get(f"{BASE_URL}/query?hello=robyn")
+@pytest.mark.parametrize("function_type", ["sync", "async"])
+def test_queries(function_type: str, session):
+    r = get(f"/{function_type}/queries?hello=robyn")
     assert r.json() == {"hello": "robyn"}
 
-    r = requests.get(f"{BASE_URL}/query")
+    r = get(f"/{function_type}/queries")
     assert r.json() == {}
-
-
-def test_request_headers(session):
-    r = requests.get(f"{BASE_URL}/request_headers")
-    assert r.status_code == 200
-    assert r.text == "This is a regular response"
-    assert "Header" in r.headers
-    assert r.headers["Header"] == "header_value"
-
-
-def test_const_request(session):
-    r = requests.get(f"{BASE_URL}/const_request")
-    assert "Hello world" in r.text
-    assert r.status_code == 200
-
-
-def test_const_request_json(session):
-    r = requests.get(f"{BASE_URL}/const_request_json")
-    assert r.status_code == 200
-    assert r.json() == {"hello": "world"}
-
-
-def test_const_request_headers(session):
-    r = requests.get(f"{BASE_URL}/const_request_headers")
-    assert r.status_code == 200
-    assert "Header" in r.headers
-    assert r.headers["Header"] == "header_value"
-
-
-def test_response_type(session):
-    r = requests.get(f"{BASE_URL}/types/response")
-    assert r.status_code == 200
-    assert r.text == "OK"
-
-
-def test_str_type(session):
-    r = requests.get(f"{BASE_URL}/types/str")
-    assert r.status_code == 200
-    assert r.text == "OK"
-
-
-def test_int_type(session):
-    r = requests.get(f"{BASE_URL}/types/int")
-    assert r.status_code == 200
-    assert r.text == "0"
-
-
-def test_async_response_type(session):
-    r = requests.get(f"{BASE_URL}/async/types/response")
-    assert r.status_code == 200
-    assert r.text == "OK"
-
-
-def test_async_str_type(session):
-    r = requests.get(f"{BASE_URL}/async/types/str")
-    assert r.status_code == 200
-    assert r.text == "OK"
-
-
-def test_async_int_type(session):
-    r = requests.get(f"{BASE_URL}/async/types/int")
-    assert r.status_code == 200
-    assert r.text == "0"