Claude CodeでCeleryのタスクを書く — モックで検証しeagerに頼らない

Claude CodeにCeleryのタスクを書かせると、ブローカーなしでテストを通すために task_always_eager = True を入れた構成が出てきやすくなります。動くので手早く見えますが、Celeryの公式ドキュメントは、eagerモードをunit testには向かないものと位置づけています。ここでは、モックを軸にした検証の形をClaude Codeに守らせるための指示と、実際のテスト断片を示します。

先に結論 — eagerは「通ったように見える」道具

eagerモードは、タスクをキューに送らずその場で実行する設定です。これは「ワーカーで起きることのエミュレーション」にすぎず、実際の挙動とは多くの食い違いがあります。つまりeagerで通ったテストは、ワーカー上で動くことの証明になりません。

公式が勧めるのは、タスクの振る舞いをモックで検証することです。タスクはWebのビューに近く、シリアライズ、メッセージヘッダー、リトライだけを担当し、実際のロジックは別の場所に置くのが理想とされています。

Claude Codeに任せるなら、この方針をCLAUDE.mdに書いておく価値があります。指示がないと、Claudeは手近な「ブローカー不要で緑になる」近道を選ぶことがあるためです。

Celeryのテストは2系統ある

公式のテストガイドは、テストを2つに分けています。

2つのAPIは互換性がありません。本記事で扱うのは前者です。Dockerを立てる後者は別の道具で、専用のドキュメントがあります。

前者のpytestプラグインは、最初は無効の状態で配布されています。有効にする方法は3つあります。

• pip install celery[pytest] で入れる
• 環境変数 PYTEST_PLUGINS=celery.contrib.pytest を設定する
• ルートの conftest.py に pytest_plugins = ("celery.contrib.pytest", ) を書く

チームで共有するなら、リポジトリに残る conftest.py に書く方法が追いやすい構成です。

タスクを薄く書かせる

モックで検証しやすいタスクは、ロジックを別モジュールに逃がした薄いタスクです。Claudeに最初から次の形で書かせると、後のテストが楽になります。以下は設計の一例です。

# proj/tasks.py
from decimal import Decimal
 
from django.db import OperationalError
 
from proj.celery import app
from proj.orders import place_order  # ロジックは別モジュール
 
 
@app.task(bind=True)
def send_order(self, product_pk, quantity, price):
    price = Decimal(price)  # JSONでは文字列として運ばれる
    try:
        place_order(product_pk, quantity, price)
    except OperationalError as exc:
        raise self.retry(exc=exc)

公式の例にもある通り、引数のDecimalはJSONで文字列になるため、タスク側で戻します。モデルのインスタンスではなく主キーを渡す点も同じです。タスクに書くのは、この変換とリトライの扱いだけです。

bind=True を付けたタスクは、第1引数にタスク自身(self)を受け取ります。リトライを書くときに必要になる形です。

モックでタスクを検証する

公式のテストガイドが示すのは、タスクを普通の関数として直接呼び、内側の呼び出しを patch で差し替える方法です。ブローカーもワーカーも要りません。

# tests/tasks/test_tasks.py
from decimal import Decimal
from unittest.mock import patch
 
import pytest
from celery.exceptions import Retry
from django.db import OperationalError
 
from proj.tasks import send_order
 
 
@patch("proj.tasks.place_order")
def test_success(place_order):
    send_order(1, 3, "30.3")
    place_order.assert_called_with(1, 3, Decimal("30.3"))
 
 
@patch("proj.tasks.place_order")
@patch("proj.tasks.send_order.retry")
def test_failure_retries(send_order_retry, place_order):
    send_order_retry.side_effect = Retry()
    place_order.side_effect = OperationalError()
 
    with pytest.raises(Retry):
        send_order(1, 3, "30.6")

2つ目のテストでは retry 自体を差し替え、Retry 例外が上がることだけを確かめています。再送の待ち時間まで本物を動かさずに、リトライ分岐に入ることを検証する書き方です。

パッチの対象は「定義元」ではなく「使われている側」です。proj.orders.place_order ではなく proj.tasks.place_order を差し替える点は取り違えやすい箇所なので、CLAUDE.mdに一行書いておくと安心です。

