バッチを自動テストする

色々なテストコードを作成してみようの講座の3レッスン目。

今回はバッチなどで使うコマンドの自動テストを学んでいきます。

前回のレッスンはこちら↓

ウサギ

前回はお前の日記アプリで、クラスや関数のテストを学んだな

わいへい

結構簡単だったね

ウサギ

実はサーバー側のテスト方法はほとんど理解できたと言っていい

ウサギ

あとはもう一つ。

よくあるテストパターンとして、
バッチの自動テストを学んでいくぞ

テスト対象の準備
機能追加する
テストコードを書く
バッチを自動テストする方法まとめ
次の講座
Djangoの基礎を学びたい方は

テスト対象の準備

わいへい

バッチ？バッチってなんすか？

ウサギ

(こいつ本当にWeb系の開発やったことあんのか？)

わいへい

心の声が口に出ちゃってるんですが。。。

バッチ処理は一定量集計したデータをまとめて処理する、的なものの総称だ。

例えばECシステムを考えてみよう。

日中に上がってきた注文データを、深夜の特定時刻にまとめて決済処理する〜とかな。

わいへい

なるほど、ためたデータに対して一気に何かやる〜的なやつか

そんじゃあバッチ機能を作ってみるぞ。

お前の日記アプリはこんな感じだったな。

adminの管理画面から記事を投稿すると、トップページに昇順で表示されるってわけだ

今回は、下記のバッチ機能を作ってみよう。

特定のコマンドを叩くと
- 現在時刻から1週間以内の記事に「new」タグをつける
- 現在時刻から1週間経過した記事から「new」タグを外す

リポジトリのクローン

ウサギ

まずはベースとなる日記アプリのリポジトリをクローンする

GitHub - yheihei/base_diary: ウサでも分かる中級Django講座の教材

ウサでも分かる中級Django講座の教材. Contribute to yheihei/base_diary development by creating an account on GitHub.

$ git clone https://github.com/yheihei/base_diary.git
$ cd base_diary

テストコードだけ見る場合

ウサギ

今回もテストコードを書いたブランチを予め用意しておいた。

答えだけ先に見たいやつは下記コマンドでブランチをチェックアウトしてくれ

GitHub - yheihei/base_diary at feature/#4

ウサでも分かる中級Django講座の教材. Contribute to yheihei/base_diary development by creating an account on GitHub.

# テストコードが書いてあるブランチを見たい場合はこちら
$ git fetch && git checkout feature/#4

今回実装する機能

今回つくる機能をもう一度おさらいしておく。

・特定のコマンドを叩くと
　・現在時刻から1週間以内の記事に「new」タグをつける
　・現在時刻から1週間経過した記事から「new」タグを外す

例えば

今7/14で
下記のように2記事入っているとする

各日記にはタグを表示するところがある。

1週間経過してない7/13の日記にはまだ新着タグがついていない。

一方、7/4の日記は、1週間経過してるのに、新着タグnewがついたままだ。

この状態で下記バッチを叩くと

$ python manage.py update_new_tag
新着タグ付与バッチが正常終了しました

7/13の日記にはnewタグがつき、

7/4の日記からはnewタグが外れたな。

この機能を作って、テストしていくぞ。

機能追加する

さっそくタグ機能を作っていく。

diary/models.pyに下記を追加する。

class Post(models.Model):
...
...
...
...
  tags = models.ManyToManyField(
    'Tag',
    blank=True,
    related_name="posts",
  )

class Tag(models.Model):
  name = models.CharField(max_length=1024)
  slug = models.CharField(max_length=1024, unique=True)
  created_at = models.DateTimeField(auto_now_add=True)
  updated_at = models.DateTimeField(auto_now=True)

  def __str__(self) -> str:
    return f'{self.name}'

次にテンプレート側でタグを表示させるようにする。

