Help us understand the problem. What is going on with this article?

PythonのFlaskでMySQLを利用したRESTfulなAPIにpytestで単体テストを追加する

More than 1 year has passed since last update.

概要

前に実装した環境でpytest を利用して単体テストができるようにしてみました。

PythonのFlaskでMySQLを利用したRESTfulなAPIをDocker環境で実装する
https://qiita.com/kai_kou/items/5d73de21818d1d582f00

ソースはこちら。

kai-kou/flask-mysql-restful-api-on-docker
https://github.com/kai-kou/flask-mysql-restful-api-on-docker

手順

上記ソースに対して、単体テストが実行できるように変更した箇所を抜粋します。

最終形は上記リポジトリのfeature/add_test ブランチに置いてます。
https://github.com/kai-kou/flask-mysql-restful-api-on-docker/tree/feature/add_test

単体テスト用のデータベースが初期化時に作成されるようにします。

mysql/sqls/initialize.sql
CREATE DATABASE hoge;
CREATE DATABASE test_hoge;
use hoge;

単体テスト用のデータベースが参照できるようにTestingConfig クラスを追加しています。

src/config.py
import os


class DevelopmentConfig:

  # SQLAlchemy
  SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://{user}:{password}@{host}/{database}?charset=utf8'.format(
    **{
      'user': os.getenv('DB_USER', 'root'),
      'password': os.getenv('DB_PASSWORD', 'hoge'),
      'host': os.getenv('DB_HOST', 'db'),
      'database': os.getenv('DB_DATABASE', 'hoge'),
    })
  SQLALCHEMY_TRACK_MODIFICATIONS = False
  SQLALCHEMY_ECHO = False

class TestingConfig:

  # SQLAlchemy
  SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://{user}:{password}@{host}/{database}?charset=utf8'.format(
    **{
      'user': os.getenv('DB_USER', 'root'),
      'password': os.getenv('DB_PASSWORD', 'hoge'),
      'host': os.getenv('DB_HOST', 'db'),
      'database': os.getenv('DB_DATABASE', 'test_hoge'),
    })
  SQLALCHEMY_TRACK_MODIFICATIONS = False
  SQLALCHEMY_ECHO = False


Config = DevelopmentConfig

単体テスト用にpytestflask_testing を追加しています。

flask_testing についてはあまり情報がなく、公式を参考にしました。

Flask-Testing Flask-Testing 0.3 documentation
https://flask-testing.readthedocs.io/en/latest/

requirements.txt
flask
sqlalchemy
flask-restful
flask-sqlalchemy
flask-migrate
pymysql
gunicorn
flask_marshmallow
marshmallow-sqlalchemy
pytest
flask_testing

単体テスト時にアプリやテーブルの初期化などを行うベースとなるクラスを定義します。

こちらは下記を参考にさせてもらいました。

Microservices with Docker, Flask, and React - Test Setup
https://testdriven.io/part-one-test-setup

create_app でテスト用の設定を読み込み、setUptearDown でテストごとにテーブルの追加やテーブルの削除が行なわれるようにしています。

src/tests/base.py
from flask_testing import TestCase

from src.app import app

from src.database import db, init_db


class BaseTestCase(TestCase):
  def create_app(self):
    app.config.from_object('src.config.TestingConfig')
    return app

  def setUp(self):
    self.app = self.app.test_client()
    db.create_all()
    db.session.commit()

  def tearDown(self):
    db.session.remove()
    db.drop_all()

実際のテストです。上記で定義したBaseTestCase クラスを継承しています。
実装しているリソースの各メソッドにアクセスできるかチェックしてるだけです。

flask_testing を利用すると、self.assert_200(response) などと書けて良いのですが、すべてのステータスコード分ないので、ちょっと微妙です。せめて201 くらい。。。

src/tests/test_hoge.py
from .base import BaseTestCase

import json

from src.app import app


class TestHogeListAPI(BaseTestCase):

  def test_get_hoges_no_data(self):
    response = self.app.get('/hoges')
    self.assert_200(response)
    assert(
      json.loads(response.get_data()) == {'items': []}
    )

  def test_delete_hoges_200(self):
    postPrms = {
      'name': 'hoge',
      'state': 'hoge'
    }
    response = self.app.post('/hoges',
        data=json.dumps(postPrms),
        content_type='application/json'
    )
    self.assert_status(response, 201)

    data = json.loads(response.get_data())
    id = data['id']
    response = self.app.get(f'/hoges/{id}')
    self.assert_status(response, 200)

  def test_get_hoges_one_data(self):
    postPrms = {
      'name': 'hoge',
      'state': 'hoge'
    }
    response = self.app.post('/hoges',
        data=json.dumps(postPrms),
        content_type='application/json'
    )
    self.assert_status(response, 201)

    response = self.app.get('/hoges')
    self.assert_200(response)
    data = json.loads(response.get_data())
    assert(len(data['items']) == 1)


