テストフレームワークの裏側 — ゼロから理解するテストランナーの設計と実装

pytest:
  test_*.py
  *_test.py
 
unittest:
  test*.py（discover のデフォルト）
 
Jest / Vitest:
  **/*.test.{ts,js}
  **/*.spec.{ts,js}
 
Go:
  *_test.go

# mini_test/discovery.py
from pathlib import Path
from fnmatch import fnmatch
 
DEFAULT_PATTERNS = ["test_*.py", "*_test.py"]
DEFAULT_IGNORE = ["__pycache__", ".venv", "venv", ".git"]
 
 
def discover_test_files(
    root_dir: str = ".",
    patterns: list[str] | None = None,
    ignore: list[str] | None = None,
) -> list[Path]:
    """glob パターンでテストファイルを再帰探索する"""
    patterns = patterns or DEFAULT_PATTERNS
    ignore = ignore or DEFAULT_IGNORE
    root = Path(root_dir)
 
    files: list[Path] = []
    for path in root.rglob("*.py"):
        # 無視すべきディレクトリをスキップ
        if any(part in ignore for part in path.parts):
            continue
        # ファイル名がいずれかのパターンにマッチするか
        if any(fnmatch(path.name, pat) for pat in patterns):
            files.append(path)
 
    return sorted(files)

# pyproject.toml の testpaths 設定がまさにこれ
[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = ["test_*.py", "*_test.py"]
python_classes = ["Test*"]
python_functions = ["test_*"]

# mini_test/collector.py
import importlib.util
import inspect
from dataclasses import dataclass, field
from typing import Callable
from pathlib import Path
 
 
@dataclass
class TestCase:
    name: str
    fn: Callable[[], None]
    module_name: str
 
 
@dataclass
class TestSuite:
    name: str
    tests: list[TestCase] = field(default_factory=list)
    setup_module: Callable[[], None] | None = None
    teardown_module: Callable[[], None] | None = None
    setup_function: Callable[[], None] | None = None
    teardown_function: Callable[[], None] | None = None
 
 
def collect_from_file(path: Path) -> TestSuite:
    """テストファイルを読み込み、test_ 関数を収集する"""
    # ファイルをモジュールとして動的にインポート
    spec = importlib.util.spec_from_file_location(path.stem, path)
    if spec is None or spec.loader is None:
        raise ImportError(f"Cannot load {path}")
 
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)  # ← モジュールのトップレベルが実行される
 
    suite = TestSuite(name=path.stem)
 
    # ライフサイクルフックの収集
    suite.setup_module = getattr(module, "setup_module", None)
    suite.teardown_module = getattr(module, "teardown_module", None)
    suite.setup_function = getattr(module, "setup_function", None)
    suite.teardown_function = getattr(module, "teardown_function", None)
 
    # test_ で始まる関数を収集
    for name, obj in inspect.getmembers(module, inspect.isfunction):
        if name.startswith("test_"):
            suite.tests.append(
                TestCase(name=name, fn=obj, module_name=path.stem)
            )
 
    return suite

# test_math.py — このファイルが exec_module() される
 
import math
 
print("この行は収集フェーズで実行される")  # モジュールレベル
 
def test_addition():
    # ← この関数は収集されるだけ。実行は後
    assert 1 + 1 == 2
 
def test_sqrt():
    # ← これも収集されるだけ
    assert math.sqrt(4) == 2.0
 
def helper_function():
    # ← test_ で始まらないので収集されない
    pass

# test_database.py — ライフサイクルフックの例
 
db = None
 
def setup_module():
    """モジュール内の全テストの前に1回だけ実行"""
    global db
    db = create_test_database()
 
def teardown_module():
    """モジュール内の全テストの後に1回だけ実行"""
    db.drop()
 
def setup_function():
    """各テスト関数の前に毎回実行"""
    db.clear_tables()
 
def teardown_function():
    """各テスト関数の後に毎回実行"""
    db.rollback()
 
def test_insert():
    db.insert({"key": "value"})
    assert db.get("key") == "value"
 
def test_delete():
    db.insert({"key": "value"})
    db.delete("key")
    assert db.get("key") is None

