JEPA4Japan · チュートリアル

テスト、デバッグ、ロギング

2,331文字 7分で読めます #Python

動作を証明し、不具合を調べ、運用に役立つ記録を残します。

コース進捗 コース目次 24レッスン中 24件を公開中

三つの道具は、三つの問いに答える

プログラムが一度動くのはうれしいことです。でも、証拠としては足りません。テスト、デバッガー、ログは、それぞれ違う形で助けます。

  1. Test 契約は今も守られる?
  2. Debugger 今、何が起きている?
  3. Log 実行中に何が起きた?
それぞれの道具が、別の種類の証拠を集めます。

デバッガーで一度調べても、繰り返せるテストは残りません。ログを大量に出しても、正しさの証明にはなりません。今の問いに答える一番小さな道具を選びます。

テストは退屈なくらい毎回同じにする

決定的なテストは入力を管理し、公開された振る舞いを調べます。時計、ネットワーク、乱数、現在のフォルダー、テスト順、別のテストの残り物へ、こっそり依存してはいけません。

  1. Arrange 管理した入力を作る
  2. Act 一つの振る舞いを実行
  3. Assert 契約と比べる
小さなテストには、明確な準備、実行、検証があります。

unittestは、testで始まるメソッドを見つけます。同じ形の境界ケースはsubTest()にきれいに並べられます。

import unittest


def pace(minutes):
    if minutes < 0:
        raise ValueError("分数を負にはできません")
    return "短い" if minutes <= 20 else "長い"


class PaceTests(unittest.TestCase):
    def test_boundaries(self):
        for minutes, expected in [(0, "短い"), (20, "短い"), (21, "長い")]:
            with self.subTest(minutes=minutes):
                self.assertEqual(pace(minutes), expected)

    def test_negative_value_fails(self):
        with self.assertRaisesRegex(ValueError, "負"):
            pace(-1)


suite = unittest.defaultTestLoader.loadTestsFromTestCase(PaceTests)
result = unittest.TestResult()
suite.run(result)
print(result.testsRun, len(result.failures), len(result.errors))

出力:

2 0 0

空の入力、最後に受理する値、最初に拒否する値、想定した失敗を試します。浮動小数点の丸めを契約が許す場合だけassertAlmostEqual()を使い、文字列、整数、コレクションには完全一致を使います。失敗したテストは、実装の全行ではなく、壊れた一つの振る舞いを指すべきです。

モックにするのは狭い出入口だけ

モックは、呼び出しを覚える偽物の協力相手です。送信先、時計、クライアントのような外部の出入口に使い、すべての純粋なhelperを偽物にはしません。

from unittest.mock import Mock


def publish(summary, sender):
    sender(summary)
    return "送信済み"


sender = Mock()
print(publish({"minutes": 45}, sender))
print(sender.call_args)

出力:

送信済み
call({'minutes': 45})

依存性注入なら出入口が見え、テストではsender.assert_called_once_with(...)を使えます。patch()が必要なら、テスト対象コードが名前を探す場所をパッチします。report.pyがsendを直接importしたなら、元の定義元ではなくreport.sendです。実在するインターフェースにはspec=やautospec=Trueを使い、ほとんどの本物のコードはつないだままにします。

コードを変える前に手掛かりを読む

例外では、トレースバックの最後の一行、つまり型とメッセージを最初に読みます。次に近いフレームを上へ進み、実際の状態が期待から初めて外れた場所を探します。ドメインの説明を加えるときは、raise NewError(...) from causeで低水準の証拠を残します。

  1. 読む 最後の一行、次にフレーム
  2. 小さくする 最小の不正入力を探す
  3. 止める breakpoint()で調べる
  4. 直す 誤った仮定を一つ変える
先に証拠を集め、その後で編集します。
def safe_ratio(part, whole, *, debug=False):
    if debug:
        breakpoint()
    if whole == 0:
        raise ValueError("wholeを0にはできません")
    return part / whole


print(safe_ratio(6, 3))

通常の出力は2.0です。debug=Trueなら、通常のpdbプロンプトでp expression、次行へ進むn、関数へ入るs、スタックを見るwhere、続けるc、終了するqを使えます。状態を変える式で調査してはいけません。自動処理の前に無条件のbreakpointを消しましょう。人がいないのに、いつまでも待つことがあります。

観測できる学習レポートを作る

ログはデータの流れです。loggerがレコードを受け入れ、handlerが出力先とレベルを選び、formatterが安定した文字列を作ります。再利用するライブラリ内のroot loggerではなく、アプリが所有するloggerを入口で設定します。

次の標準ライブラリだけの完全なプロジェクトをobservable_report.pyとして保存します。

