FastAPI を使ってWEBアプリを作ってみる　その4

fastapi==0.63.0
uvicorn==0.13.3
pydantic==1.7.3

databases[postgresql]==0.4.1
SQLAlchemy==1.3.22
alembic==1.5.2
psycopg2==2.8.6

pytest==6.2.1 # 追加
pytest-asyncio==0.14.0 # 追加
httpx==0.16.1 # 追加
asgi-lifespan==1.0.1 # 追加
docker==4.4.1 # 追加

$ touch backend/tests/__init__.py 
$ touch backend/tests/conftest.py
$ touch backend/tests/test_hedgehogs.py backend/tests/utility.py

import time
from functools import wraps
from typing import Any, Callable, Type

import psycopg2


def do_with_retry(
    catching_exc: Type[Exception], reraised_exc: Type[Exception], error_msg: str
) -> Callable:  # pragma: no cover
    def outer_wrapper(call: Callable) -> Callable:
        @wraps(call)
        def inner_wrapper(*args: Any, **kwargs: Any) -> Any:
            delay = 0.001
            for i in range(15):
                try:
                    return call(*args, **kwargs)
                except catching_exc:
                    time.sleep(delay)
                    delay *= 2
            else:  # pragma: no cover
                raise reraised_exc(error_msg)

        return inner_wrapper

    return outer_wrapper


@do_with_retry(psycopg2.Error, RuntimeError, "Cannot start postgres server")
def ping_postgres(dsn: str) -> None:  # pragma: no cover
    conn = psycopg2.connect(dsn)
    cur = conn.cursor()
    cur.execute('select pid, state from pg_stat_activity;')
    cur.close()
    conn.close()

import os
import subprocess
import uuid
import warnings

import alembic
import docker as pydocker
import pytest
from alembic.config import Config
from app.db.repositories.hedgehogs import HedgehogsRepository
from app.models.hedgehog import HedgehogCreate, HedgehogInDB
from asgi_lifespan import LifespanManager
from databases import Database
from fastapi import FastAPI
from httpx import AsyncClient


from tests.utility import ping_postgres

config = Config("alembic.ini")


@pytest.fixture(scope="session")
def docker() -> pydocker.APIClient:
    # base url is the unix socket we use to communicate with docker
    return pydocker.APIClient(base_url="unix://var/run/docker.sock", version="auto")


@pytest.fixture(scope="session", autouse=True)
def postgres_container(docker: pydocker.APIClient) -> None:
    """
    Use docker to spin up a postgres container for the duration of the testing session.
    Kill it as soon as all tests are run.
    DB actions persist across the entirety of the testing session.
    """
    warnings.filterwarnings("ignore", category=DeprecationWarning)

    image = "postgres:12.1-alpine"
    docker.pull(image)

    # create the new container using
    # the same image used by our database
    command = """head -1 /proc/self/cgroup|cut -d/ -f3"""
    bin_own_container_id = subprocess.check_output(['sh', '-c', command])
    own_container_id = bin_own_container_id.decode().replace('\n', '')
    inspection = docker.inspect_container(own_container_id)

    network = list(inspection["NetworkSettings"]["Networks"].keys())[0]
    networking_config = docker.create_networking_config({
        network: docker.create_endpoint_config()
    })

    container_name = f"test-postgres-{uuid.uuid4()}"
    container = docker.create_container(
        image=image,
        name=container_name,
        detach=True,
        networking_config=networking_config
    )
    docker.start(container=container["Id"])

    inspection = docker.inspect_container(container["Id"])
    ip_address = inspection['NetworkSettings']['Networks'][network]['IPAddress']
    dsn = f"postgresql://postgres:postgres@{ip_address}/postgres"

    try:
        ping_postgres(dsn)
        os.environ['CONTAINER_DSN'] = dsn
        alembic.command.upgrade(config, "head")
        yield container
    finally:
        docker.kill(container["Id"])
        docker.remove_container(container["Id"])


@pytest.fixture
def app() -> FastAPI:
    from app.api.server import get_application
    return get_application()


@pytest.fixture
def db(app: FastAPI) -> Database:
    return app.state._db


@pytest.fixture
async def client(app: FastAPI) -> AsyncClient:
    async with LifespanManager(app):
        async with AsyncClient(
            app=app,
            base_url="http://testserver",
            headers={"Content-Type": "application/json"}
        ) as client:
            yield client

import logging
import os  # 追加

from app.core.config import DATABASE_URL
from databases import Database
from fastapi import FastAPI

