pytestを使ってCRUDのテストをしてみた

プログラミング言語特集

#Python

下地泰寛

2020.08.31

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

pytestはpythonのテストフレームワークです。最近pytestを使う機会がありましたのでデータベースのCRUD(Create, Read, Updaet, Delete)について簡易的なテストケースを作成していきます。

環境

環境は以下の通りです。

python: 3.7.0
psycopg2: 2.0
pytest: 3.8.0
PostgreSQL: 12.4

pytestを使ってみる

pytestについての詳細はこちらを参考にしました。

pytest 使い方まとめ

pytest：フィクスチャ(fixture)の使い方

まずはじめに、assertを使ったテストを確認するため、以下のようにsimple_pytest.pyを作成します。

import pytest

def test_simple():
    result = 1
    expected = 1
    print("test_start")
    print("result:{}".format(result))
    print("expected:{}".format(expected))
    
    assert result == expected
    
    print("end_start")

printの結果を表示するために-sをつけてsimple_pytest.py実行します。すると以下のように結果とprint文が返ってきました。

$ pytest simple.py -s
================================================= test session starts ==================================================
platform darwin -- Python 3.7.0, pytest-3.8.0, py-1.6.0, pluggy-0.7.1
rootdir: /*****/simple_pytest, inifile:
plugins: remotedata-0.3.0, openfiles-0.3.0, doctestplus-0.1.3, arraydiff-0.2
collected 1 item                                                                                                       

simple.py test_start
result:1
expected:1
end_start
.

=============================================== 1 passed in 0.04 seconds ===============================================

CRUDテストの前準備

テストテーブル

idとnameの2つのカラムを持ったテストテーブルを作成します。DDLは以下になります。

CREATE TABLE public.pytest(id integer, name varchar(10));

conftestの作成

では次にデータベースへの接続設定をconftest.pyに書いていきます。conftest.pyは置かれたディレクトリ以下のすべてのテストで有効になります。ここでは接続情報として、cursorという関数を作成します。テスト完了後にはrollbackするように設定します。

import pytest
import psycopg2

DATABASE = データベース名
HOSTNAME = ホスト名
USERNAME = ユーザー名
USERPASSWORD = パスワード
PORTFORWORD = ポート番号

def get_connection():
    conn = psycopg2.connect(
        host = HOSTNAME,
        port = PORTFORWORD,
        user = USERNAME,
        password = USERPASSWORD,
        database = DATABASE)        
    return conn

@pytest.fixture
def cursor():
    with get_connection() as conn:
        with conn.cursor() as cur:
            print("\n" + "START TEST")
            yield cur
            print("\n" + "END TEST")
        conn.rollback()  # テスト完了後にロールバックする

ファイル配置

作成したconftest.pyとテストファイルcrud_test.pyを以下のように配置します。crud_test.pyテスト実行時に同じディレクトリにあるconftest.pyを確認します。

tree -L 1
.
├── __pycache__
├── conftest.py
└── crud_test.py

CRUDテスト

準備が整いましたので、データの作成（Create）、読み出し（Read）、更新（Update）、削除（Delete）のテスト内容をcrud_test.pyに書いていきます。

Create

まず始めにcreate(insert)のテストを行います。2件のデータを挿入後、テーブルのレコード数を取得します。取得したレコード数が2件になることを想定します。

def test_create(cursor):  # insertのテスト
    # 2件のデータを挿入
    insert_query = "insert into public.pytest values(1, 'test1'), (2, 'test2');"
    cursor.execute(insert_query)

    # 件数確認
    counts_query = "select count(1) from public.pytest;"
    cursor.execute(counts_query)
    result = cursor.fetchone()[0]

    expected = 2  # 2件のデータを入れたため2件の結果を想定

    print("result:{}".format(result))
    print("expected:{}".format(expected))
    assert result == expected

結果としてinsertした2件のレコード数が取得できpassしました。

$ pytest crud_test.py -s
...
crud_test.py 
START TEST
result:2
expected:2
.
END TEST

===================================================== 1 passed in 0.08 seconds ======================================================

Read

Read(select)のテストを行います。2件のデータを挿入します。そしてデータを取得します。読み取りテストの結果として取得した結果が[(1, 'test1'), (2, 'test2')]の配列になることを想定します。

def test_read(cursor):  # selectのテスト
    # 2件のデータを挿入
    insert_query = "insert into public.pytest values(1, 'test1'), (2, 'test2');"
    cursor.execute(insert_query)

    # データ取得
    select_query = "select * from public.pytest;"
    cursor.execute(select_query)
    result = cursor.fetchall()

    expected =  [(1, 'test1'), (2, 'test2')]  # insertの入力と同じデータが入っている確認

    print("result:{}".format(result))
    print("expected:{}".format(expected))
    assert result == expected

selectで取得した結果が想定と同じ配列になりpassしました。

$ pytest crud_test.py -s
...
crud_test.py 
START TEST
result:[(1, 'test1'), (2, 'test2')]
expected:[(1, 'test1'), (2, 'test2')]
.
END TEST

===================================================== 1 passed in 0.11 seconds ======================================================

Update

Updateのテストを行います。先ほどと同じく2件のデータを挿入後に、id=2のnameをtest10に更新します。取得時はid=2のnameのみに限定しtest10になっているか確認します。

def test_update(cursor):
    # 2件のデータを挿入
    insert_query = "insert into public.pytest values(1, 'test1'), (2, 'test2');"
    cursor.execute(insert_query)

    # id=2のnameをtest10に変更する
    update_query = "update public.pytest set name = 'test10' where id = 2;"
    cursor.execute(update_query)

    # データ取得
    select_query = "select name from public.pytest where id = 2;"
    cursor.execute(select_query)
    result = cursor.fetchone()[0]

    expected =  'test10' # id=2のnameがtest10になっていることを確認

    print("result:{}".format(result))
    print("expected:{}".format(expected))
    assert result == expected

id=2のnameがtest10に更新されているためpassしました。

$ pytest crud_test.py -s
...
crud_test.py 
START TEST
result:test10
expected:test10
.
END TEST

===================================================== 1 passed in 0.12 seconds ======================================================

Delete

Deleteのテストを行います。2件のデータ挿入後に全件削除するため取得結果が0であることを想定します。

def test_delete(cursor):
    # 2件のデータを挿入
    insert_query = "insert into public.pytest values(1, 'test1'), (2, 'test2');"
    cursor.execute(insert_query)

    # データの削除
    delete_query = "delete from public.pytest;"
    cursor.execute(delete_query)

    # 件数確認
    counts_query = "select count(1) from public.pytest;"
    cursor.execute(counts_query)
    result = cursor.fetchone()[0]

    expected = 0  # 全件削除しているので0件を想定結果とする

    print("result:{}".format(result))
    print("expected:{}".format(expected))
    assert result == expected

実施すると取得結果が0となりテストにpassしました。

$ pytest crud_test.py -s
...
crud_test.py 
START TEST
result:0
expected:0
.
END TEST

===================================================== 1 passed in 0.10 seconds ======================================================