<html>
  <head>
    <title>とあるエンジニアの日記帳</title>
  </head>
  <body>
    {% load static %}
    <h1>とあるエンジニアの日記帳</h1>
    {% for post in posts %}
      <div>
        <h2>{{ post.title }}</h2>
        <div>投稿 : {{ post.user }}</div>
        <div>{{ post.body |safe }}</div>
        <div>作成時刻 : {{ post.created_at }}</div>
        <div>更新時刻 : {{ post.updated_at }}</div>
        <div>カテゴリー : 
        {% for category in post.categories.all %}
          <span>{{ category }},</span>
        {% endfor %}
        <div>タグ : 
        {% for tag in post.tags.all %}
          <span>{{ tag }},</span>
        {% endfor %}
        </div>
      </div>
    {% empty %}
      <p>No Diaries.</p>
    {% endfor %}
  </body>
</html>

これでタグがトップに表示されるようになった。

最後にバッチ機能を作っていく。

diary/management/commands/update_new_tag.py に下記を記述する。

from django.core.management.base import BaseCommand, CommandError
from diary.models import Post, Tag
from django.utils import timezone
import datetime


class Command(BaseCommand):
  '''新着タグ付与バッチ'''
  help = '投稿時刻が現在時刻から1週間以内の記事に新着タグをつける。1週間経過した記事は新着タグを外す。'

  def handle(self, *args, **options):
    tag = Tag.objects.get(slug='new')

    now = timezone.now()
    for post in list(Post.objects.all()):
      if post.created_at >= now - datetime.timedelta(weeks=1):
        # 投稿から1週間以内の記事に新着タグをつける
        post.tags.set([tag])
      else:
        # 1週間経過した記事は新着タグを外す
        post.tags.remove(tag)
      post.save()
    self.stdout.write(self.style.SUCCESS('新着タグ付与バッチが正常終了しました'))

コマンドラインで実行する処理は、BaseCommandクラスを継承すれば作れる。

公式の下記を見ると作り方がよくわかるので詳しくは下記を読んでみてくれ。

カスタム django-admin コマンドの実装 | Django ドキュメント

The web framework for perfectionists with deadlines.

この状態で、

$ python manage.py update_new_tag

をすれば、全記事の新着タグが更新される。

テストコードを書く

テストの雛形をつくる

では、テストコードを書いてみる。

まずはtests.pyの雛形を作るぞ。

from django.test import TestCase
from django.core.management import call_command
import freezegun

from diary.models import Post, Tag


# Create your tests here.
class 新着タグ付与バッチ(TestCase):
  '''#4 バッチを自動テストする'''

  # テスト用データを投入する
  fixtures = ['diary/fixtures/tests/test_update_new_tag.json']

  def setUp(self):
    pass

  def tearDown(self):
    pass

diary/fixtures/tests/test_update_new_tag.json のfixtureも用意するぞ。

fixtureはDBに入れる初期データだ。

[
{
    "model": "diary.user",
    "pk": 1,
    "fields": {
        "password": "pbkdf2_sha256$180000$v5YxxVKCObi7$SrCWUES4Qedh+fsHJIx2az1cMgS6wC4O7JapYTbj1/4=",
        "last_login": "2021-06-27T04:38:37.894Z",
        "is_superuser": true,
        "username": "yhei",
        "first_name": "",
        "last_name": "",
        "email": "yheihei0126@gmail.com",
        "is_staff": true,
        "is_active": true,
        "date_joined": "2021-06-07T13:05:56.903Z",
        "groups": [],
        "user_permissions": []
    }
},
{
    "model": "diary.post",
    "pk": 1,
    "fields": {
        "user": 1,
        "title": "新着がついているが1週間経過した記事",
        "body": "",
        "created_at": "2021-06-01T13:00:00.000Z",
        "updated_at": "2021-06-01T13:00:00.000Z",
        "tags": [1]
    }
},
{
    "model": "diary.post",
    "pk": 2,
    "fields": {
        "user": 1,
        "title": "新着タグなし、1週間未経過記事",
        "body": "",
        "created_at": "2021-06-06T13:00:00.000Z",
        "updated_at": "2021-06-06T13:00:00.000Z",
        "tags": []
    }
},
{
    "model": "diary.tag",
    "pk": 1,
    "fields": {
        "name": "new",
        "slug": "new",
        "created_at": "2021-06-01T12:00:00.000Z",
        "updated_at": "2021-06-01T12:00:00.000Z"
    }
}
]

