tl;dr

  • Python の retry モジュール超便利
  • で, ちゃんと意図した通りにリトライされているか心配になった
  • リトライされているかユニットテストのレベルでテストしたい

さあ, どーしよう.

尚, 本記事で利用している環境は以下の通り.

$ sw_vers
ProductName:    Mac OS X
ProductVersion: 10.11.6
BuildVersion:   15G20015

$ python --version
Python 3.6.4

retry モジュール

retry モジュール

retry モジュールは, まさにその名前の通り, 指定した関数のリトライ処理を簡単に実現するモジュール.

Easy to use retry decorator.

pypi.org

retry モジュールのシンプルな使い方

以下のようにデコレータを付与することで, 指定した回数, 失敗時の遅延時間等で対象となる関数において再処理を実施してくれる.

import logging
from retry import retry

@retry(tries=3, delay=2, backoff=2)
def zero_division():
    try:
        return 10 / 0
    except ZeroDivisionError as ex:
        logging.error(ex)
        raise

上記のコードの場合, zero_division() を最大 3 回試行し, エラーが発生した場合には 1 回目の試行は 2 秒後, backoff が定義されているので, 2 回目の試行は 4 秒後に行う. 実際にコードを動かしてみると, 以下のように, 初回を含めると 3 回試行されていることが判る.

>>> import logging
>>> from retry import retry
>>>
>>> @retry(tries=3, delay=2, backoff=2)
... def zero_division():
...   try:
...       return 10 / 0
...   except ZeroDivisionError as ex:
...       logging.error(ex)
...       raise
...
>>> zero_division()
ERROR:root:division by zero
WARNING:retry.api:division by zero, retrying in 2 seconds...
ERROR:root:division by zero
WARNING:retry.api:division by zero, retrying in 4 seconds...
ERROR:root:division by zero

わざわざ, 自分でリトライ処理を書く必要が無いのが嬉しい. もし, retry モジュールを使わずにリトライ処理を実装するとしたら, 以下のようになるかもしれない.

import logging
from time import sleep
TRY = 3

def zero_division_no_retry_module():
    for r in range(1, TRY):
        try:
            return 10 / 0
        except ZeroDivisionError as ex:
            logging.warn('%s, %d 回目, %d 秒待機します.' % (ex, r, (r * 2)))
            sleep (r * 2)
    logging.error('%d 回目, エラーが発生しました. 処理を終了します.' % TRY)
    return False

試しに実行してみる.

>>> import logging
>>> from time import sleep
>>> TRY = 3
>>>
>>> def zero_division_no_retry_module():
...     for r in range(1, TRY):
...         try:
...             return 10 / 0
...         except ZeroDivisionError as ex:
...             logging.warn('%s, %d 回目, %d 秒待機します.' % (ex, r, (r * 2)))
...             sleep (r * 2)
...     logging.error('%d 回目, エラーが発生しました. 処理を終了します.' % TRY)
...     return False
...
>>> zero_division_no_retry_module()
WARNING:root:division by zero, 1 回目, 2 秒待機します.
WARNING:root:division by zero, 2 回目, 4 秒待機します.
ERROR:root:3 回目, エラーが発生しました. 処理を終了します.
False

んー, retry を使うよりは複雑な感じ.

ということで, そのリトライ, ちゃんとリトライしてますか?

まずは…

先程からちょいちょい出てくる, 0 除算のコードを再掲.

# file name: retry_sample.py
import logging

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def zero_division():
    try:
        return 10 / 0
    except ZeroDivisionError as ex:
        logger.error(ex)
        raise

以下のようなテストを書いた.

# file name: test_retry_sample.py
import unittest
import retry_sample

class TestRetrySample(unittest.TestCase):
    def test_retry_sample_exception(self):
        with self.assertRaises(ZeroDivisionError):
            retry_sample.zero_division()

例外が起きることを期待する.