ワーカーが要る検証だけ、埋め込みワーカーで書く

シリアライズやルーティングまで通したいときは、プラグインの celery_worker フィクスチャが使えます。別スレッドでワーカーを起動し、テストが終わると停止します。

# conftest.py
import pytest
 
 
@pytest.fixture(scope="session")
def celery_config():
    return {
        "broker_url": "amqp://",
        "result_backend": "rpc",
    }

テスト本体は、conftest.py ではなくテストファイルに置きます。pytestは conftest.py からテスト関数を集めないためです。

# tests/integration/test_worker.py
def test_add(celery_app, celery_worker):
    @celery_app.task
    def mul(x, y):
        return x * y
 
    celery_worker.reload()
    assert mul.delay(4, 4).get(timeout=10) == 16

この例ではブローカーに amqp:// を指定しているため、ブローカーなしの単体テストとは別枠で扱います。celery_config にはテスト用アプリの設定を返し、1つのテストだけ設定を変えたいときは @pytest.mark.celery(result_backend='rpc') のようなマークを使います。

注意点が2つあります。

• 既定では、実行中のタスクの完了を最大10秒待ち、超えると例外になります。celery_worker_parameters() が返す辞書の shutdown_timeout で変えられます
• geventとeventletのプールは、スイート全体でmonkeypatchを有効にしていない限り使えません

Claudeに書かせるなら、「埋め込みワーカーを使うテストは tests/integration/ に分け、ブローカーが無い環境では実行しない」と決めておくと、単体テストの速さが守られます。

eagerを使うなら何が欠けるか

それでもeagerを使いたい場面はあります。例えば、ビューからタスクを呼ぶ経路で、呼び出しの配線だけ確かめたいときです。その場合も、設定の意味を知っておくと誤解が減ります。

task_eager_propagates が無効のままだと、タスク内の例外は呼び出し元に伝わりません。例外を握ったままテストが緑になる可能性があり、eagerで通ったように見えるテストの落とし穴の一つです。保存についても、既定ではeagerの結果はバックエンドに書かれません。task_store_eager_result は task_always_eager が有効で、かつ task_ignore_result が無効のときにだけ効きます。

左のように書くと、Claudeはeagerを検証の主役に据えます。右のように範囲と理由を書けば、eagerの出番は配線の確認に絞られます。

冪等性を指示に入れる

Celeryは、ワーカーが落ちたときにメッセージが別のワーカーへ再配送されることがある設計です。公式は、タスク関数は冪等であるのが理想だと述べています。同じ引数で複数回呼ばれても、意図しない副作用が出ない関数という意味です。

既定では、ワーカーはタスクを実行する直前にメッセージを確認応答(ack)します。冪等なタスクなら acks_late を有効にして、タスクの完了後にackさせることもできます。ただし acks_late を有効にすると、ワーカーが実行の途中で落ちたときに同じタスクが複数回実行されることがあるため、冪等性が前提になります。

この性質はClaudeが勝手に守ってくれるものではないので、指示とテストの両方で固定します。

# tests/tasks/test_orders.py
from decimal import Decimal
 
from proj.orders import Order, place_order
 
 
def test_place_order_is_idempotent(db):
    place_order(1, 3, Decimal("30.3"))
    place_order(1, 3, Decimal("30.3"))  # 同じ引数で2回
 
    assert Order.objects.filter(product_id=1).count() == 1

db フィクスチャはpytest-djangoのもので、テスト用のデータベースを使えるようにします。Order は例示用のモデルで、import先は手元のプロジェクトに合わせて読み替えてください。このテストは place_order の実装に重複防止(一意制約や既存チェック)が入っていることを前提にした、例示のテストです。Claudeに「同じ引数で2回呼んでも副作用が1回分で済むことをテストで示す」と頼めば、実装側にも重複防止が入る方向に動きます。

コミット前にタスクを投げない

Djangoを使うなら、もう一つ指示に入れておきたい落とし穴があります。transaction.atomic の中で .delay() を呼ぶと、タスクが先に動いてコミット前のデータを探しにいく競合が起きることがあります。公式は、記事を作成してから主キーをタスクに渡す例でこれを説明しています。