Suite "test_math" の実行フロー:
  ┌─ setup_module()         ← モジュール全体で1回
  │
  │  ┌─ setup_function()    ← テストごとに1回
  │  │  test_addition() 実行
  │  └─ teardown_function()
  │
  │  ┌─ setup_function()
  │  │  test_sqrt() 実行
  │  └─ teardown_function()
  │
  └─ teardown_module()      ← モジュール全体で1回

# mini_test/runner.py
import time
from dataclasses import dataclass
 
 
@dataclass
class TestResult:
    name: str
    suite_name: str
    status: str  # "passed" or "failed"
    duration: float
    error: Exception | None = None
 
 
def run_suite(suite: TestSuite) -> list[TestResult]:
    results: list[TestResult] = []
 
    # setup_module — モジュール全体の初期化
    if suite.setup_module:
        suite.setup_module()
 
    for test_case in suite.tests:
        # setup_function — 各テスト前の初期化
        if suite.setup_function:
            suite.setup_function()
 
        start = time.perf_counter()
        try:
            # テスト本体の実行
            test_case.fn()
            results.append(TestResult(
                name=test_case.name,
                suite_name=suite.name,
                status="passed",
                duration=time.perf_counter() - start,
            ))
        except Exception as exc:
            results.append(TestResult(
                name=test_case.name,
                suite_name=suite.name,
                status="failed",
                duration=time.perf_counter() - start,
                error=exc,
            ))
 
        # teardown_function — 各テスト後のクリーンアップ
        if suite.teardown_function:
            suite.teardown_function()
 
    # teardown_module — モジュール全体のクリーンアップ
    if suite.teardown_module:
        suite.teardown_module()
 
    return results

# 各テストは独立した try-except で囲まれる
try:
    test_case.fn()
    # → テストが例外を投げなければ PASS
except Exception as exc:
    # → テストが例外を投げたら FAIL（他のテストには影響しない）

# assert 文の本質
assert 1 + 1 == 2
# ↓ 以下と等価
if not (1 + 1 == 2):
    raise AssertionError

# ユーザーが書くコード
def test_greeting():
    result = greet("world")
    assert result == "Hello, world!"
 
# pytest が内部で書き換えたコード（概念）
def test_greeting():
    result = greet("world")
    _left = result
    _right = "Hello, world!"
    if not (_left == _right):
        raise AssertionError(
            f"assert {_left!r} == {_right!r}\n"
            f"  where {_left!r} = greet('world')"
        )

# mini_test/assertion_rewrite.py
import ast
import textwrap
 
 
class AssertRewriter(ast.NodeTransformer):
    """assert 文を詳細なエラーメッセージ付きに変換する"""
 
    def visit_Assert(self, node: ast.Assert) -> ast.AST:
        # assert a == b の場合のみリライト
        if isinstance(node.test, ast.Compare):
            return self._rewrite_compare(node)
        return node
 
    def _rewrite_compare(self, node: ast.Assert) -> list[ast.stmt]:
        compare = node.test
        if not isinstance(compare, ast.Compare) or len(compare.ops) != 1:
            return [node]
 
        # 左辺と右辺を一時変数に保存
        stmts: list[ast.stmt] = []
 
        # __mt_left = <left expr>
        stmts.append(ast.Assign(
            targets=[ast.Name(id="__mt_left", ctx=ast.Store())],
            value=compare.left,
            lineno=node.lineno,
            col_offset=node.col_offset,
        ))
 
        # __mt_right = <right expr>
        stmts.append(ast.Assign(
            targets=[ast.Name(id="__mt_right", ctx=ast.Store())],
            value=compare.comparators[0],
            lineno=node.lineno,
            col_offset=node.col_offset,
        ))
 
        # if not (__mt_left <op> __mt_right): raise AssertionError(...)
        op_symbols = {
            "Eq": "==", "NotEq": "!=", "Lt": "<", "LtE": "<=",
            "Gt": ">", "GtE": ">=", "Is": "is", "IsNot": "is not",
            "In": "in", "NotIn": "not in",
        }
        op_name = type(compare.ops[0]).__name__
        op_symbol = op_symbols.get(op_name, op_name)
        raise_node = ast.Raise(
            exc=ast.Call(
                func=ast.Name(id="AssertionError", ctx=ast.Load()),
                args=[ast.JoinedStr(values=[
                    ast.Constant(value="assert "),
                    ast.FormattedValue(
                        value=ast.Name(id="__mt_left", ctx=ast.Load()),
                        conversion=ord("r"),
                    ),
                    ast.Constant(value=f" {op_symbol} "),
                    ast.FormattedValue(
                        value=ast.Name(id="__mt_right", ctx=ast.Load()),
                        conversion=ord("r"),
                    ),
                ])],
                keywords=[],
            ),
        )
        stmts.append(ast.If(
            test=ast.UnaryOp(
                op=ast.Not(),
                operand=ast.Compare(
                    left=ast.Name(id="__mt_left", ctx=ast.Load()),
                    ops=compare.ops,
                    comparators=[ast.Name(id="__mt_right", ctx=ast.Load())],
                ),
            ),
            body=[raise_node],
            orelse=[],
        ))
 
        # 全ノードに行番号を付与
        for stmt in stmts:
            ast.fix_missing_locations(stmt)
        return stmts

