TDDとリスク別テスト戦略で変更を証明する

到達目標

1. 同じ振る舞い要求に対するRED、GREEN、REFACTORを別々の実行結果とsource hashで証明できる

   • 実subprocessのreturn code、source hash、behavior hash、mutation結果を含むRED-GREEN-REFACTOR履歴
   • TDD cycleと製品全体のテスト戦略が同義ではない理由を説明する6分間の解説

2. 単体、統合、契約、性質、メタモルフィックの各テストを、検出したい失敗と費用へ対応付けられる

   • 実subprocessのreturn code、source hash、behavior hash、mutation結果を含むRED-GREEN-REFACTOR履歴
   • 失敗影響、観測境界、実行費用からテスト層を選び、過剰・不足を反証する判断記録

3. 順序と時刻へ依存する非決定的障害を固定seedと時計で再現し、mutationでテスト感度を検証できる

   • 失敗影響、観測境界、実行費用からテスト層を選び、過剰・不足を反証する判断記録
   • 順序と時刻に依存する未知の非決定的障害について再現条件と回帰戦略を組み立てた報告

能力の進行

1. recognize

   失敗する期待、最小実装、振る舞いを保つ整理をRED-GREEN-REFACTORとして識別できる

   証拠: 実subprocessのreturn code、source hash、behavior hash、mutation結果を含むRED-GREEN-REFACTOR履歴

2. explain

   テスト層ごとの観測境界、実行費用、偽陰性と偽陽性の違いを説明できる

   証拠: TDD cycleと製品全体のテスト戦略が同義ではない理由を説明する6分間の解説

3. apply

   一つの要求を実行可能な例へ落とし、失敗を確認してから最小実装とrefactorを行える

   証拠: 実subprocessのreturn code、source hash、behavior hash、mutation結果を含むRED-GREEN-REFACTOR履歴

4. diagnose

   固定seed、時計、順序、環境差を分離し、flaky testの再現条件を証拠で特定できる

   証拠: 失敗影響、観測境界、実行費用からテスト層を選び、過剰・不足を反証する判断記録

5. lead

   変更リスクへ複数のテスト技法を配分し、mutationと本番観測を含む品質戦略を更新できる

   証拠: 順序と時刻に依存する未知の非決定的障害について再現条件と回帰戦略を組み立てた報告

なぜ重要か

testの役割は安心を演出することではなく、変更について誤り得る具体的な主張を実行可能にすることだ。greenやcoverage率だけでは、期待が正しいこと、重要な境界を観測したこと、利用者のriskを減らしたことは証明できない。

TDDは小さな設計feedback loopとして有効だが、製品全体のtest戦略と同義ではない。process境界、consumerとの契約、非決定的な時刻と順序、本番でしか見えない失敗には、別の観測と反証が要る。

メンタルモデル

まずriskを「どの境界で、何が壊れ、誰にどの影響が出るか」で記述する。次に、その失敗を最短で観測できる層へtestを置く。test数の多さではなく、異なる失敗仮説への感度と診断可能性を設計する。

動く例で考える

割引計算を実subprocessでREDからmutation killまで進める

前提

価格は0以上、割引率は0から100まで。合計は入力順に依存せず、価格0を追加しても変わらない。

入力

150円に10%を適用すると135.0円となる純粋関数、SQLite repository、provider response、2205件の有限入力、順序交換と0追加を五技法へ割り当てる。

操作

明示した単体test IDをNotImplementedErrorの未実装、最小実装、入力検証を分離した実装へ順に実行する。五技法は別test IDで個別実行し、seed 11から14と固定schedule・時計でflaky fixtureを実行して安全な算術mutantを試す。

観測

REDは非0、GREENとREFACTORは0。後二者のsource SHA-256は異なるがbehavior SHA-256は同じで、mutantは非0になる。

結論

履歴は文字列で自己申告せず、実行command、return code、hash、seedをJSONへ保存する。これによりreviewerが各主張を再実行できる。

