pytest の monkeypatch を threading を使って遅延パッチする方法

• 概要
• 環境
• ステータスを監視するコード
• monkeypatch を使って別のステータスを返却するようにする
• monkeypatch を遅延させるために threading を使う
• 最後に

概要

例えばステータスを監視するようなコードがある場合にステータスが非同期で変わります
ただそのステータスのレスポンス情報を monkeypatch してしまうと一定のレスポンス情報しか返らなくなりステータスを監視するループから抜け出せなくなります
そんな場合に threading を使って monkeypatch を遅延して適用する方法を使えば monkeypatch を使ってもうまくループを抜け出せます

環境

• macOS 10.15.6
• Python 3.8.5

ステータスを監視するコード

例えば以下のようなステータスを監視するコードを考えます
これは while ループで 5 秒ごとにステータス情報をチェックし stopped 以外に慣ればループを抜けます

• vim app.py

import time

def get_status():
    return "stopped"

def check_status():
    while get_status() == "stopped":
        print("status is stopped")
        time.sleep(5)
    return "status is running"

この状態だと get_status は常に stopped を返すためループを抜けれません

monkeypatch を使って別のステータスを返却するようにする

上記のコードを monkeypatch を使ってテストしループを抜け出せるようにしてみます

• vim test_app.py

from app import check_status

def test_check_status(monkeypatch):
    monkeypatch.setattr('app.get_status', lambda: "running")
    result = check_status()
    thread1.join()
    assert (result == "status is running")

• pipenv run pytest ./test_app.py -s

こんな感じで monkeypatch すれば簡単にループを抜けるテストが書けます
しかしここで問題なのはループ内の処理を一度も実行しないためカバレッジが上がりません
なので一回以上ループ内の処理をさせるように monkeypatch を遅延して適用したくなります

monkeypatch を遅延させるために threading を使う

monkeypatch を当てるための簡単なスレッドを作成すれば解決できました
テストのコードを以下のように書き換えます

import threading
import time
from app import check_status

def patch(monkeypatch):
    time.sleep(15)
    monkeypatch.setattr('app.get_status', lambda: "running")


def test_check_status(monkeypatch):
    thread1 = threading.Thread(target=patch, args=(monkeypatch,))
    thread1.start()
    result = check_status()
    thread1.join()
    assert (result == "status is running")

15 秒後に monkeypatch が当たるように変更してみます
threading は必ず start -> join してメインのスレッドに戻ってくるようにしましょう
これで実行すると「status is stopping」が 3 回表示された後にちゃんとループを抜けてテストが正常終了するのが確認できると思います

最後に

monkeypatch + threading で遅延パッチする方法を紹介しました
非同期な処理をパッチする場合には使えるテクニックかなと思います