logger = logging.getLogger(__name__)


async def connect_to_db(app: FastAPI) -> None:
    CONTAINER_DSN = os.environ.get('CONTAINER_DSN', '')  # 追加
    DB_URL = CONTAINER_DSN if CONTAINER_DSN else DATABASE_URL  # 追加
    database = Database(DB_URL, min_size=2, max_size=5)

    try:
        await database.connect()
        app.state._db = database
    except Exception as e:
        logger.warn("--- DATABASE CONNECTION ERROR ---")
        logger.warn(e)
        logger.warn("--- DATABASE CONNECTION ERROR ---")


async def close_db_connection(app: FastAPI) -> None:
    try:
        await app.state._db.disconnect()
    except Exception as e:
        logger.warn("--- DATABASEDISCONNECT ERROR ---")
        logger.warn(e)
        logger.warn("--- DATABASE DISCONNECT ERROR ---")

import logging
import os
import pathlib
import sys
from logging.config import fileConfig

import alembic
from sqlalchemy import engine_from_config, pool

sys.path.append(str(pathlib.Path(__file__).resolve().parents[3]))
from app.core.config import DATABASE_URL  # noqa

config = alembic.context.config
fileConfig(config.config_file_name)
logger = logging.getLogger("alembic.env")


def run_migrations_online() -> None:
    CONTAINER_DSN = os.environ.get('CONTAINER_DSN', '')
    DB_URL = CONTAINER_DSN if CONTAINER_DSN else DATABASE_URL
    logger.info('Run migrate on {0}'.format(str(DB_URL)))

    connectable = config.attributes.get("connection", None)
    config.set_main_option("sqlalchemy.url", str(DB_URL))

    if connectable is None:
        connectable = engine_from_config(
            config.get_section(config.config_ini_section),
            prefix="sqlalchemy.",
            poolclass=pool.NullPool,
        )

    with connectable.connect() as connection:
        alembic.context.configure(
            connection=connection,
            target_metadata=None
        )
        with alembic.context.begin_transaction():
            alembic.context.run_migrations()


run_migrations_online()

import pytest
from fastapi import FastAPI
from httpx import AsyncClient
from starlette.status import HTTP_404_NOT_FOUND, HTTP_422_UNPROCESSABLE_ENTITY


class TestHedgehogsRoutes:
    @pytest.mark.asyncio
    async def test_routes_exist(
        self,
        app: FastAPI,
        client: AsyncClient
    ) -> None:
        res = await client.post(app.url_path_for("hedgehogs:create-hedgehog"), json={})
        assert res.status_code != HTTP_404_NOT_FOUND

    @pytest.mark.asyncio
    async def test_invalid_input_raises_error(
        self,
        app: FastAPI,
        client: AsyncClient
    ) -> None:
        res = await client.post(app.url_path_for("hedgehogs:create-hedgehog"), json={})
        assert res.status_code == HTTP_422_UNPROCESSABLE_ENTITY

$ pytest -vv
======================= test session starts =======================
platform linux -- Python 3.9.1, pytest-6.2.1, py-1.10.0, pluggy-0.13.1 -- /usr/local/bin/python3
cachedir: .pytest_cache
rootdir: /backend
plugins: asyncio-0.14.0
collected 2 items

tests/test_hedgehogs.py::TestHedgehogsRoutes::test_routes_exist PASSED [ 50%]
tests/test_hedgehogs.py::TestHedgehogsRoutes::test_invalid_input_raises_error PASSED [100%]

======================= 2 passed, in 4.37s ========================

import pytest
from fastapi import FastAPI
from httpx import AsyncClient
from starlette.status import HTTP_404_NOT_FOUND, HTTP_422_UNPROCESSABLE_ENTITY

pytestmark = pytest.mark.asyncio


class TestHedgehogsRoutes:
    async def test_routes_exist(
        self,
        app: FastAPI,
        client: AsyncClient
    ) -> None:
        res = await client.post(app.url_path_for("hedgehogs:create-hedgehog"), json={})
        assert res.status_code != HTTP_404_NOT_FOUND

    async def test_invalid_input_raises_error(
        self,
        app: FastAPI,
        client: AsyncClient
    ) -> None:
        res = await client.post(app.url_path_for("hedgehogs:create-hedgehog"), json={})
        assert res.status_code == HTTP_422_UNPROCESSABLE_ENTITY