次のfixtureはharness自身が専用のTemporaryDirectoryを作り、その配下だけへ教材fileを書く。各subprocessは専用cwdと明示test IDを使うため、呼出元のsymlinkやambient testを発見しない。shell、外部network、秘密情報を使わず、各実行へ5秒の上限を設ける。

python3.13 - <<'PY'
import hashlib
import json
import os
from pathlib import Path
import subprocess
import sys
from tempfile import TemporaryDirectory

HARNESS = "test_strategy_lab_v1"
TEST_MODULE = "test_strategy_fixture"
TEST_IDS = {
    "unit": f"{TEST_MODULE}.PureUnitTests.test_discount",
    "integration": (
        f"{TEST_MODULE}.SQLiteIntegrationTests."
        "test_repository_round_trip"
    ),
    "contract": (
        f"{TEST_MODULE}.ProviderContractTests."
        "test_schema_and_semantics"
    ),
    "property": (
        f"{TEST_MODULE}.BoundedPropertyTests."
        "test_discount_bounds"
    ),
    "metamorphic": (
        f"{TEST_MODULE}.MetamorphicTests."
        "test_permutation_and_zero_item"
    ),
}
TEST_SOURCE = """
import sqlite3
import unittest
from order_discount import DiscountRepository, provider_response, total

class PureUnitTests(unittest.TestCase):
    def test_discount(self):
        self.assertEqual(total([100, 50], 10), 135.0)

class SQLiteIntegrationTests(unittest.TestCase):
    def test_repository_round_trip(self):
        connection = sqlite3.connect(":memory:")
        try:
            repository = DiscountRepository(connection)
            repository.save("order-1", [100, 50], 10)
            self.assertEqual(repository.load_total("order-1"), 135.0)
        finally:
            connection.close()

class ProviderContractTests(unittest.TestCase):
    def test_schema_and_semantics(self):
        response = provider_response("order-1", [100, 50], 10)
        self.assertEqual(
            set(response),
            {"order_id", "total", "currency"},
        )
        self.assertIs(type(response["order_id"]), str)
        self.assertIs(type(response["total"]), float)
        self.assertEqual(response["currency"], "JPY")
        self.assertEqual(response["total"], total([100, 50], 10))

class BoundedPropertyTests(unittest.TestCase):
    def test_discount_bounds(self):
        for first in range(21):
            for second in range(21):
                for rate in (0, 25, 50, 75, 100):
                    subtotal = first + second
                    result = total([first, second], rate)
                    self.assertGreaterEqual(result, 0)
                    self.assertLessEqual(result, subtotal)

class MetamorphicTests(unittest.TestCase):
    def test_permutation_and_zero_item(self):
        for prices, rate in [([30, 70], 25), ([5, 10, 15], 50)]:
            transformed = list(reversed(prices)) + [0]
            self.assertEqual(total(prices, rate), total(transformed, rate))
"""

RED_SOURCE = """
import sqlite3

def total(prices, rate_percent):
    raise NotImplementedError("total is not implemented")

class DiscountRepository:
    def __init__(self, connection):
        self.connection = connection

def provider_response(order_id, prices, rate_percent):
    raise NotImplementedError("provider response is not implemented")
"""

GREEN_SOURCE = """
import json
import sqlite3

def total(prices, rate_percent):
    subtotal = sum(prices)
    return round(subtotal * (100 - rate_percent) / 100, 2)

class DiscountRepository:
    def __init__(self, connection):
        self.connection = connection
        self.connection.execute(
            "CREATE TABLE IF NOT EXISTS orders "
            "(order_id TEXT PRIMARY KEY, payload TEXT NOT NULL)"
        )

    def save(self, order_id, prices, rate_percent):
        payload = json.dumps(
            {"prices": prices, "rate_percent": rate_percent},
            sort_keys=True,
        )
        self.connection.execute(
            "INSERT INTO orders(order_id, payload) VALUES (?, ?)",
            (order_id, payload),
        )
        self.connection.commit()

    def load_total(self, order_id):
        row = self.connection.execute(
            "SELECT payload FROM orders WHERE order_id = ?",
            (order_id,),
        ).fetchone()
        if row is None:
            raise KeyError(order_id)
        payload = json.loads(row[0])
        return total(payload["prices"], payload["rate_percent"])

def provider_response(order_id, prices, rate_percent):
    return {
        "order_id": order_id,
        "total": total(prices, rate_percent),
        "currency": "JPY",
    }
"""