対策は delay_on_commit() です。Djangoの on_commit を使い、トランザクションがコミットされてからタスクを起動します。Celery 5.4で追加されたメソッドなので、5.4より前のバージョンでは transaction.on_commit に functools.partial でタスクの呼び出しを渡します。

@transaction.atomic
def create_article(request):
    article = Article.objects.create()
    expand_abbreviations.delay_on_commit(article.pk)
    return HttpResponseRedirect("/articles/")

eagerモードでは、タスクがその場で同期実行されるため、この競合が再現しません。eagerで緑になったテストは、本番でだけ起きる順序の問題を見逃します。

CLAUDE.mdとルールに固定する

ここまでの方針を、CLAUDE.mdにまとめます。Claude CodeはCLAUDE.mdを毎セッションの開始時に読みますが、強制される設定ではなく文脈として扱います。そのため、検証できる具体的な書き方にします。公式の推奨は、1ファイル200行未満です。

## Celery
- タスクは薄く保つ。ロジックは別モジュールに置き、タスクは
  シリアライズ・引数の型変換・retryだけを持つ
- 単体テストはモックで書く。`patch` の対象は使われている側
  (`proj.tasks.place_order`)
- `task_always_eager` を検証の根拠にしない。使う場合は
  `task_eager_propagates` も有効にし、配線の確認だけに使う
- タスクは冪等にする。同じ引数で2回呼ぶテストを必ず付ける
- Djangoのトランザクション内では `delay_on_commit()` を使う
- 埋め込みワーカーのテストは `tests/integration/` に置く
- 変更後は `pytest -q tests/tasks` を実行し、出力を読んでから完了とする

Celery関連のファイルを触るときだけ効かせたいなら、パス指定付きのルールにする方法もあります。.claude/rules/ に置いたファイルに paths を書くと、一致するファイルをRead、Write、Editしたときに読み込まれます。

---
paths:
  - "**/tasks.py"
  - "tests/tasks/**/*.py"
---
 
# Celeryタスクのルール
- ロジックは別モジュールへ。タスクは薄く保つ
- テストはモックで書き、eagerを検証の根拠にしない

pytestの命名や配置のルールは、別記事のClaude CodeにpytestのCLAUDE.md規約を教えるにまとめています。Celeryの規約はそこに足す形で書けます。

検証をClaudeに回させる

規約を書いても、実行して出力を読まなければ検証にはなりません。「書いたら pytest を走らせ、失敗したら出力を読んで直す」という繰り返しをClaudeに行わせます。

毎回の許可プロンプトを避けたいなら、.claude/settings.json に許可ルールを足せます。ワイルドカードは Bash(npm run *) のように書く形式です。

{
  "permissions": {
    "allow": [
      "Bash(pytest *)"
    ]
  }
}

pytest -q tests/tasks

テストが通るまで止めない構成にしたい場合は、停止時にテストを検証させるagent hookの使い方をClaude Codeのagent hookで「テストが通るまで止めない」を検証させるで扱っています。FastAPIと組み合わせるなら、Claude CodeでFastAPIバックエンドを構築するの実装フローが前提になります。

よくあるつまずき

eagerのテストだけ緑で、本番で落ちる

eagerはエミュレーションなので、シリアライズ、ルーティング、トランザクションの順序を通りません。モックのテストと、必要な箇所の埋め込みワーカーのテストを足します。

パッチが効かない

差し替え先が定義元のモジュールになっていないか見ます。この記事の例では、タスクが参照している proj.tasks.place_order を差し替えています。

Retryの分岐を検証したい

公式の例と同じく retry を差し替えて Retry 例外を上げさせると、再送そのものを動かさずに分岐だけを検証できます。

埋め込みワーカーのテストが終わらない

完了待ちの上限(既定10秒)を超えると例外になります。時間のかかるタスクは shutdown_timeout を見直します。

まとめ

Celeryのタスクをテストする出発点は、eagerを避けてモックで検証し、実際のワーカーが要る箇所だけ埋め込みワーカーに任せる分担です。Claude Codeにはこれを、薄いタスク、冪等性のテスト、delay_on_commit の使い方、検証コマンドの実行まで含めてCLAUDE.mdに書いて渡します。エミュレーションが通ったことと、ワーカーで動くことは別の話だと、指示の中で区別しておくのが要点です。