# mini_test/assertions.py
from typing import Any
 
 
class AssertionError(Exception):
    def __init__(self, message: str, actual: Any = None, expected: Any = None):
        super().__init__(message)
        self.actual = actual
        self.expected = expected
 
 
def assert_equal(actual: Any, expected: Any) -> None:
    """== による等値比較"""
    if actual != expected:
        raise AssertionError(
            f"Expected {expected!r}, got {actual!r}",
            actual=actual,
            expected=expected,
        )
 
 
def assert_is(actual: Any, expected: Any) -> None:
    """is による同一性比較"""
    if actual is not expected:
        raise AssertionError(
            f"Expected {expected!r} (id={id(expected)}), "
            f"got {actual!r} (id={id(actual)})",
            actual=actual,
            expected=expected,
        )
 
 
def assert_true(value: Any) -> None:
    if not value:
        raise AssertionError(f"Expected truthy, got {value!r}")
 
 
def assert_raises(exc_type: type, fn: Any, *args: Any, **kwargs: Any) -> None:
    """関数が特定の例外を投げることを検証"""
    try:
        fn(*args, **kwargs)
    except exc_type:
        return  # 期待通りの例外 → PASS
    except Exception as e:
        raise AssertionError(
            f"Expected {exc_type.__name__}, got {type(e).__name__}: {e}"
        )
    raise AssertionError(
        f"Expected {exc_type.__name__} to be raised, but nothing was raised"
    )
 
 
def assert_in(item: Any, container: Any) -> None:
    if item not in container:
        raise AssertionError(
            f"Expected {item!r} to be in {container!r}"
        )
 
 
def assert_length(obj: Any, length: int) -> None:
    actual = len(obj)
    if actual != length:
        raise AssertionError(
            f"Expected length {length}, got {actual}"
        )

# == → 値の等値比較（__eq__ メソッド）
assert [1, 2] == [1, 2]     # ✅ 値が同じなら OK
assert {"a": 1} == {"a": 1} # ✅ 構造が同じなら OK
 
# is → オブジェクトの同一性比較（id が同じか）
a = [1, 2]
b = [1, 2]
assert a is a  # ✅ 同じオブジェクト
assert a is b  # ❌ 値は同じだが異なるオブジェクト
 
# 注意: 小さい整数は CPython がキャッシュする
assert 1 is 1        # ✅ CPython の実装詳細（-5〜256はキャッシュ）
assert 1000 is 1000  # ⚠️ CPython ではたまたま True になることもある

assert 1 == 2       → 1 == 2 が False → AssertionError
assert not 1 == 2   → not (1 == 2) が True → PASS
assert not 1 == 1   → not (1 == 1) が False → AssertionError

# mini_test/spy.py
from typing import Any, Callable
from dataclasses import dataclass, field
 
_UNSET = object()  # 「未設定」を表すセンチネル
 
 
@dataclass
class CallRecord:
    args: tuple
    kwargs: dict
    return_value: Any = None
    exception: Exception | None = None
 
 