import logging
import sys
import unittest
from dataclasses import dataclass
from io import StringIO
from unittest.mock import Mock


@dataclass(frozen=True)
class Record:
    topic: str
    minutes: int


def parse_record(line):
    topic, separator, minutes_text = line.partition("|")
    if not separator or not topic.strip():
        raise ValueError("topic|minutesの形が必要です")
    try:
        minutes = int(minutes_text)
    except ValueError as cause:
        raise ValueError("分数は整数で指定してください") from cause
    if minutes <= 0:
        raise ValueError("分数は正である必要があります")
    return Record(topic.strip(), minutes)


def load_records(lines, logger):
    records = []
    for number, line in enumerate(lines, start=1):
        try:
            record = parse_record(line)
        except ValueError as error:
            logger.warning(
                "行%dをスキップ:%s",
                number,
                error,
                extra={"event": "invalid_record"},
            )
        else:
            records.append(record)
    logger.info(
        "%d件を読み込み",
        len(records),
        extra={"event": "import_complete"},
    )
    return records


def summarize(records):
    return {"count": len(records), "minutes": sum(r.minutes for r in records)}


def publish_summary(records, publisher):
    summary = summarize(records)
    publisher(summary)
    return summary


def configure_logger(stream):
    logger = logging.getLogger("observable_report")
    logger.handlers.clear()
    logger.propagate = False
    logger.setLevel(logging.INFO)
    handler = logging.StreamHandler(stream)
    handler.setLevel(logging.INFO)
    handler.setFormatter(logging.Formatter("%(levelname)s %(event)s %(message)s"))
    logger.addHandler(handler)
    return logger


class ReportTests(unittest.TestCase):
    def test_boundaries(self):
        cases = [("テスト | 1", Record("テスト", 1)), (" ログ | 45 ", Record("ログ", 45))]
        for text, expected in cases:
            with self.subTest(text=text):
                self.assertEqual(parse_record(text), expected)

    def test_invalid_minutes_fail(self):
        with self.assertRaisesRegex(ValueError, "整数"):
            parse_record("テスト | many")

    def test_logs_are_stable(self):
        stream = StringIO()
        records = load_records(["テスト | 20", "不正 | zero"], configure_logger(stream))
        self.assertEqual(records, [Record("テスト", 20)])
        self.assertEqual(
            stream.getvalue().splitlines(),
            [
                "WARNING invalid_record 行2をスキップ:分数は整数で指定してください",
                "INFO import_complete 1件を読み込み",
            ],
        )

    def test_publisher_doorway(self):
        publisher = Mock()
        summary = publish_summary([Record("テスト", 20)], publisher)
        self.assertEqual(summary, {"count": 1, "minutes": 20})
        publisher.assert_called_once_with(summary)


suite = unittest.defaultTestLoader.loadTestsFromTestCase(ReportTests)
test_result = unittest.TestResult()
suite.run(test_result)
if not test_result.wasSuccessful():
    raise AssertionError(test_result.failures + test_result.errors)
print(f"テスト:{test_result.testsRun}件、失敗:0件、エラー:0件")

logger = configure_logger(sys.stdout)
records = load_records(["テスト | 30", "不正 | many", "ログ | 25"], logger)
summary = summarize(records)
print(f"合計:{summary['count']}件、{summary['minutes']}分")

python3 observable_report.pyで実行します。

テスト:4件、失敗:0件、エラー:0件
WARNING invalid_record 行2をスキップ:分数は整数で指定してください
INFO import_complete 2件を読み込み
合計:2件、55分

テストは自分のStringIOを所有し、propagate = Falseが上位loggerからの重複を防ぎます。時刻、マシンのパス、global設定、ネットワークでテストは揺れません。安定したeventフィールドは検索できる文脈になります。秘密や予約済みのLogRecord名をextraへ入れてはいけません。

3つの小さなチャレンジ

  1. 0分、空のテーマ、受理できる最小の1分をsubtestへ追加する。
  2. 注入したモック送信先にRuntimeErrorを送出させ、失敗が見えるままだとテストする。
  3. すべてのログに安定したsourceフィールドを足し、formatterと完全一致テストを更新する。

第20章へ進めるかな?

  • 成功、境界、失敗に対する決定的なunittestを書ける
  • 繰り返すケースへsubTest()で名前を付けられる
  • 内部helper全部ではなく、協力相手の出入口一つだけをモックにできる
  • 編集前にトレースバックを読み、breakpoint()を意図して使える
  • 名前付きlogger、handler、level、formatter、安定した文脈を設定できる
  • レポートを実行し、振る舞い、ログ、送信先をテストした

次章では、これらの技術を使い、完成したコマンドラインアプリを磨き、パッケージ化し、届けます。