不安定なテスト(flaky test)を直す:時計・乱数・順序・スレッド

ときどき落ちるテストを Python で 5 種類再現し、失敗率を実測(25%・21%・70%・54%・8%)。UTC の時計注入、シード、後片付け、ロック、イベント待ちで直し、同じ回数の再測定で 0 にするまでの手順。

シリーズ:テスト・デバッグ講座 5/6(上級)。 CI が「だいたい緑」で、理由を考えずに「再実行」ボタンを押すようになってきた人向けです。この講座のあとは、flaky なテストを狙って再現し、失敗率を数え、症状ではなく原因を取り除けるようになります。

新人研修を終えて最初のプルリクエストを出し、コードレビューで「テストは?」と聞かれてテストを書いた。ところが翌週、そのテストが CI でたまに落ちる。先輩は「再実行したら通ったよ」と言う。これが flaky(フレーキー)なテスト、つまり同じコードで通ったり落ちたりするテストです。困るのは再実行の手間だけではありません。チームが赤いビルドを無視する癖をつけ、障害対応のときに本物の失敗がノイズに埋もれます。この講座では、現場で特に多い 5 つの原因(時計、乱数、テストの順序、スレッド、固定 sleep)をわざと起こし、何回に 1 回落ちるかを数え、直して、もう一度数えます。

環境: Python 3.9.6(macOS 標準の Python)、unittest のみでプラグインなし、Apple Silicon Mac (M1 Pro)、8 コア。下の割合はすべて実際に数えた値で、N を必ず併記しています。乱数そのものが論点でない箇所はシードを固定しました。パスは ~/demo/… に短縮しています。前提は 講座 3:はじめての単体テスト と 講座 4:テストダブル・モック・フィクスチャ です。

flaky とは何か、直すときのたった 1 つのルール

flaky なテストには「隠れた入力」があります。テスト対象のコードは毎回同じでも、テストが管理していない何か、つまり壁時計、乱数生成器、直前に走ったテスト、OS のスレッドの割り当て、マシンの混み具合が実行ごとに変わります。直し方の考え方はいつも同じで、隠れた入力を見つけて、明示的な入力にすることです。

自分をごまかさないためのルールは 1 つ。修正前と修正後の失敗率を、同じ N で測るまでは直ったと言わない。 もともとほとんど緑だったテストなら、1 回緑になっても何の証拠にもなりません。そこで以下の各節には「修正前」と「修正後」の数字が必ずあります。

いちばん簡単な測定道具は再実行ループで、どのテストランナーでも使えます。

fails=0
for i in $(seq 1 100); do
  python3 -m unittest -q test_coupon 2>/dev/null || fails=$((fails + 1))
done
echo "$fails/100 failed"
判断図。出発点はときどき落ちるテスト。まず、単独でループ実行しても落ちるか。いいえなら順序か共有状態:順序を入れ替えて確かめ、setUp か addCleanup で状態を戻す。はいなら、時計やタイムゾーンで結果が変わるか。はいなら UTC の時計を 1 つ注入し境界テストを足す。いいえなら乱数を使うか。はいならケースごとにシードを決めて表示する。それ以外はスレッドか待ち方:切り替え間隔を極小にして競合を出し、ロックを使い、sleep ではなくイベントを待つ。
ときどき落ちるテストの切り分け。どの問いも、コードを眺めるのではなくスクリプトで回せる実行で答える。

原因 1:シードのない乱数

クーポンコードの生成器です。仕様は「数字を最低 1 文字含む」で、テストはまさにそれを確かめています。

# coupon.py
import random
import string

ALPHABET = string.ascii_letters + string.digits   # 62 文字

def make_code(length=8, rng=random):
    """クーポンコードを返す。仕様:数字を最低 1 文字含む。"""
    return "".join(rng.choice(ALPHABET) for _ in range(length))
# test_coupon.py
class CouponTest(unittest.TestCase):
    def test_code_has_a_digit(self):
        code = make_code()
        self.assertTrue(any(c.isdigit() for c in code), code)

8 回続けて流すと OK, FAILED, OK, FAILED, OK, OK, FAILED, OK でした。落ちた回の出力はこうです。