class Spy:
    """関数をラップして呼び出しを記録するスパイ"""
 
    def __init__(self, original: Callable | None = None):
        self._original = original
        self._mock_return: Any = _UNSET
        self._mock_impl: Callable | None = None
        self.calls: list[CallRecord] = []
 
    def __call__(self, *args: Any, **kwargs: Any) -> Any:
        record = CallRecord(args=args, kwargs=kwargs)
        self.calls.append(record)
 
        # モックの戻り値が設定されている場合
        if self._mock_return is not _UNSET:
            record.return_value = self._mock_return
            return self._mock_return
 
        # 差し替えた実装がある場合
        impl = self._mock_impl or self._original
        if impl:
            try:
                result = impl(*args, **kwargs)
                record.return_value = result
                return result
            except Exception as exc:
                record.exception = exc
                raise
 
        return None
 
    @property
    def call_count(self) -> int:
        return len(self.calls)
 
    def mock_return_value(self, value: Any) -> "Spy":
        self._mock_return = value
        return self
 
    def mock_implementation(self, fn: Callable) -> "Spy":
        self._mock_impl = fn
        self._mock_return = _UNSET
        return self
 
    def reset(self) -> None:
        self._mock_impl = None
        self._mock_return = _UNSET
        self.calls.clear()

def test_spy_records_calls():
    add = Spy(lambda a, b: a + b)
 
    add(1, 2)
    add(3, 4)
 
    assert add.call_count == 2
    assert add.calls[0].args == (1, 2)
    assert add.calls[0].return_value == 3
 
def test_spy_mock_return():
    fetch_user = Spy()
    fetch_user.mock_return_value({"name": "Alice"})
 
    user = fetch_user()
    assert user["name"] == "Alice"

# mini_test/patch.py
from typing import Any
 
 
class Patch:
    """オブジェクトの属性を一時的に差し替えるコンテキストマネージャ"""
 
    def __init__(self, obj: Any, attr: str, replacement: Any = None):
        self.obj = obj
        self.attr = attr
        self.replacement = replacement or Spy()
        self._original: Any = None
 
    def __enter__(self) -> Any:
        # 元の属性を保存して差し替え
        self._original = getattr(self.obj, self.attr)
        setattr(self.obj, self.attr, self.replacement)
        return self.replacement
 
    def __exit__(self, *exc_info: Any) -> None:
        # 元の属性を復元
        setattr(self.obj, self.attr, self._original)

# 使用例
import os
 
def test_patch_environ():
    with Patch(os, "getcwd", Spy(lambda: "/fake/path")) as mock_cwd:
        result = os.getcwd()
        assert result == "/fake/path"
        assert mock_cwd.call_count == 1
 
    # with ブロックを抜けると自動的に復元
    assert os.getcwd() != "/fake/path"

# mini_test/reporter.py
from abc import ABC, abstractmethod
 
 
class Reporter(ABC):
    @abstractmethod
    def on_suite_start(self, suite: TestSuite) -> None: ...
 
    @abstractmethod
    def on_test_result(self, result: TestResult) -> None: ...
 
    @abstractmethod
    def on_suite_end(self, suite: TestSuite, results: list[TestResult]) -> None: ...
 
    @abstractmethod
    def on_run_end(self, all_results: list[TestResult]) -> None: ...

# mini_test/console_reporter.py
import time
 
 
class ConsoleReporter(Reporter):
    def __init__(self):
        self._start_time = 0.0
 
    def on_suite_start(self, suite: TestSuite) -> None:
        print(f"\n  {suite.name}")
        self._start_time = time.perf_counter()
 
    def on_test_result(self, result: TestResult) -> None:
        icon = "✓" if result.status == "passed" else "✗"
        color = "\033[32m" if result.status == "passed" else "\033[31m"
        reset = "\033[0m"
        duration = f"{result.duration * 1000:.1f}"
 
        print(f"    {color}{icon}{reset} {result.name} ({duration}ms)")
 
        if result.error:
            print(f"      {result.error}")
 
    def on_suite_end(self, suite: TestSuite, results: list[TestResult]) -> None:
        pass  # スイート単位のサマリは省略
 
    def on_run_end(self, all_results: list[TestResult]) -> None:
        passed = sum(1 for r in all_results if r.status == "passed")
        failed = sum(1 for r in all_results if r.status == "failed")
        total = len(all_results)
        duration = (time.perf_counter() - self._start_time) * 1000
 
        print("\n" + "─" * 40)
        if failed > 0:
            print(f"\033[31m  Tests: {passed} passed, {failed} failed, {total} total\033[0m")
        else:
            print(f"\033[32m  Tests: {passed} passed, {total} total\033[0m")
        print(f"  Time:  {duration:.0f}ms")