import pytest
from fastapi import FastAPI
from httpx import AsyncClient
from starlette.status import (HTTP_201_CREATED, 
                              HTTP_404_NOT_FOUND, 
                              HTTP_422_UNPROCESSABLE_ENTITY)

pytestmark = pytest.mark.asyncio


@pytest.fixture
def new_hedgehog():
    return HedgehogCreate(
        name="test hedgehog",
        description="test description",
        age=0.0,
        color_type="SOLT & PEPPER",
    )


class TestHedgehogsRoutes:
# ...省略

class TestCreateHedgehog:
    async def test_valid_input_creates_hedgehog(
        self, 
        app: FastAPI,
        client: AsyncClient,
        new_hedgehog: HedgehogCreate
    ) -> None:
        res = await client.post(
            app.url_path_for("hedgehogs:create-hedgehog"),
            json={"new_hedgehog": new_hedgehog.dict()}
        )
        assert res.status_code == HTTP_201_CREATED
        created_hedgehog = HedgehogCreate(**res.json())
        assert created_hedgehog == new_hedgehog

# ...省略
class TestCreateHedgehog:
    async def test_valid_input_creates_hedgehog(
        self,
        app: FastAPI,
        client: AsyncClient, 
        new_hedgehog: HedgehogCreate
    ) -> None:
        res = await client.post(
            app.url_path_for("hedgehogs:create-hedgehog"),
            json={"new_hedgehog": new_hedgehog.dict()}
        )
        assert res.status_code == HTTP_201_CREATED
        created_hedgehog = HedgehogCreate(**res.json())
        assert created_hedgehog == new_hedgehog

    @pytest.mark.parametrize(
        "invalid_payload, status_code",
        (
            (None, 422),
            ({}, 422),
            ({"name": "test_name"}, 422),
            ({"age": 2}, 422),
            ({"name": "test_name", "description": "test"}, 422),
        ),
    )
    async def test_invalid_input_raises_error(
        self,
        app: FastAPI,
        client: AsyncClient,
        invalid_payload: dict,
        status_code: int
    ) -> None:
        res = await client.post(
            app.url_path_for("hedgehogs:create-hedgehog"),
            json={"new_hedgehog": invalid_payload}
        )
        assert res.status_code == status_code

# ...省略
from app.models.hedgehog import HedgehogCreate, HedgehogInDB
from starlette.status import (HTTP_200_OK, HTTP_201_CREATED,
                              HTTP_404_NOT_FOUND,
                              HTTP_422_UNPROCESSABLE_ENTITY)
# ...省略

class TestGetHedgehog:
    async def test_get_hedgehog_by_id(
        self,
        app: FastAPI,
        client: AsyncClient
    ) -> None:
        res = await client.get(app.url_path_for(
            "hedgehogs:get-hedgehog-by-id",
            id=1
        ))
        assert res.status_code == HTTP_200_OK
        hedgehog = HedgehogInDB(**res.json())
        assert hedgehog.id == 1

from fastapi import APIRouter, Body, Depends, HTTPException
from starlette.status import HTTP_201_CREATED, HTTP_404_NOT_FOUND
# ...省略

@router.get("/{id}/", response_model=HedgehogPublic,
            name="hedgehogs:get-hedgehog-by-id")
async def get_hedgehog_by_id(
    id: int, hedgehogs_repo: HedgehogsRepository = Depends(
        get_repository(HedgehogsRepository
                       ))
) -> HedgehogPublic:
    hedgehog = await hedgehogs_repo.get_hedgehog_by_id(id=id)
    if not hedgehog:
        raise HTTPException(
            status_code=HTTP_404_NOT_FOUND,
            detail="指定されたidのハリネズミは見つかりませんでした"
        )
    return hedgehog

from app.db.repositories.base import BaseRepository
from app.models.hedgehog import HedgehogCreate, HedgehogInDB

CREATE_HEDGEHOG_QUERY = """
    INSERT INTO hedgehogs (name, description, age, color_type)
    VALUES (:name, :description, :age, :color_type)
    RETURNING id, name, description, age, color_type;
"""
GET_HEDGEHOG_BY_ID_QUERY = """
    SELECT id, name, description, age, color_type
    FROM hedgehogs
    WHERE id = :id;
"""


