Pytestで行うPythonのテストでtimeout設定をする
はじめに
データ事業本部のkobayashiです。
Pythonでテストを書いていると、外部APIの呼び出しやデータベースへの接続、複雑な処理などで、想定以上に時間がかかってしまうテストケースに遭遇することがあります。特にCIパイプラインでテストが無限にハングしてしまうと、全体の開発フローがストップしてしまい、大きな問題となります。
今回は、pytestでテストにタイムアウトを設定できるpytest-timeoutというプラグインを試してみました。
pytest-timeoutとは
pytest-timeoutは、個々のテストやテストセッション全体にタイムアウトを設定できるpytestプラグインです。テストが無限ループやデッドロックなどで永遠に終わらない問題を防ぎ、CI/CD環境でのテスト実行を安定させることができます。
主な特徴としては以下になります。
- テスト関数ごと、またはテストセッション全体にタイムアウト時間を設定可能
- 2つのタイムアウト方式(signal/thread)をサポート
- デバッガ使用時の自動タイムアウト無効化
- 設定ファイル、環境変数、コマンドライン、デコレータなど柔軟な設定方法
- フィクスチャのsetup/teardownにも適用可能
pytest-timeoutを使ってみる
環境
今回使用した環境は以下の通りです。
Python 3.12.8
pytest 8.3.4
pytest-timeout 2.3.1
インストール
pipで簡単にインストールできます。
$ pip install pytest-timeout
基本的な使い方
まず、時間のかかる処理を含むテストコードを用意します。
import pytest
import time
import requests
def fetch_data_from_api(url: str, timeout: int = 5) -> dict:
"""外部APIからデータを取得する"""
try:
response = requests.get(url, timeout=timeout)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
print(f"APIエラー: {e}")
return {}
def process_large_dataset(size: int) -> list:
"""大規模データセットを処理する"""
result = []
for i in range(size):
# 重い処理をシミュレート
time.sleep(0.001)
result.append(i ** 2)
return result
class TestBasicTimeout:
"""基本的なタイムアウトのテスト例"""
def test_quick_process(self):
"""高速に完了するテスト"""
result = process_large_dataset(100)
assert len(result) == 100
@pytest.mark.timeout(1) # 1秒でタイムアウト
def test_slow_process_with_timeout(self):
"""タイムアウトが設定されたテスト"""
result = process_large_dataset(500)
assert len(result) == 500
@pytest.mark.timeout(5)
def test_api_call_with_timeout(self):
"""API呼び出しのタイムアウトテスト"""
# テスト用のAPIエンドポイント(遅延を含む)
result = fetch_data_from_api("https://httpbin.org/delay/10")
assert isinstance(result, dict)
コマンドラインでのタイムアウト設定
全体的なタイムアウトを設定してテストを実行してみます。
$ pytest test_basic.py --timeout=10
上記のようにpytestのオプションに--timeoutを指定することで、全てのテストに対して一律でタイムアウトを設定できます。
また各テストケースのデコレータで@pytest.mark.timeout(1)のように指定することもできます。
実行してみるとタイムアウト設定が有効になっていることが、出力に表示されます。
def process_large_dataset(size: int) -> list:
"""大規模データセットを処理する"""
result = []
for i in range(size):
# 重い処理をシミュレート
> time.sleep(0.001)
E Failed: Timeout (>1.0s) from pytest-timeout.
pytest-timeoutは、デバッガ使用時に自動的にタイムアウトを無効化しますが--timeoutを使うことで明示的に無効化することも可能です。
# タイムアウトを無効化
$ pytest --timeout=0
# または環境変数で
$ PYTEST_TIMEOUT=0 pytest
高度な使い方
タイムアウトメソッドの選択
pytest-timeoutには2つのタイムアウト方式があります:
- signal方式:Unix系システムでのデフォルト。SIGALRMシグナルを使用
- thread方式:Windows環境や、signalが使えない場合に使用
import pytest
import time
import threading
def blocking_operation():
"""ブロッキング処理のシミュレーション"""
time.sleep(10)
return "完了"
def threaded_operation():
"""マルチスレッド処理のシミュレーション"""
results = []
def worker():
time.sleep(0.5)
results.append("thread_done")
threads = []
for _ in range(5):
t = threading.Thread(target=worker)
t.start()
threads.append(t)
for t in threads:
t.join()
return results
class TestTimeoutMethods:
"""タイムアウトメソッドのテスト"""
@pytest.mark.timeout(2, method='signal')
def test_with_signal_method(self):
"""signalメソッドでのタイムアウト"""
# このテストは2秒でタイムアウトする
result = blocking_operation()
assert result == "完了"
@pytest.mark.timeout(2, method='thread')
def test_with_thread_method(self):
"""threadメソッドでのタイムアウト"""
result = threaded_operation()
assert len(result) == 5
設定ファイルでのタイムアウト設定
pytest.ini、pyproject.toml、setup.cfgなどで設定できます:
[pytest]
timeout = 300
timeout_method = thread
timeout_func_only = false
[tool.pytest.ini_options]
timeout = 300
timeout_method = "signal"
セッション全体のタイムアウト
テストスイート全体の実行時間を制限することもできます:
$ pytest --session-timeout=3600 # 1時間でセッション全体をタイムアウト
実践的な使用例
統合テストでのタイムアウト設定
単純な単体テストだけでなく、実際の開発現場でよく遭遇するデータベースを使った統合テストでのタイムアウト設定をしてみます。
- 外部リソースへの依存 - データベース、API、ファイルシステムなど外部リソースを扱うテストは、接続の失敗やレスポンスの遅延により予想以上に時間がかかる可能性があるのでタイムアウト設定を行います。
- CI環境での安定性 - CI環境ではデータベースの接続が不安定になったり、ネットワークの問題でテストがハングすることがあります。タイムアウトを設定することで、CI/CDパイプライン全体がブロックされるのを防ぎます。
- リソースリークの防止 - データベース接続やトランザクションが適切にクローズされない場合、リソースリークが発生する可能性があります。タイムアウトにより、こうした問題を早期に検出します。
- パフォーマンス劣化の検出 - データ量の増加やクエリの複雑化により、徐々にパフォーマンスが劣化する問題を、タイムアウトによって検出します。
以下の例では、SQLiteを使った大量データのマイグレーション処理に対してタイムアウトを設定しています。
import pytest
import time
import sqlite3
from contextlib import contextmanager
@contextmanager
def database_connection(db_path: str):
"""データベース接続のコンテキストマネージャー"""
conn = None
try:
conn = sqlite3.connect(db_path, timeout=5.0)
yield conn
finally:
if conn:
conn.close()
def migrate_large_table(conn, table_size: int):
"""大規模テーブルのマイグレーション処理"""
cursor = conn.cursor()
# テーブル作成
cursor.execute("""
CREATE TABLE IF NOT EXISTS large_table (
id INTEGER PRIMARY KEY,
data TEXT
)
""")
# データ挿入(バッチ処理)
batch_size = 1000
for batch_start in range(0, table_size, batch_size):
batch_data = []
for i in range(batch_start, min(batch_start + batch_size, table_size)):
batch_data.append((i, f"data_{i}"))
cursor.executemany("INSERT INTO large_table VALUES (?, ?)", batch_data)
conn.commit()
time.sleep(0.01) # 処理遅延のシミュレーション
class TestIntegration:
"""統合テストのタイムアウト例"""
@pytest.mark.timeout(30) # 30秒でタイムアウト
def test_database_migration(self, tmp_path):
"""データベースマイグレーションのテスト"""
db_path = tmp_path / "test.db"
with database_connection(str(db_path)) as conn:
# 10000レコードのマイグレーション
migrate_large_table(conn, 10000)
# 結果確認
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM large_table")
count = cursor.fetchone()[0]
assert count == 10000
@pytest.mark.timeout(5)
def test_quick_database_operation(self, tmp_path):
"""高速なデータベース操作のテスト"""
db_path = tmp_path / "test_quick.db"
with database_connection(str(db_path)) as conn:
cursor = conn.cursor()
cursor.execute("CREATE TABLE test (id INTEGER, value TEXT)")
cursor.execute("INSERT INTO test VALUES (1, 'test_data')")
conn.commit()
cursor.execute("SELECT value FROM test WHERE id = 1")
result = cursor.fetchone()
assert result[0] == "test_data"
まとめ
pytest-timeoutは、テストの実行時間を管理し、ハングしたテストや過度に長いテストを検出するための強力なツールです。
- 柔軟な設定方法(デコレータ、コマンドライン、設定ファイル)により、様々な粒度でタイムアウトを管理できる
- signal/threadの2つの方式により、環境に応じた最適な方法を選択できる
- デバッガとの連携により、開発時の利便性を損なわない
- CI/CD環境でのテスト実行の安定性を大きく向上させる
テストスイートが大規模になり、外部リソースとの連携が増えるほど、タイムアウトの重要性は高まります。pytest-timeoutを導入することで、より堅牢で予測可能なテスト環境を構築できるようになります。
最後まで読んで頂いてありがとうございました。