$ python -m unittest tests.test_retry_sample -v
test_retry_sample (tests.test_retry_sample.TestRetrySample) ... division by zero
ok

----------------------------------------------------------------------
Ran 1 test in 0.001s

OK

例外が起きることはテストで確認出来た. しかも, logging が設定されているので, ログも出力されている.

リトライ

先程の 0 除算コードを以下のように修正し, retry 処理を追加した.

import logging
from retry import retry

logger = logging.getLogger()
logger.setLevel(logging.INFO)

@retry(tries=3, delay=2, backoff=2)
def zero_division():
    try:
        return 10 / 0
    except ZeroDivisionError as ex:
        logger.error(ex)
        raise

retry モジュールにより, 初回を含めて合計 3 回, この関数が実行されることを意図している.

これを, 先程書いたテストで確認してみる.

$ python -m unittest tests.test_retry_sample -v
test_retry_sample (tests.test_retry_sample.TestRetrySample) ... division by zero
division by zero
division by zero
ok

----------------------------------------------------------------------
Ran 1 test in 6.012s

OK

例外メッセージが 3 回出力されているし, retry モジュールの delay = 2backoff = 2 が効いているので, 初回から 2 秒後, 4 秒後にリトライされてテスト自体の処理時間は 6 秒 (くらい) 掛かっていることが確認できた. 間違いなくリトライしているんだと思う.

で, リトライされていることをテストする

先程の通り, 例外メッセージが 3 回出力されていることから, retry モジュールによって再処理が行われていることが確認出来たが…やはり, ちゃんと処理が 3 回行われることをテストで確認したい.

ということで, testfixtures モジュールを利用して, 例外メッセージの出力を捕捉することで, retry による再処理が実施されていることをテストする.

A collection of helpers and mock objects for unit tests and doc tests.

pypi.org

testfixturesは,

TestFixtures is a collection of helpers and mock objects that are useful when writing unit tests or doc tests.

上記の通り, ユニットテストなどで利用する mock オブジェクトをお手軽に作成できるモジュール. 今回は, 標準出力や標準エラーへの出力をキャプチャするクラスを利用して, 以下のようにテストを書いた.

import unittest
from testfixtures import OutputCapture

import retry_sample

class TestRetrySample(unittest.TestCase):
    def test_retry_sample_exception(self):
        with self.assertRaises(ZeroDivisionError):
            retry_sample.zero_division()

    def test_retry_sample_exception_retry(self):
        with OutputCapture() as output:
            with self.assertRaises(ZeroDivisionError):
                retry_sample.zero_division()
            output.compare('\n'.join([
                'division by zero',
                'division by zero',
                'division by zero',
                ''
            ]))

division by zero という例外メッセージ (エラーログ) が 3 回出力されることを期待するテスト. retry モジュールによる試行回数は 3 回なので, ログ自体も 3 回出力されるはず.

テストを実行すると, 以下のように出力される.

$ python -m unittest tests.test_retry_sample -v
test_retry_sample_exception (tests.test_retry_sample.TestRetrySample) ... division by zero
division by zero
division by zero
ok
test_retry_sample_exception_retry (tests.test_retry_sample.TestRetrySample) ... ok

----------------------------------------------------------------------
Ran 2 tests in 12.013s

OK

なんか, ログ出力の表示が若干ウザいけど, 意図した結果となった.

また, LogCapture クラスを利用することで, 上記のようにテスト結果出力にログが交じることもなくテストをすることを出来たので, LogCapture クラスを利用したほうが良いと思う.

import unittest
from testfixtures import LogCapture
import logging

import retry_sample

class TestRetrySample(unittest.TestCase):
    def test_retry_sample_exception_retry_log_output(self):
        with LogCapture(level=logging.ERROR) as output:
            with self.assertRaises(ZeroDivisionError):
                retry_sample.zero_division()
            output.check(
                ('root', 'ERROR', 'division by zero'),
                ('root', 'ERROR', 'division by zero'),
                ('root', 'ERROR', 'division by zero'))