REFACTOR_SOURCE = """
import json
import sqlite3

def _validate(prices, rate_percent):
    # Rejecting out-of-contract values here keeps the domain function bounded.
    if any(type(price) not in (int, float) or price < 0 for price in prices):
        raise ValueError("prices must be non-negative numbers")
    if type(rate_percent) not in (int, float) or not 0 <= rate_percent <= 100:
        raise ValueError("rate_percent must be between 0 and 100")

def total(prices, rate_percent):
    _validate(prices, rate_percent)
    subtotal = sum(prices)
    return round(subtotal * (1 - rate_percent / 100), 2)

class DiscountRepository:
    def __init__(self, connection):
        self.connection = connection
        self.connection.execute(
            "CREATE TABLE IF NOT EXISTS orders "
            "(order_id TEXT PRIMARY KEY, payload TEXT NOT NULL)"
        )

    def save(self, order_id, prices, rate_percent):
        _validate(prices, rate_percent)
        payload = json.dumps(
            {"prices": list(prices), "rate_percent": rate_percent},
            sort_keys=True,
        )
        self.connection.execute(
            "INSERT INTO orders(order_id, payload) VALUES (?, ?)",
            (order_id, payload),
        )
        self.connection.commit()

    def load_total(self, order_id):
        row = self.connection.execute(
            "SELECT payload FROM orders WHERE order_id = ?",
            (order_id,),
        ).fetchone()
        if row is None:
            raise KeyError(order_id)
        payload = json.loads(row[0])
        return total(payload["prices"], payload["rate_percent"])

def provider_response(order_id, prices, rate_percent):
    return {
        "order_id": order_id,
        "total": total(prices, rate_percent),
        "currency": "JPY",
    }
"""

MUTANT_SOURCE = REFACTOR_SOURCE.replace(
    "1 - rate_percent / 100",
    "rate_percent / 100",
)
def sha256_text(text):
    return hashlib.sha256(text.encode("utf-8")).hexdigest()

def absolute_single_path(value, label):
    if (
        type(value) is not str
        or not value
        or "\0" in value
        or os.pathsep in value
    ):
        raise RuntimeError(f"{label} is invalid")
    path = Path(value)
    if not path.is_absolute():
        raise RuntimeError(f"{label} is invalid")
    return path

def leading_absolute_path(value, label):
    if type(value) is not str or not value or "\0" in value:
        raise RuntimeError(f"{label} is invalid")
    leading = value.split(os.pathsep, 1)[0]
    path = Path(leading)
    if not leading or not path.is_absolute():
        raise RuntimeError(f"{label} is invalid")
    return path

def isolated_environment(overrides=None):
    environment = {
        "PATH": os.environ.get("PATH", ""),
        "PYTHONHASHSEED": "0",
    }
    if sys.platform.startswith("linux"):
        has_runtime_root = "pythonLocation" in os.environ
        has_loader = "LD_LIBRARY_PATH" in os.environ
        if has_runtime_root and not has_loader:
            raise RuntimeError("setup-python runtime contract is incomplete")
        if has_runtime_root:
            runtime_root = absolute_single_path(
                os.environ.get("pythonLocation"),
                "setup-python runtime root",
            )
            loader_path = leading_absolute_path(
                os.environ.get("LD_LIBRARY_PATH"),
                "setup-python loader path",
            )
            executable_path = absolute_single_path(
                sys.executable,
                "Python runtime executable path",
            )
            try:
                canonical_root = runtime_root.resolve(strict=True)
                canonical_executable = executable_path.resolve(strict=True)
                runtime_library = canonical_root / "lib"
                # The three setup-python paths must describe one mounted runtime;
                # accepting any independently valid path would reintroduce an
                # ambient loader injection boundary.
                if (
                    not canonical_root.is_dir()
                    or not canonical_executable.is_file()
                    or canonical_executable.parent != canonical_root / "bin"
                    or runtime_library.is_symlink()
                    or not runtime_library.is_dir()
                ):
                    raise RuntimeError("setup-python runtime contract is invalid")
                canonical_library = runtime_library.resolve(strict=True)
                canonical_loader = loader_path.resolve(strict=True)
            except OSError as error:
                raise RuntimeError("setup-python runtime contract is invalid") from error
            if canonical_loader != canonical_library:
                raise RuntimeError("setup-python runtime contract is invalid")
            environment["LD_LIBRARY_PATH"] = str(canonical_library)
    if overrides is not None:
        allowed = {"FIXTURE_SEED", "FIXED_CLOCK_TICK", "ORDER_MODE"}
        if (
            type(overrides) is not dict
            or any(type(key) is not str for key in overrides)
            or not set(overrides) <= allowed
            or any(
                type(value) is not str or "\0" in value
                for value in overrides.values()
            )
        ):
            raise RuntimeError("subprocess environment override is invalid")
        environment.update(overrides)
    return environment

