Initial commit

tests/__init__.py

@@ -0,0 +1 @@
+"""Tests for zangramru."""

tests/conftest.py

@@ -0,0 +1,99 @@
+from collections.abc import AsyncGenerator
+from typing import Any
+
+import pytest
+from fastapi import FastAPI
+from httpx import ASGITransport, AsyncClient
+from piccolo.conf.apps import Finder
+from piccolo.engine.postgres import PostgresEngine
+from piccolo.table import create_tables, drop_tables
+
+from zangramru.settings import settings
+from zangramru.web.application import get_app
+
+
+@pytest.fixture(scope="session")
+def anyio_backend() -> str:
+    """
+    Backend for anyio pytest plugin.
+
+    :return: backend name.
+    """
+    return "asyncio"
+
+
+async def drop_database(engine: PostgresEngine) -> None:
+    """
+    Drops test database.
+
+    :param engine: engine connected to postgres database.
+    """
+    await engine.run_ddl(
+        "SELECT pg_terminate_backend(pg_stat_activity.pid) "  # noqa: S608
+        "FROM pg_stat_activity "
+        f"WHERE pg_stat_activity.datname = '{settings.db_base}' "
+        "AND pid <> pg_backend_pid();",
+    )
+    await engine.run_ddl(
+        f"DROP DATABASE {settings.db_base};",
+    )
+
+
+@pytest.fixture(autouse=True)
+async def setup_db() -> AsyncGenerator[None, None]:
+    """
+    Fixture to create all tables before test and drop them after.
+
+    :yield: nothing.
+    """
+    engine = PostgresEngine(
+        config={
+            "database": "postgres",
+            "user": settings.db_user,
+            "password": settings.db_pass,
+            "host": settings.db_host,
+            "port": settings.db_port,
+        },
+    )
+    await engine.start_connection_pool()
+
+    db_exists = await engine.run_ddl(
+        f"SELECT 1 FROM pg_database WHERE datname='{settings.db_base}'"  # noqa: S608
+    )
+    if db_exists:
+        await drop_database(engine)
+    await engine.run_ddl(f"CREATE DATABASE {settings.db_base}")
+    tables = Finder().get_table_classes()
+    create_tables(*tables, if_not_exists=True)
+
+    yield
+
+    drop_tables(*tables)
+    await drop_database(engine)
+
+
+@pytest.fixture
+def fastapi_app() -> FastAPI:
+    """
+    Fixture for creating FastAPI app.
+
+    :return: fastapi app with mocked dependencies.
+    """
+    application = get_app()
+    return application  # noqa: RET504
+
+
+@pytest.fixture
+async def client(
+    fastapi_app: FastAPI, anyio_backend: Any
+) -> AsyncGenerator[AsyncClient, None]:
+    """
+    Fixture that creates client for requesting server.
+
+    :param fastapi_app: the application.
+    :yield: client for the app.
+    """
+    async with AsyncClient(
+        transport=ASGITransport(fastapi_app), base_url="http://test", timeout=2.0
+    ) as ac:
+        yield ac

tests/test_echo.py

@@ -0,0 +1,19 @@
+import uuid
+
+from fastapi import FastAPI
+from httpx import AsyncClient
+from starlette import status
+
+
+async def test_echo(fastapi_app: FastAPI, client: AsyncClient) -> None:
+    """
+    Tests that echo route works.
+
+    :param fastapi_app: current application.
+    :param client: client for the app.
+    """
+    url = fastapi_app.url_path_for("send_echo_message")
+    message = uuid.uuid4().hex
+    response = await client.post(url, json={"message": message})
+    assert response.status_code == status.HTTP_200_OK
+    assert response.json()["message"] == message

tests/test_zangramru.py

@@ -0,0 +1,15 @@
+from fastapi import FastAPI
+from httpx import AsyncClient
+from starlette import status
+
+
+async def test_health(client: AsyncClient, fastapi_app: FastAPI) -> None:
+    """
+    Checks the health endpoint.
+
+    :param client: client for the app.
+    :param fastapi_app: current FastAPI application.
+    """
+    url = fastapi_app.url_path_for("health_check")
+    response = await client.get(url)
+    assert response.status_code == status.HTTP_200_OK