# mini_test/runner.py（レポーター統合版）
 
def run(suites: list[TestSuite], reporter: Reporter) -> list[TestResult]:
    all_results: list[TestResult] = []
 
    for suite in suites:
        reporter.on_suite_start(suite)
        results = run_suite(suite)
 
        for result in results:
            reporter.on_test_result(result)
 
        reporter.on_suite_end(suite, results)
        all_results.extend(results)
 
    reporter.on_run_end(all_results)
    return all_results

# mini_test/cli.py
import sys
from mini_test.discovery import discover_test_files
from mini_test.collector import collect_from_file
from mini_test.runner import run
from mini_test.console_reporter import ConsoleReporter
 
 
def main() -> None:
    # Phase 1: ディスカバリ
    files = discover_test_files()
    print(f"Found {len(files)} test file(s)\n")
 
    # Phase 2: 収集
    suites = [collect_from_file(f) for f in files]
 
    # Phase 3–6: 実行 + レポート
    reporter = ConsoleReporter()
    results = run(suites, reporter)
 
    # 終了コード
    failed = any(r.status == "failed" for r in results)
    sys.exit(1 if failed else 0)
 
 
if __name__ == "__main__":
    main()

$ python -m mini_test
 
Found 2 test file(s)
 
  test_math
    ✓ test_addition (0.3ms)
    ✓ test_division_by_zero (0.1ms)
 
  test_string
    ✓ test_trim (0.2ms)
    ✗ test_upper (0.4ms)
      AssertionError: Expected 'HELLO', got 'Hello'
 
────────────────────────────────────────
  Tests: 3 passed, 1 failed, 4 total
  Time:  12ms

並列実行のアーキテクチャ:
  メインプロセス（コーディネーター）
    │
    ├── Worker 1 (subprocess): test_math.py
    ├── Worker 2 (subprocess): test_string.py
    ├── Worker 3 (subprocess): test_api.py
    └── Worker 4 (subprocess): test_ui.py
 
  各ワーカーは独立したプロセスで、
  グローバル状態の干渉がない。
  結果はプロセス間通信でメインプロセスに集約される。

# 並列実行の概念的な実装
from multiprocessing import Pool
from pathlib import Path
 
def run_file(path: str) -> list[TestResult]:
    suite = collect_from_file(Path(path))
    return run_suite(suite)
 
def run_in_parallel(files: list[Path], num_workers: int = 4) -> list[TestResult]:
    with Pool(num_workers) as pool:
        nested = pool.map(run_file, [str(f) for f in files])
    return [r for results in nested for r in results]  # flatten

# 概念的な実装
import json
from pathlib import Path
from typing import Any
 
def assert_matches_snapshot(actual: Any, snapshot_path: Path) -> None:
    serialized = json.dumps(actual, indent=2, sort_keys=True, default=str)
 
    if snapshot_path.exists():
        stored = snapshot_path.read_text()
        if serialized != stored:
            raise AssertionError(
                f"Snapshot mismatch. Run with --update to update.\n"
                f"  Expected: {stored[:100]}...\n"
                f"  Got:      {serialized[:100]}..."
            )
    else:
        # 初回実行時: スナップショットを保存
        snapshot_path.parent.mkdir(parents=True, exist_ok=True)
        snapshot_path.write_text(serialized)

coverage.py の仕組み:
  1. sys.settrace() でトレースフックを設定
  2. Python インタプリタが各行の実行時にフックを呼ぶ
  3. フックが実行された行番号を set に記録
  4. テスト終了後、ソースファイルの全行数と照合してカバレッジを計算
 
Python 3.12+ の sys.monitoring:
  1. sys.monitoring はよりオーバーヘッドの少ない新しい API
  2. LINE イベントを監視して行の実行を記録
  3. coverage.py 7.x がこの新 API に対応