「新着がついているが1週間経過した記事」をpostのid=1に入れる。”created_at”: “2021-06-01T13:00:00.000Z”として6/1の記事にしている。

「新着タグなし、1週間未経過記事」をpostのid=2に入れる。”created_at”: “2021-06-06T13:00:00.000Z”として、6/6の記事にしている。

わいへい

でもさ。今7/14だよ？

わいへい

このままバッチ実行したら、1週間経過したとみなされて、全記事のnewタグが外れるんじゃない？

お前にしては察しがいい。その通りだ。

テスト実行時刻を操作する

バッチは実行時の時刻がビジネスロジックに関わってくることがよくある。

そういう場合、freezegunを使ってテスト実行時刻を固定化するのが定石だ。

まずはfreezegunをinstallする。

$ pip install freezegun

次に、下記のようにtest関数に@freezegun.freeze_time('2021-06-09 13:00')のデコレーターを使ってやると、test関数内が指定した日時に固定される。

...
...
import freezegun
from django.utils import timezone
...

class 新着タグ付与バッチ(TestCase):
  '''#4 バッチを自動テストする'''
...
...

  @freezegun.freeze_time('2021-06-09 13:00')
  def test_1(self):
    # test_1実行時刻が 2021-06-09 13:00 になる
    print(timezone.now())

試しにこのテストを実行して、timezone.now()の出力値をprintしてみる。

$ python manage.py test
Creating test database for alias 'default'...
System check identified no issues (0 silenced).
2021-06-09 13:00:00+00:00
.
----------------------------------------------------------------------
Ran 1 test in 0.048s

OK

わいへい

日時が6/9になっている！

バッチ実行をする

バッチ実行はcall_command関数を使う。プログラム内からバッチを叩く関数だ。

from django.core.management import call_command
...
...
  @freezegun.freeze_time('2021-06-09 13:00')
  def test_1(self):
    '''バッチを実行すると 1週間経過した記事 の新着タグが外れること'''
    call_command('update_new_tag')

バッチを実行すると新着タグが外れる or 付与されるテストを書く

いよいよassertを行う。

バッチ実行後に、1週間経過した記事のnewタグが外れていれば良い。

こんな感じに書いてみる。

  @freezegun.freeze_time('2021-06-09 13:00')
  def test_1(self):
    '''バッチを実行すると 1週間経過した記事 の新着タグが外れること'''
    call_command('update_new_tag')

    # 新着がついているが1週間経過した記事 を取得
    post = Post.objects.get(pk=1)
    self.assertEqual(0, len(post.tags.all()))  # newタグがなくなっている

さらに、1週間経っていない記事にnewタグを付与するテストも書くぞ。

  @freezegun.freeze_time('2021-06-09 13:00')
  def test_2(self):
    '''バッチを実行すると 新着タグなし、1週間未経過記事 に新着タグがつくこと'''
    call_command('update_new_tag')

    # 新着タグなし、1週間未経過記事 を取得
    post = Post.objects.get(pk=2)
    self.assertEqual(1, len(post.tags.all()))
    # 新着タグがついている
    self.assertEqual('new', post.tags.first().slug)  # newタグがついている

テストを実行してみる。

$ python manage.py test -v 2
...
...
test_1 (diary.tests.新着タグ付与バッチ)
バッチを実行すると 1週間経過した記事 の新着タグが外れること ... 新着タグ付与バッチが正常終了しました
ok
test_2 (diary.tests.新着タグ付与バッチ)
バッチを実行すると 新着タグなし、1週間未経過記事 に新着タグがつくこと ... 新着タグ付与バッチが正常終了しました
ok

わいへい

大丈夫そうだね！

準正常のテストを書いて、機能追加をする

これで機能はできただろうか？　実はそんなことはない。

わいへい

え、でも仕様通りの機能が作れていると思うけど

