- tl;dr
- retry モジュール
— retry モジュール
— retry モジュールのシンプルな使い方 - ということで, そのリトライ, ちゃんとリトライしてますか？
— まずは…
— リトライ
— で, リトライされていることをテストする - 以上
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 モジュールは, まさにその名前の通り, 指定した関数のリトライ処理を簡単に実現するモジュール.
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 = 2 と backoff = 2 が効いているので, 初回から 2 秒後, 4 秒後にリトライされてテスト自体の処理時間は 6 秒 (くらい) 掛かっていることが確認できた. 間違いなくリトライしているんだと思う.
で, リトライされていることをテストする
先程の通り, 例外メッセージが 3 回出力されていることから, retry モジュールによって再処理が行われていることが確認出来たが…やはり, ちゃんと処理が 3 回行われることをテストで確認したい.
ということで, testfixtures モジュールを利用して, 例外メッセージの出力を捕捉することで, retry による再処理が実施されていることをテストする.
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)
意図した通りにリトライは行われているようで安心. ただ, より厳密 (delay や backoff から算出される時間等) に対してテストを行いたい場合には, 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 モジュールを利用した際のリトライ処理が意図した回数行われているかをテストする方法を考えてみた. 今回は, エラー表示やログメッセージを捕捉して, その件数をカウントすることでテストするという手法を実装してみたが, もっと良い方法があるかもしれない. ということで, 改めて思いついたら続編を書いてみたいと思う.