初回の試行と, retry モジュールによる再試行の合計 3 回分のログが出力されることを期待している.

テストを走らせると, 以下のように出力される.

$ python -m unittest tests.test_retry_sample.TestRetrySample.test_retry_sample_exception_retry_log_output -v 
test_retry_sample_exception_retry_log_output (tests.test_retry_sample.TestRetrySample ) ... ok

----------------------------------------------------------------------
Ran 1 test in 6.010s

OK

テストが通ることを確認した. つまり, retry モジュールによるリトライが実施されることがテストコードにより確認された. また, 実行時間を見る限り, retry モジュールによる試行が行われていることが判る. 念の為, 以下のように 1 行のみログが出力されることを期待するテストコードを変更して実行してみると…

  $ diff -u test_retry_sample.py.bk test_retry_sample.py
--- test_retry_sample.py.bk     2018-05-27 23:35:47.000000000 +0900
+++ test_retry_sample.py        2018-05-27 23:35:56.000000000 +0900
@@ -27,6 +27,4 @@
             with self.assertRaises(ZeroDivisionError):
                 retry_sample.zero_division()
             output.check(
-                ('root', 'ERROR', 'division by zero'),
-                ('root', 'ERROR', 'division by zero'),
                 ('root', 'ERROR', 'division by zero'))

実際のログ出力は 3 行となる為, 以下のように, テストの結果は FAIL となる.

$ python -m unittest tests.test_retry_sample.TestRetrySample.test_retry_sample_exception_retry_log_output -v
test_retry_sample_exception_retry_log_output (tests.test_retry_sample.TestRetrySample) ... FAIL

======================================================================
FAIL: test_retry_sample_exception_retry_log_output (tests.test_retry_sample.TestRetrySample)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/path/to/python-unittest/tests/test_retry_sample.py", line 30, in test_retry_sample_exception_retry_log_output
    ('root', 'ERROR', 'division by zero'))
  File "/path/to/.pyenv/versions/3.6.4/lib/python3.6/site-packages/testfixtures/logcapture.py", line 180, in check
    recursive=self.recursive_check
  File "/path/to/.pyenv/versions/3.6.4/lib/python3.6/site-packages/testfixtures/comparison.py", line 563, in compare
    raise AssertionError(message)
AssertionError: sequence not as expected:

same:
(('root', 'ERROR', 'division by zero'),)

expected:
()

actual:
(('root', 'ERROR', 'division by zero'), ('root', 'ERROR', 'division by zero'))

----------------------------------------------------------------------
Ran 1 test in 6.041s

FAILED (failures=1)

意図した通りにリトライは行われているようで安心. ただ, より厳密 (delaybackoff から算出される時間等) に対してテストを行いたい場合には, retry モジュール自体のテストコードを見てみると, 以下のようにテストを行っているので, これを真似てテストを書くことが出来そう.

...
def test_retry(monkeypatch):
    mock_sleep_time = [0]

    def mock_sleep(seconds):
        mock_sleep_time[0] += seconds

    monkeypatch.setattr(time, 'sleep', mock_sleep)

    hit = [0]

    tries = 5
    delay = 1
    backoff = 2

    @retry(tries=tries, delay=delay, backoff=backoff)
    def f():
        hit[0] += 1
        1 / 0

    with pytest.raises(ZeroDivisionError):
        f()
    assert hit[0] == tries
    assert mock_sleep_time[0] == sum(
        delay * backoff ** i for i in range(tries - 1))
...

以上

retry モジュールを利用した際のリトライ処理が意図した回数行われているかをテストする方法を考えてみた. 今回は, エラー表示やログメッセージを捕捉して, その件数をカウントすることでテストするという手法を実装してみたが, もっと良い方法があるかもしれない. ということで, 改めて思いついたら続編を書いてみたいと思う.

元記事はこちら

リトライ処理が適切に行われるかをテストする方法を考えた Python 編