じゃあ例えば、newタグがDB内に存在しなかったらどうなる？

わいへい

あ〜。え〜っと。

どうなるか分からなかったら、テストを書いてみればいい。

ちょっと書いてみてくれ。

わいへい

え〜っと…

  @freezegun.freeze_time('2021-06-09 13:00')
  def test_3(self):
    '''新着タグがDBにない場合に、バッチを実行したら...'''
    # 新着タグを全削除してからバッチ実行
    Tag.objects.all().delete()

    call_command('update_new_tag')

わいへい

test関数の最初で、新着タグを全削除してからcall_commandを呼んでみた。

わいへい

このままテストを実行すると〜。。。

$ python manage.py test diary.tests.新着タグ付与バッチ.test_3
Creating test database for alias 'default'...
System check identified no issues (0 silenced).
E
======================================================================
ERROR: test_3 (diary.tests.新着タグ付与バッチ)
新着タグがDBにない場合に、バッチを実行したら...
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/Users/yhei/.virtualenvs/env1/lib/python3.8/site-packages/freezegun/api.py", line 778, in wrapper
    result = func(*args, **kwargs)
  File "/Users/yhei/github/base_diary/diary/tests.py", line 48, in test_3
    call_command('update_new_tag')
  File "/Users/yhei/.virtualenvs/env1/lib/python3.8/site-packages/django/core/management/__init__.py", line 168, in call_command
    return command.execute(*args, **defaults)
  File "/Users/yhei/.virtualenvs/env1/lib/python3.8/site-packages/django/core/management/base.py", line 369, in execute
    output = self.handle(*args, **options)
  File "/Users/yhei/github/base_diary/diary/management/commands/update_new_tag.py", line 12, in handle
    tag = Tag.objects.get(slug='new')
  File "/Users/yhei/.virtualenvs/env1/lib/python3.8/site-packages/django/db/models/manager.py", line 82, in manager_method
    return getattr(self.get_queryset(), name)(*args, **kwargs)
  File "/Users/yhei/.virtualenvs/env1/lib/python3.8/site-packages/django/db/models/query.py", line 415, in get
    raise self.model.DoesNotExist(
diary.models.Tag.DoesNotExist: Tag matching query does not exist.

----------------------------------------------------------------------
Ran 1 test in 0.049s

FAILED (errors=1)
Destroying test database for alias 'default'...

わいへい

ぎゃー！エラーが出た！

騒ぐな。エラー内容を見ろ。

  File "/Users/yhei/github/base_diary/diary/management/commands/update_new_tag.py", line 12, in handle
    tag = Tag.objects.get(slug='new')

単純に下記の★部分でエラーが出てるだけだ。

class Command(BaseCommand):
...
  def handle(self, *args, **options):
    tag = Tag.objects.get(slug='new')  # ★ここで diary.models.Tag.DoesNotExist
...

全てのタグが削除されたら、そりゃTag.objects.get(slug='new')でエラーが出るわな。

わいへい

あ、なるほど

タグを作ってないとき、このバッチは機能しない。

なので、タグが作られていないときは

自動でnewタグを作成し
newタグをつけたり外したりする

という仕様を付け加えてみよう。

わいへい

じゃあ、早速機能実装を。。。

まあ待て。

せっかくだからテストコードから書いてみてくれ。

わいへい

え、でも。。。機能を書きながらやったほうが早いんじゃ。。。

テストコードで仕様を表現してから機能を実装したほうがいい。

完成形ができたかどうかはテストが証明してくれる。

わいへい

(めんどくせえな。。。) こんな感じかな

  @freezegun.freeze_time('2021-06-09 13:00')
  def test_3(self):
    '''新着タグがDBにない場合に、バッチを実行したら...'''
    # 新着タグを全削除してからバッチ実行
    Tag.objects.all().delete()

    call_command('update_new_tag')

    # 新着タグなし、1週間未経過記事 を取得
    post = Post.objects.get(pk=2)
    self.assertEqual(1, len(post.tags.all()))
    # 自動で新着タグが生成され、付与されている
    self.assertEqual('new', post.tags.first().slug)

OK。

次は機能実装だ。

class Command(BaseCommand):
  '''新着タグ付与バッチ'''
  help = '投稿時刻が現在時刻から1週間以内の記事に新着タグをつける。1週間経過した記事は新着タグを外す。'

  def handle(self, *args, **options):
    tag, _ = Tag.objects.get_or_create(
      slug='new',
      defaults=dict(
        name='new',
        slug='new'
      ),
    )

    now = timezone.now()
    for post in list(Post.objects.all()):
      if post.created_at >= now - datetime.timedelta(weeks=1):
        # 投稿から1週間以内の記事に新着タグをつける
        post.tags.set([tag])
      else:
        # 1週間経過した記事は新着タグを外す
        post.tags.remove(tag)
      post.save()
    self.stdout.write(self.style.SUCCESS('新着タグ付与バッチが正常終了しました'))

わいへい

get_or_createを使って、newタグがなければ、newタグを作る実装にしてみた

わいへい

テストを実行すると・・・

$ python manage.py test diary.tests.新着タグ付与バッチ.test_3
Creating test database for alias 'default'...
System check identified no issues (0 silenced).
新着タグ付与バッチが正常終了しました
.
----------------------------------------------------------------------
Ran 1 test in 0.087s

OK
Destroying test database for alias 'default'...

OK！！

大丈夫なようだな。

この通り、気になる挙動があったらテストコードを先に書いて不具合をあぶり出していく方法もオススメだ。

コマンドの引数をテストする

ウサギ

最後にコマンドに引数がある場合のテストを追加してみよう

例えば、以下のような機能を追加してみる。

manage.py update_new_tag --new_tag_slug=タグ名 とコマンドを叩くと
- 付与されるタグがnewではなく--new_tag_slugで指定したタグになる
- 指定したタグがなければ新規作成して付与する

実装は下記のような感じになる。

class Command(BaseCommand):
...
...
  def add_arguments(self, parser):
    parser.add_argument('--new_tag_slug', type=str)

  def handle(self, *args, **options):
    tag_slug = options.get('new_tag_slug') if options.get('new_tag_slug') else 'new'
    # tag = Tag.objects.get(slug=tag_slug)
    tag, _ = Tag.objects.get_or_create(
      slug=tag_slug,
      defaults=dict(
        name=tag_slug,
        slug=tag_slug
      ),
    )
...
...

ウサギ

add_argumentsの部分で引数の定義をし

ウサギ

options.get(‘new_tag_slug’) で与えられた引数を取得する

tag_slug = options.get('new_tag_slug') if options.get('new_tag_slug') else 'new'

↑の部分、慣れない記述方法かとは思う。三項演算子ってやつだ。

options.get('new_tag_slug')が指定されたときは、
options.get('new_tag_slug')を変数tag_slugに入れる。

options.get('new_tag_slug')が存在しなければ、'new'という文字列を変数tag_slugに入れるって書き方だ。

わいへい

new_tag_slug が指定されなければ、いつも通り newタグを使うってことか

そうだ。

ではcall_commandを使って、new_tag_slugを指定した時のテストを書いてみるぞ。

  @freezegun.freeze_time('2021-06-09 13:00')
  def test_4(self):
    '''新着タグをコマンドから指定できること'''
    call_command('update_new_tag', new_tag_slug='new2')

    # 新着タグなし、1週間未経過記事 を取得
    post = Post.objects.get(pk=2)
    self.assertEqual(1, len(post.tags.all()))
    # 指定した新着タグが生成され、付与されている
    self.assertEqual('new2', post.tags.first().slug)

call_commandの引数にnew_tag_slugを追加してやるだけだ。

簡単だな。

$ python manage.py test diary.tests.新着タグ付与バッチ.test_4 -v 2
...
...
test_4 (diary.tests.新着タグ付与バッチ)
新着タグをコマンドから指定できること ... 新着タグ付与バッチが正常終了しました
ok

----------------------------------------------------------------------
Ran 1 test in 0.133s

OK

わいへい

テストもOKだね！