class HedgehogsRepository(BaseRepository):
    async def create_hedgehog(self, *, new_hedgehog: HedgehogCreate) -> HedgehogInDB:
        query_values = new_hedgehog.dict()
        hedgehog = await self.db.fetch_one(
            query=CREATE_HEDGEHOG_QUERY,
            values=query_values
        )
        return HedgehogInDB(**hedgehog)

    async def get_hedgehog_by_id(self, *, id: int) -> HedgehogInDB:
        hedgehog = await self.db.fetch_one(
            query=GET_HEDGEHOG_BY_ID_QUERY,
            values={"id": id}
        )
        if not hedgehog:
            return None
        return HedgehogInDB(**hedgehog)

# ...省略
from app.db.repositories.hedgehogs import HedgehogsRepository
from app.models.hedgehog import HedgehogCreate, HedgehogInDB
# ...省略

@pytest.fixture
async def test_hedgehog(db: Database) -> HedgehogInDB:
    hedgehog_repo = HedgehogsRepository(db)
    new_hedgehog = HedgehogCreate(
        name="fake hedgehog name",
        description="fake description",
        age=2.2,
        color_type="SOLT & PEPPER",
    )
    return await hedgehog_repo.create_hedgehog(new_hedgehog=new_hedgehog)

# ...省略
from app.models.hedgehog import HedgehogCreate, HedgehogInDB
# ...省略

class TestGetHedgehog:
    async def test_get_hedgehog_by_id(
        self,
        app: FastAPI,
        client: AsyncClient,
        test_hedgehog: HedgehogInDB
    ) -> None:
        res = await client.get(app.url_path_for(
            "hedgehogs:get-hedgehog-by-id",
            id=test_hedgehog.id
        ))
        assert res.status_code == HTTP_200_OK
        hedgehog = HedgehogInDB(**res.json())
        assert hedgehog == test_hedgehog

# ...省略

class TestGetHedgehog:
    async def test_get_hedgehog_by_id(
        self,
        app: FastAPI,
        client: AsyncClient,
        test_hedgehog: HedgehogInDB
    ) -> None:
        res = await client.get(app.url_path_for(
            "hedgehogs:get-hedgehog-by-id",
            id=test_hedgehog.id
        ))
        assert res.status_code == HTTP_200_OK
        hedgehog = HedgehogInDB(**res.json())
        assert hedgehog == test_hedgehog

    @pytest.mark.parametrize(
        "id, status_code",
        (
            (500, 404),
            (-1, 404),
            (None, 422),
        ),
    )
    async def test_wrong_id_returns_error(
        self, app: FastAPI, client: AsyncClient, id: int, status_code: int
    ) -> None:
        res = await client.get(app.url_path_for("hedgehogs:get-hedgehog-by-id", id=id))
        assert res.status_code == status_code

$ pytest -vv
====================== test session starts =======================
platform linux -- Python 3.9.1, pytest-6.2.1, py-1.10.0, pluggy-0.13.1 -- /usr/local/bin/python3
cachedir: .pytest_cache
rootdir: /backend
plugins: asyncio-0.14.0
collected 12 items                                               

tests/test_hedgehogs.py::TestHedgehogsRoutes::test_routes_exist PASSED [  8%]
tests/test_hedgehogs.py::TestHedgehogsRoutes::test_invalid_input_raises_error PASSED [ 16%]
tests/test_hedgehogs.py::TestCreateHedgehog::test_valid_input_creates_hedgehog PASSED [ 25%]
tests/test_hedgehogs.py::TestCreateHedgehog::test_invalid_input_raises_error[None-422] PASSED [ 33%]
tests/test_hedgehogs.py::TestCreateHedgehog::test_invalid_input_raises_error[invalid_payload1-422] PASSED [ 41%]
tests/test_hedgehogs.py::TestCreateHedgehog::test_invalid_input_raises_error[invalid_payload2-422] PASSED [ 50%]
tests/test_hedgehogs.py::TestCreateHedgehog::test_invalid_input_raises_error[invalid_payload3-422] PASSED [ 58%]
tests/test_hedgehogs.py::TestCreateHedgehog::test_invalid_input_raises_error[invalid_payload4-422] PASSED [ 66%]
tests/test_hedgehogs.py::TestGetHedgehog::test_get_hedgehog_by_id PASSED [ 75%]
tests/test_hedgehogs.py::TestGetHedgehog::test_wrong_id_returns_error[500-404] PASSED [ 83%]
tests/test_hedgehogs.py::TestGetHedgehog::test_wrong_id_returns_error[-1-404] PASSED [ 91%]
tests/test_hedgehogs.py::TestGetHedgehog::test_wrong_id_returns_error[None-422] PASSED [100%]

======================= 12 passed in 4.49s =======================