def execute(command, workspace, environment=None):
    environment = isolated_environment(environment)
    return subprocess.run(
        command,
        cwd=workspace,
        check=False,
        capture_output=True,
        text=True,
        timeout=5,
        env=environment,
    )

def execute_test(test_id, workspace):
    command = [sys.executable, "-m", "unittest", "-q", test_id]
    return command, execute(command, workspace)

def phase(name, source, workspace):
    (workspace / "order_discount.py").write_text(
        source,
        encoding="utf-8",
    )
    test_id = TEST_IDS["unit"]
    command, result = execute_test(test_id, workspace)
    behavior_command = [
        sys.executable,
        "-c",
        (
            "import json; from order_discount import total; "
            "print(json.dumps([total([100,50],10), "
            "total([30,70],25)]))"
        ),
    ]
    behavior = execute(behavior_command, workspace)
    behavior_hash = None
    if behavior.returncode == 0:
        behavior_hash = sha256_text(behavior.stdout.strip())
    elif name != "RED":
        raise RuntimeError(behavior.stderr)
    failure_reason = result.stderr if result.returncode else ""
    return {
        "name": name,
        "test_id": test_id,
        "command": command,
        "returncode": result.returncode,
        "source_sha256": sha256_text(source),
        "behavior_sha256": behavior_hash,
        "failure_reason": failure_reason,
    }

FLAKY_SOURCE = """
import json
import os
import random

seed = int(os.environ["FIXTURE_SEED"])
tick = int(os.environ["FIXED_CLOCK_TICK"])
order_mode = os.environ["ORDER_MODE"]
operations = ["reserve", "commit"]
if order_mode == "seeded":
    random.Random(seed).shuffle(operations)
elif order_mode == "reverse":
    operations.reverse()
elif order_mode != "correct":
    raise ValueError("unknown order mode")
passed = operations == ["reserve", "commit"] and tick % 2 == 0
print(json.dumps({"order": operations, "outcome": "pass" if passed else "fail"}))
raise SystemExit(0 if passed else 1)
"""

def execute_schedule(workspace, seed, tick, order_mode):
    environment = {
        "FIXTURE_SEED": str(seed),
        "FIXED_CLOCK_TICK": str(tick),
        "ORDER_MODE": order_mode,
    }
    result = execute(
        [sys.executable, "flaky_fixture.py"],
        workspace,
        environment,
    )
    observation = json.loads(result.stdout)
    return {
        "seed": seed,
        "clock_tick": tick,
        "order_mode": order_mode,
        "order": observation["order"],
        "outcome": observation["outcome"],
        "returncode": result.returncode,
    }