class TestHogeAPI(BaseTestCase):

  def test_get_hoges_404(self):
    response = self.app.get('/hoges/xxx')
    self.assert_404(response)

  def test_post_hoges_201(self):
    prms = {
      'name': 'hoge',
      'state': 'hoge'
    }
    response = self.app.post('/hoges',
        data=json.dumps(prms),
        content_type='application/json'
    )
    self.assert_status(response, 201)

  def test_put_hoges_204(self):
    postPrms = {
      'name': 'hoge',
      'state': 'hoge'
    }
    response = self.app.post('/hoges',
        data=json.dumps(postPrms),
        content_type='application/json'
    )
    self.assert_status(response, 201)

    data = json.loads(response.get_data())
    id = data['id']
    putPrms = {
      'name': 'hoge2',
      'state': 'hoge2'
    }
    response = self.app.put(f'/hoges/{id}',
        data=json.dumps(putPrms),
        content_type='application/json'
    )
    self.assert_status(response, 204)

  def test_delete_hoges_204(self):
    postPrms = {
      'name': 'hoge',
      'state': 'hoge'
    }
    response = self.app.post('/hoges',
        data=json.dumps(postPrms),
        content_type='application/json'
    )
    self.assert_status(response, 201)

    data = json.loads(response.get_data())
    id = data['id']
    response = self.app.delete(f'/hoges/{id}')
    self.assert_status(response, 204)

  def test_get_hoges_200(self):
    postPrms = {
      'name': 'hoge',
      'state': 'hoge'
    }
    response = self.app.post('/hoges',
        data=json.dumps(postPrms),
        content_type='application/json'
    )
    self.assert_status(response, 201)

    data = json.loads(response.get_data())
    id = data['id']
    response = self.app.get(f'/hoges/{id}')
    self.assert_status(response, 200)

単体テストを実行してみます。コンテナ内・外どちらで実行してもおkです。

> docker-compose exec api pytest
============================= test session starts =============================
platform linux -- Python 3.6.6, pytest-3.8.2, py-1.7.0, pluggy-0.7.1
rootdir: /src, inifile:
collected 8 items

tests/test_hoge.py ........                                             [100%]

========================== 8 passed in 5.44 seconds ===========================

はい。
既存のコードに手を触れず、単体テストを追加することができました。
一度構成がまとまるとあとはスムーズに開発が進められそうです^^

参考

PythonのFlaskでMySQLを利用したRESTfulなAPIをDocker環境で実装する
https://qiita.com/kai_kou/items/5d73de21818d1d582f00

Flask-Testing Flask-Testing 0.3 documentation
https://flask-testing.readthedocs.io/en/latest/

Microservices with Docker, Flask, and React - Test Setup
https://testdriven.io/part-one-test-setup

kai_kou
2004年からWeb系のシステムエンジニアとして開発、運用、マネジメントを経験。現在はアイレット(クラウドパック)に所属。 べ、別にいいね貰えたからってモチベーションが上がって記事とコードの品質があがるわけじゃないんだからね///
https://twitter.com/k_aik_ou
cloudpack
Amazon Web Services (AWS) の導入設計、環境構築、運用・保守をサポートするマネジドホスティングサービス
https://cloudpack.jp/
Why not register and get more from Qiita?
  1. We will deliver articles that match you
    By following users and tags, you can catch up information on technical fields that you are interested in as a whole
  2. you can read useful information later efficiently
    By "stocking" the articles you like, you can search right away
Comments
No comments
Sign up for free and join this conversation.
If you already have a Qiita account
Why do not you register as a user and use Qiita more conveniently?
You need to log in to use this function. Qiita can be used more conveniently after logging in.
You seem to be reading articles frequently this month. Qiita can be used more conveniently after logging in.
  1. We will deliver articles that match you
    By following users and tags, you can catch up information on technical fields that you are interested in as a whole
  2. you can read useful information later efficiently
    By "stocking" the articles you like, you can search right away
ユーザーは見つかりませんでした