# 概念的な実装
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
 
class TestRerunHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if event.src_path.endswith(".py"):
            print(f"\nFile changed: {event.src_path}")
            run_tests()
 
def watch_mode(root_dir: str) -> None:
    print("Watching for changes...")
    observer = Observer()
    observer.schedule(TestRerunHandler(), root_dir, recursive=True)
    observer.start()
    observer.join()

pytest:
  - ディスカバリ: ファイル名 + 関数名の規約（test_ プレフィックス）
  - アサーション: 組み込み assert のリライティング（AST 変換）
  - フィクスチャ: デコレータ + 引数名による依存性注入
  - プラグイン: pluggy ベースのプラグインシステム
  - 並列化: pytest-xdist（execnet）
 
unittest:
  - ディスカバリ: TestCase サブクラスの test* メソッド
  - アサーション: assertEqual, assertTrue 等のメソッド群
  - フィクスチャ: setUp / tearDown メソッド
  - プラグイン: TestRunner のカスタマイズ
  - 並列化: 標準では未サポート
 
最大の違い:
  pytest は「assert 文をそのまま使える」仕組みを
  AST リライトで実現した。これにより
  unittest の self.assertEqual(a, b) のような
  冗長なメソッド呼び出しが不要になった。

// Go: テストは関数シグネチャの規約で識別される
func TestAdd(t *testing.T) {
    result := Add(1, 1)
    if result != 2 {
        t.Errorf("Add(1, 1) = %d, want 2", result)
    }
}
 
// サブテスト — describe/it に近い階層化
func TestMath(t *testing.T) {
    t.Run("addition", func(t *testing.T) {
        if Add(1, 1) != 2 {
            t.Fatal("1+1 should be 2")
        }
    })
    t.Run("division by zero", func(t *testing.T) {
        defer func() {
            if r := recover(); r == nil {
                t.Fatal("expected panic")
            }
        }()
        Divide(1, 0)
    })
}

Go testing の特徴:
  - テストディスカバリ: go test がコンパイル時に Test* 関数を収集
  - アサーション: 標準ライブラリにはない（if + t.Error が基本）
  - 並列化: t.Parallel() で明示的にオプトイン
  - ベンチマーク: Benchmark* 関数で統合
  - ファジング: Fuzz* 関数で統合（Go 1.18+）

Jest:
  - ディスカバリ: *.test.js / *.spec.js のファイル名パターン
  - 収集: describe/it のコールバック登録パターン
  - アサーション: expect().toBe() の Fluent API
  - モック: jest.mock() による自動モジュールモック
  - 並列化: child_process（fork）がデフォルト
 
Vitest:
  - Jest 互換の API だが、Vite のエコシステムを活用
  - ESM ネイティブ対応
  - tinypool（worker_threads ベース）で並列化

# pytest: フィクスチャは依存性注入
import pytest
 
@pytest.fixture
def database():
    """テスト用DBの作成と破棄"""
    db = create_test_db()
    yield db        # ← テストに db を注入
    db.cleanup()    # ← テスト後のクリーンアップ
 
@pytest.fixture
def user(database):
    """database フィクスチャに依存するフィクスチャ"""
    return database.create_user(name="Alice")
 
def test_insert(database):
    # 引数名 "database" がフィクスチャ名と一致 → 自動注入
    database.insert({"key": "value"})
    assert database.get("key") == "value"
 
def test_user_name(user):
    # user フィクスチャが自動注入 → 内部で database も自動解決
    assert user.name == "Alice"
 
# パラメトリックテスト
@pytest.mark.parametrize("input_val,expected", [
    (1, 1),
    (2, 4),
    (3, 9),
])
def test_square(input_val, expected):
    assert input_val ** 2 == expected

pytest のフィクスチャ解決:
  1. テスト関数の引数名を inspect.signature で取得
  2. 引数名とフィクスチャ名のマッチング
  3. 依存グラフの構築（フィクスチャが他のフィクスチャに依存可能）
  4. トポロジカルソートで実行順序を決定
  5. スコープ（function / class / module / session）に応じたキャッシュ