with TemporaryDirectory(prefix="test-strategy-lab-") as workspace_text:
    workspace = Path(workspace_text).resolve(strict=True)
    (workspace / f"{TEST_MODULE}.py").write_text(
        TEST_SOURCE,
        encoding="utf-8",
    )
    phases = [
        phase("RED", RED_SOURCE, workspace),
        phase("GREEN", GREEN_SOURCE, workspace),
        phase("REFACTOR", REFACTOR_SOURCE, workspace),
    ]
    assert [item["returncode"] for item in phases] == [1, 0, 0]
    assert "NotImplementedError: total is not implemented" in (
        phases[0]["failure_reason"]
    )

    (workspace / "order_discount.py").write_text(
        REFACTOR_SOURCE,
        encoding="utf-8",
    )
    boundaries = {
        "unit": "pure-function",
        "integration": "stdlib-sqlite-repository-adapter",
        "contract": "consumer-provider-schema-and-semantics",
        "property": "bounded-exhaustive",
        "metamorphic": "input-transformation-relation",
    }
    evidence_text = {
        "unit": "150 at 10 percent produces 135",
        "integration": "SQLite in-memory repository round trip",
        "contract": "schema types and semantic total agree",
        "property": "21x21x5 bounded input combinations",
        "metamorphic": "permutation and zero item preserve total",
    }
    strategy_evidence = []
    for kind, test_id in TEST_IDS.items():
        command, result = execute_test(test_id, workspace)
        item = {
            "kind": kind,
            "test_id": test_id,
            "command": command,
            "returncode": result.returncode,
            "passed": result.returncode == 0,
            "boundary": boundaries[kind],
            "evidence": evidence_text[kind],
        }
        if kind == "property":
            item["generation"] = "bounded-exhaustive"
            item["case_count"] = 21 * 21 * 5
        if kind == "metamorphic":
            item["relation"] = (
                "permutation-and-zero-item-preserve-total"
            )
        strategy_evidence.append(item)
    assert all(item["passed"] for item in strategy_evidence)

    (workspace / "flaky_fixture.py").write_text(
        FLAKY_SOURCE,
        encoding="utf-8",
    )
    seed_sequence = [11, 12, 13, 14]
    schedules = [
        execute_schedule(workspace, seed, seed, "seeded")
        for seed in seed_sequence
    ]
    fixed_order_variable_clock = [
        execute_schedule(workspace, 11, tick, "correct")["outcome"]
        for tick in (2, 3)
    ]
    fixed_clock_variable_order = [
        execute_schedule(workspace, 11, 2, order_mode)["outcome"]
        for order_mode in ("correct", "reverse")
    ]

    (workspace / "order_discount.py").write_text(
        MUTANT_SOURCE,
        encoding="utf-8",
    )
    mutation_test_id = TEST_IDS["integration"]
    mutation_command, mutation_result = execute_test(
        mutation_test_id,
        workspace,
    )
    assert mutation_result.returncode != 0
    report = {
        "harness": HARNESS,
        "fixture": "order-discount-v1",
        "workspace": {
            "temporary_directory_used": True,
            "all_subprocesses_used_workspace_cwd": True,
            "explicit_test_ids": list(TEST_IDS.values()),
        },
        "phases": phases,
        "strategy_evidence": strategy_evidence,
        "nondeterministic_defect": {
            "seed_sequence": seed_sequence,
            "schedules": schedules,
            "outcomes": [item["outcome"] for item in schedules],
            "isolation": {
                "fixed_order_variable_clock": (
                    fixed_order_variable_clock
                ),
                "fixed_clock_variable_order": (
                    fixed_clock_variable_order
                ),
            },
            "root_cause": "order-and-clock-coupling",
        },
        "mutation": {
            "source_sha256": sha256_text(MUTANT_SOURCE),
            "test_id": mutation_test_id,
            "command": mutation_command,
            "returncode": mutation_result.returncode,
            "failure_reason": mutation_result.stderr,
            "killed": mutation_result.returncode != 0,
        },
        "external_network_used": False,
    }
    print(json.dumps(report, ensure_ascii=False, sort_keys=True))
PY

トレードオフと失敗モード