FAIL: test_code_has_a_digit (test_coupon.CouponTest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "~/demo/flaky/test_coupon.py", line 8, in test_code_has_a_digit
    self.assertTrue(any(c.isdigit() for c in code), code)
AssertionError: False is not true : ZYRtMYHA

テストは正しく、コードが間違っています。数字が入る保証がどこにもありません。8 文字すべてが英字 52 文字から選ばれる確率は (52/62)^8 = 24.48%。実測では 10,000 回中 2,509 回(25.09%)、2 回目は 2,487 回(24.87%)でした。

よくある間違い → 実際に起きること → 直し方。 最初に思いつくのはテストの先頭で random.seed(1) を呼ぶことです。シード 1 ならコードは iK2ZWeqh で数字を含み、テストは永遠に通ります。シード 4 なら ptgUzEjf で永遠に落ちます。シード固定はバグを直さず、コイン投げを凍らせるだけです。役に立つのは多数のシードで回し、失敗のたびに再現用のシードを表示する形です。

SEED = int(os.environ.get("TEST_SEED", "1114"))

class CouponTest(unittest.TestCase):
    def check_many(self, fn):
        master = random.Random(SEED)
        for i in range(200):
            seed = master.randrange(2**32)
            code = fn(rng=random.Random(seed))
            with self.subTest(seed=seed):
                self.assertTrue(any(c.isdigit() for c in code), f"{code} (seed={seed})")

これでバグは毎回、決定的に落ちます。TEST_SEED=1114 では 200 ケース中 FAILED (failures=44)、最初の失敗は LoequqNo (seed=4195436511) で、REPL で make_code(rng=random.Random(4195436511)) を呼ぶと同じ LoequqNo が出ます。本当の修正はコード側で、数字を 1 文字わざと入れます。

def make_code_fixed(length=8, rng=random):
    chars = [rng.choice(ALPHABET) for _ in range(length - 1)]
    chars.insert(rng.randrange(length), rng.choice(string.digits))
    return "".join(chars)

修正後:10,000 シードで失敗 0。ここから身につけたい習慣は 2 つ。乱数生成器を引数で渡せるようにする(rng=random)こと、そしてシードを必ず表示することです。

原因 2:時計、日付の変わり目、夏時間

これは夜に落ちるので、障害対応の当番をいちばん苦しめます。期日のラベルを返す関数です。

# due.py
from datetime import date, datetime, timedelta, timezone

def due_label(due: date, today: date) -> str:
    if due == today:
        return "today"
    if due == today + timedelta(days=1):
        return "tomorrow"
    return "later" if due > today else "overdue"

def label_for_now_flaky(due: date) -> str:
    return due_label(due, datetime.now().date())          # ローカルの壁時計

def label_for(due: date, now_utc: datetime) -> str:       # 時計は 1 つ、引数で受け取る
    return due_label(due, now_utc.astimezone(timezone.utc).date())

flaky なテストは「明日」を UTC で作り、ローカルの日付と比べています。

def test_tomorrow(self):
    due = (datetime.now(timezone.utc) + timedelta(days=1)).date()   # UTC の日付
    self.assertEqual(label_for_now_flaky(due), "tomorrow")          # ローカルの日付

TZ=UTC なら通ります。UTC 06:00 ごろ、UTC より 8 時間遅れたゾーン(TZ=XST8、架空のゾーン名)でマシンの時計を動かすと、ローカルではまだ「昨日」なので落ちました。

AssertionError: 'later' != 'tomorrow'
- later
+ tomorrow

どのくらいの頻度で落ちるかは時刻しだいです。待つ代わりに、固定した時計で UTC の 1 日 1,440 分をすべて走査し、各分で同じロジックを評価しました。

マシンの時計 失敗する分(1,440 分中) 割合 最初の失敗(UTC) 修正後
UTC-8 480 33.3% 00:00 0
UTC-5 300 20.8% 00:00 0
UTC 0 0% なし 0
UTC+3 180 12.5% 21:00 0

「CI サーバーで、決まった時間帯だけ落ちる」という典型的な報告はこの表で説明がつきます。CI は UTC、開発者の PC は別のゾーンで動いていて、失敗する時間帯の長さはちょうど時差と同じです。

直したテストは時計を引数で渡し、境界を固定します。何時に流しても意味が変わりません。

class FixedDueTest(unittest.TestCase):
    def check(self, now):
        due = (now + timedelta(days=1)).date()
        self.assertEqual(label_for(due, now), "tomorrow")

    def test_tomorrow_one_second_before_midnight(self):
        self.check(datetime(2026, 11, 14, 23, 59, 59, tzinfo=timezone.utc))

    def test_tomorrow_at_midnight(self):
        self.check(datetime(2026, 11, 15, 0, 0, tzinfo=timezone.utc))

    def test_month_and_year_end(self):
        self.check(datetime(2026, 12, 31, 23, 30, tzinfo=timezone.utc))

夏時間(DST)。 1 日はいつも 86,400 秒とは限りません。架空の夏時間ゾーン(TZ="XST5XDT,M3.2.0,M11.1.0"、3 月第 2 日曜と 11 月第 1 日曜に切り替わる)でローカルの午前 0 時をタイムスタンプにし、86,400 で割りました。

2026-03-08 -> 2026-03-09 23 h int(seconds/86400) = 0 | date subtraction = 1
2026-11-01 -> 2026-11-02 25 h int(seconds/86400) = 1 | date subtraction = 1

int(seconds / 86400) は 3 月 8 日と 9 日の差を 0 日と言います。TZ=UTC では 4 組とも 24 時間でした。date 同士の引き算は常に正しい値です。時刻は UTC で保存・比較し、ローカル時刻への変換は表示のときだけにします。

よくある間違い:datetime.now をパッチする。 素直なモックは組み込み型で失敗します。

  File "…/python3.9/unittest/mock.py", line 1502, in __enter__
    setattr(self.target, self.attribute, new_attr)
TypeError: can't set attributes of built-in/extension type 'datetime.datetime'

自分のモジュール内の datetime という名前を、now() が固定値を返すサブクラスに差し替えることはできます(試したところ、flaky な関数は 23:59:59 で tomorrow を返しました)。ただ、時計を引数で渡すほうが単純で、依存が関数のシグネチャに見えます。

原因 3:テストの順序と共有状態

「単独なら通る、全体だと落ちる」はほぼ確実に共有状態です。日本円の価格計算モジュールで再現します。

# pricing.py
DEFAULTS = {"tax_rate": 0.10, "currency": "JPY"}
config = DEFAULTS            # バグ 1:コピーではなく別名

def set_option(key, value):
    config[key] = value

def price_with_tax(amount):
    return round(amount * (1 + config["tax_rate"]), 0)
class PricingTest(unittest.TestCase):
    def test_default_tax(self):
        self.assertEqual(pricing.price_with_tax(1000), 1100)

    def test_zero_tax_promo(self):
        pricing.set_option("tax_rate", 0.0)       # バグ 2:元に戻さない
        self.assertEqual(pricing.price_with_tax(1000), 1000)

    def test_rounding(self):
        self.assertEqual(pricing.price_with_tax(1980), 2178)

unittest はメソッドをアルファベット順に実行するので、test_zero_tax_promo はたまたま最後に走り、全体は緑です。名前が後ろに来るテストを 1 つ足すか、並列ワーカーにテストを分けると壊れます。Python 3.9 の python3 -m unittest には順序を入れ替えるオプションがないため、テストを読み込んで random.Random(seed)(シード 0〜199)で並べ替え、毎回モジュールを読み直して状態をまっさらにする 25 行ほどのランナーを書きました。結果は 200 通り中 139 通りが失敗(理論値 2/3。promo のテストが最後でないと落ちる)。最初に落ちたのはシード 0 の順序です。

first failing order (seed 0): ['test_default_tax', 'test_zero_tax_promo', 'test_rounding']
AssertionError: 1980.0 != 2178

修正はバグ 1 つにつき 1 つ。コード側は config = dict(DEFAULTS) と reset() 関数。テスト側は、アサーションが失敗しても必ず走る後片付けです。

class PricingTest(unittest.TestCase):
    def setUp(self):
        self.addCleanup(pricing.reset)          # テストが失敗しても実行される

    def test_zero_tax_promo_patch_dict(self):
        with mock.patch.dict(pricing.config, {"tax_rate": 0.0}):
            self.assertEqual(pricing.price_with_tax(1000), 1000)

修正後:200 通りの順序で失敗 0。config = DEFAULTS のままだと「既定値」自体が書き換わるので、reset() を作っても役に立たなかった点に注意してください。

原因 4:スレッド間の競合

4 つのスレッドが、共有カウンタにそれぞれ 50,000 回加算します。

class HitCounter:
    def __init__(self):
        self.value = 0
    def hit(self):
        self.value += 1            # 読む・足す・書く の 3 段階

期待値は 200,000。6 回流すと 194148、156216、199794、OK、104471、OK でした。GIL があっても += は不可分ではなく、読んでから書くまでの間にスレッドが切り替わり得ます。切り替えはおよそ sys.getswitchinterval() 秒ごと(このマシンでは 0.005)で、これが次の意外な結果の理由です。

設定(各 100 回) 失敗 1 回あたり
ロックなし、4 x 10,000 回 0 6.2 ms
ロックなし、4 x 20,000 回 0 12.2 ms
ロックなし、4 x 40,000 回 23 24.2 ms
ロックなし、4 x 50,000 回 54 30.4 ms
ロックなし、4 x 60,000 回 46 36.4 ms
ロックなし、4 x 1,000、sys.setswitchinterval(1e-6) 100 2.3 ms
threading.Lock、4 x 1,000、setswitchinterval(1e-6) 0 5.9 ms
threading.Lock、4 x 50,000 0 59.9 ms

小さなテスト(1 スレッド 10,000 回)は 一度も落ちませんでした。各スレッドが 5 ms のタイムスライス内に仕事を終えるからです。競合バグがレビューも単体テストもすり抜け、本番の負荷で初めて出るのはこの仕組みです。覚えておきたいのは 6 行目で、切り替え間隔を極小にすると、まれな競合が 2 ms のテストで 100% 再現するようになります。確実に落ちるようになってからロックを入れ、同じ厳しい設定で直ったことを確かめます。

class SafeHitCounter:
    def __init__(self):
        self.value = 0
        self._lock = threading.Lock()
    def hit(self):
        with self._lock:
            self.value += 1

ロックで 200,000 回の加算コストは約 2 倍(30.4 ms → 59.9 ms)になりました。カウンタなら問題ありませんが、ホットな箇所なら測る価値があります。GIL がスレッドをどう切り替えるかは スレッド・プロセス・asyncio を、単一スレッドのイベントループ特有の順序の罠は JavaScript の Promise とイベントループ を参照してください。

原因 5:sleep で待つテスト

バックグラウンドのワーカーが結果を計算し、テストは「十分に待つ」。

class ThumbnailWorker:
    def __init__(self):
        self.result = None
        self.done = threading.Event()

    def submit(self, n):
        def run():
            self.result = sum(i * i for i in range(n))   # CPU で約 16 ms
            self.done.set()
        threading.Thread(target=run, daemon=True).start()
w = ThumbnailWorker(); w.submit(400_000)
time.sleep(0.02)                      # 「これで足りるはず」
assert w.result is not None

マシンが空いていれば 200 回中失敗 0。そこに実際のテストスイートでよくある状況、つまり別のテストが止め忘れた CPU を使い続けるスレッドを 1 本足しました。200 回を 4 セット流すと 17・9・19・21 回 落ちました(平均 8.3%)。忙しいスレッドが GIL を半分握るので、16 ms の仕事がおよそ 2 倍かかります。面白いことに、OS レベルの負荷(8 コアに CPU を食うプロセス 12 個)では 200 回中 1 回しか落ちず、本当の引き金は同じプロセス内のスレッドでした。

待ち方(忙しいスレッドあり、200 回) 失敗 平均の待ち時間
time.sleep(0.02) 9〜21 33 ms
time.sleep(0.05) 0 63.4 ms
w.done.wait(timeout=5) 0 20.1 ms

長い sleep は全実行を遅くすることで flaky を「直し」ます。イベント待ちは仕事が終わった瞬間に戻るので、確実で、しかも最速です。タイムアウトは本物のバグのときに固まらず失敗させるためだけにあります。

止め忘れたスレッドには専用のチェックを付けましょう。次の基底クラスは、スレッドを残したままのテストを失敗させます。

class NoLeakedThreads(unittest.TestCase):
    def setUp(self):
        before = set(threading.enumerate())
        def check():
            leaked = [t.name for t in threading.enumerate() if t not in before and t.is_alive()]
            self.assertEqual(leaked, [], "test left threads running")
        self.addCleanup(check)

poller スレッドを起動して止め忘れたテストは AssertionError: Lists differ: ['poller'] != [] で失敗し、さらに FAILED と表示したあとプロセス全体が固まりました。デーモンでないスレッドが Python を生かし続けるためで、5 秒後に alarm で止めました(終了コード 142)。CI のジョブが「テストは終わったのに終わらない」ときは、まずこれを疑ってください。

修正前の失敗率の棒グラフ。シードなしの乱数コード:25.1%。UTC-5 の時計で 1 日の全分を走査した日付テスト:20.8%。ランダム順で共有状態:69.5%。4 スレッド x 50,000 回の競合:54%。忙しいスレッドがある状態で固定 20 ms の sleep:平均 8.3%。どれも修正後は 0。
修正前の失敗率。Apple Silicon Mac (M1 Pro)、Python 3.9.6 で測定。修正後は同じ測り方で失敗 0。N:コード 10,000 個、1,440 分、200 通りの順序、100 回、200 回 x 4。

修正前と修正後を 1 枚の表に

原因 修正前 修正後 直し方
シードなしの乱数 2,509/10,000(25.09%) 0/10,000 シード 性質をコードで保証、ケースごとにシード、シードを表示
ローカル日付 vs UTC UTC-5 で 300/1,440 分 0/1,440 UTC の時計を 1 つ注入、23:59:59 と 00:00 を試す
共有状態・順序入替 139/200 通り 0/200 既定値をコピー、addCleanup(reset)、patch.dict
スレッド競合 4 x 50,000 54/100 0/100 threading.Lock、setswitchinterval(1e-6) で再現
sleep(0.02)・忙しいスレッド 66/800 0/200 Event.wait(timeout=…)、スレッドの残りを検出

隔離するか、消すか。 今日直せない flaky テストは、測った失敗率と一緒にはっきり印を付けます(例:@unittest.skip("flaky: issue-123, 54/100"))。そうすればチームが赤いビルドを無視する訓練にならずに済みます。「3 回リトライ」を恒久的な解決にしてはいけません。講座 6 では、10% で落ちる flaky テストが git bisect に間違ったコミットを指させる様子を測ります。

正直な限界・バージョンの注記

  • すべて Python 3.9.6 のみで実行しました。sys.setswitchinterval は Python 3.2 からあります。フリースレッド版の Python(3.13 以降)やほかのバージョンでは競合の頻度が変わり得ますが、試していません。
  • 3.9 の python3 -m unittest には順序のランダム化がありません。pytest-randomly などのプラグインは存在しますが、ここではインストールも実行もしていません。
  • 夏時間のデモは架空の POSIX ゾーン文字列で、実在の地域ではありません。実際のゾーン規則は tz データベースから来て、法改正で変わります。
  • 失敗率はマシンしだいです。数字は「このくらい大きくなり得る」目安として読み、上のループで自分の環境を測ってください。
  • flaky な失敗を追うときは、シード・時計の値・テスト順序をログに残すと次の発生を再現できます。勘ではなくログが本当の原因を見つけた例は AI エージェントの誤診をログで見抜いた話 にあります。シリーズ全体の位置づけは プログラミング学習ロードマップ へ。

練習問題

  1. shuffle_deck() が同じカードを 2 回出さないことを確かめるテストが、先週 CI で 1 回だけ落ちました。まず何を 2 つ足しますか?
  2. 「前日分」のファイルを選ぶレポート処理が date.today() を使っています。CI は UTC、社内のサーバーは別のゾーンです。どんな値をテストで固定しますか?

解答

  1. 乱数生成器を shuffle_deck(rng=...) で渡せるようにし、subTest で多数のシードを回して失敗時にシードを表示します。そのうえで、コードに触る前にループで失敗率を測ります。1 回でも見えた失敗は、測定で否定されるまで本物です。
  2. now(UTC)を引数で渡し、少なくとも UTC の 23:59:59 と 00:00:00、月初、1 月 1 日を試します。ローカル時刻を使う必要があるなら、固定したゾーンで夏時間の切り替え日もテストします。

前へ: テスト・デバッグ講座 4:テストダブル・モック・フィクスチャ · 次へ: テスト・デバッグ講座 6:git bisect で退行を探す

よくある質問

flaky なテストを直したと、どうやって証明しますか?

同じ N で修正前と修正後の失敗率を測ります。ループで繰り返し実行する(今回は 100〜10,000 回)、シード固定でテスト順序を入れ替える、固定した時計で 1 日の全分を走査する、のどれかです。今回のテストは修正前に 25%・21%・70%・54%・8% 落ち、修正後は同じ回数で 0 回でした。CI が 1 回緑になっただけでは何も証明できません。

単独では通るのに、全体で流すと落ちるのはなぜですか?

別のテストが共有状態(モジュール変数の dict、シングルトン、環境変数、ファイル)を変えて戻していないからです。今回のデモでは税率を 0 にしたまま戻さないテストがあり、200 通りの順序のうち 139 通りで失敗しました。setUp で addCleanup を登録して戻すか unittest.mock.patch.dict を使い、既定値の dict は参照ではなくコピーします。

datetime.datetime.now を直接モックできないのはなぜですか?

datetime.datetime は組み込み型で属性を置き換えられないからです。Python 3.9.6 で mock.patch('datetime.datetime.now') は TypeError: can't set attributes of built-in/extension type 'datetime.datetime' になりました。現在時刻を引数(時計)として関数に渡すか、自分のモジュール内の datetime という名前をサブクラスで差し替えます。

  • #テスト
  • #上級
  • #flaky test
  • #Python
  • #unittest
  • #スレッド
  • #タイムゾーン

t.laboratory

実際に作って運用しているアプリ・サーバー・サービスで起きた問題と、その解決に使ったコードを記録する技術ブログです。

運営者情報 · アプリ

↑