pytestでSQLAlchemyの接続先をオンメモリSQLite3に挿げ替える

目的

外部への依存度を減らしつつ、DB関連コードのユニットテストを書くこと。

方法

表題の通り。

コード例

"""pytestによるSQLiteの接続先の挿げ替え例."""
from typing import Iterator, List, Tuple
from unittest.mock import patch

from pytest import fixture
from sqlalchemy import create_engine, text


def get_stocks() -> List[Tuple]:
    """Product code."""
    user = "user"
    password = "password"
    host = "xxxx.xxxxxxx.us-east-2.rds.amazonaws.com"
    database = "mysql"

    engine = create_engine(
        f"mysql+pymysql://{user}:{password}@{host}/{database}"
    )

    with engine.begin() as connection:
        return list(connection.execute(text("select * from stocks;")))


@fixture(autouse=True)
def mock_db() -> Iterator[None]:
    """Fixture."""
    engine = create_engine("sqlite://")
    with engine.begin() as connection:
        connection.execute(
            text(
                "CREATE TABLE stocks "
                "(date text, trans text, symbol text, qty integer, price real)"
            )
        )
        connection.execute(
            text(
                "INSERT INTO stocks VALUES "
                "('2006-01-05','BUY','RHAT', 100, 35.14)"
            )
        )

    with patch(f"{__name__}.create_engine", return_value=engine):
        yield


def test_get_stocks() -> None:
    """Test code."""
    assert get_stocks() == [
        ("2006-01-05", "BUY", "RHAT", 100, 35.14),
    ]