• 誤診: TDDなら全test戦略が完成し、結合・consumer・運用testは不要である。 反証: TDDはfeedbackの進め方であり、どの境界とriskをtestするかは別の設計判断である。
• 誤診: green、coverage、contract testのどれかが高ければ製品は正しい。 反証: 誤ったoracle、未観測の認可、version差、実行されないassertionはgreenのまま残るため、mutationと異種証拠が必要になる。
• 偽陽性: 実装と無関係な共有時刻、順序、外部serviceへ依存したtestは正しい変更を止める。
• 偽陰性: mockが実契約より寛容、またはassertionがstatus codeだけなら、意味の破壊を見逃す。
• 費用: 高忠実度testほど診断が遅くなるとは限らないが、fixture維持と実行時間を測り、最小境界へ寄せる必要がある。

知識チェック

1. REDが期待どおり失敗したと判断するには、非0だけでなく何を読むべきか。
2. GREENとREFACTORでsource hashが違い、behavior hashが同じことは何を示し、何を保証しないか。
3. 割引結果を直接計算できない場合、「入力へ0円を足しても結果が変わらない」はどの技法になるか。
4. 固定seedだけでflaky testが再現しないとき、次に固定する候補を三つ挙げよ。
5. mutationが生き残った場合、test追加とmutant無効化をどう区別するか。

出典と次の学習

ISTQB CTFLはtest processと技法を、Test Driven Development: By Exampleは小さいcycleを整理する。flaky testの実証研究は再現条件の分類を、metamorphic testingの原典はoracleを直接得にくい問題への関係検証を示す。Pact Specificationはconsumer-provider間のmessage契約を扱う。

次はcore10で、実装前の脅威仮説を資産・境界・control・verificationへ接続する。90日後には、実際に逃した欠陥と遅すぎたtestを基にportfolioを更新する。

実践ラボ

割引計算をREDからmutation killまで証明する

提出成果物: RED-GREEN-REFACTOR履歴とリスク別テスト戦略

1. harness自身が作る専用一時workspaceで、明示した単体test IDをNotImplementedErrorの未実装へ実行し、期待したfailure reasonを持つREDを記録する
2. 最小実装のGREENと入力検証を分離したREFACTORへ同じ単体test IDを個別subprocessで実行し、source hashの差とbehavior hashの一致を保存する
3. 純粋関数、stdlib SQLite repository adapter、consumer-provider schemaと意味、2205件のbounded exhaustive property、入力変換metamorphic関係を別test ID・別subprocessで実行する
4. 固定seed・schedule・時計を組み合わせ、順序だけと時計だけを変える対照実験で非決定的障害のpassとfailを切り分ける
5. 割引式を意図的に変えた安全なproduction mutantをSQLite integration test IDで実行し、AssertionErrorでkillした証拠を記録する

説明して理解を確かめる

6分で、REDを先に観測する意味、coverageが正しさを保証しない理由、TDDだけでは契約・統合・運用リスクを覆えない理由を説明する。

アセスメント

1. 問い: 同じテストがCIでだけ1%失敗する。すぐretryしてgreenにする前に何を固定し、何を保存するか。

   期待する証拠: seed、時刻、実行順序、環境、入力、return codeを固定・保存し、仮説ごとの再現率を比較する計画

2. 問い: 100% statement coverageの割引機能に誤計算が残った。次にどの証拠を追加するか。

   期待する証拠: 境界値、性質、メタモルフィック関係、mutant、契約相手の期待をリスクへ対応付けた説明

別問題へ転用する

順序と時刻に依存する非決定的障害を再現し、リスク別テスト戦略で診断する

復習スケジュール

1. 1日後

   REDが実装の欠如ではなくtest自体の誤りである可能性をどう反証するか

2. 7日後

   性質testとメタモルフィックtestのoracleは何が違うか

3. 30日後

   flaky testのretryを許す前に保存すべき再現条件を四つ挙げる

4. 90日後

   REDが実装の欠如ではなくtest自体の誤りである可能性をどう反証するか

評価ルーブリック

出典

以下の外部資料は利用者が選択したときだけ開きます。

• ISTQB Certified Tester Foundation Level Syllabus v4.0.1 (standard)
• Test Driven Development: By Example (primary)
• An Empirical Analysis of Flaky Tests (peer-reviewed)
• Metamorphic Testing: A New Approach for Generating Next Test Cases (primary)
• Pact Specification (primary)