はじめに
「とりあえず動くスクリプト」を書くのは簡単です。しかし、機能が増えるにつれてコードは複雑になり、変更のたびに「どこか壊れていないか」という不安がつきまといます。
この記事では TDD（テスト駆動開発） と アジャイル開発 の考え方を組み合わせて、CSV解析CLIツールをゼロから作っていきます。
TDDの基本サイクル
TDDは以下の3ステップを繰り返します。
RED → テストを書く（最初は必ず失敗する）
GREEN → テストが通る最小限の実装をする
REFACTOR → コードを整理する（テストは通ったまま）
アジャイル×TDD
アジャイル開発では機能をスプリント単位で段階的に育てます。各スプリントの中でTDDのサイクルを回すことで、「動く・テストされている・整理されている」状態を常に保ちながら開発を進められます。
Sprint 1: CSVの読み込みと型変換
Sprint 2: 集計・フィルタリング・ソート
Sprint 3: CSVへの書き出し
Sprint 4: CLIツール化
完成物の全体像
# 集計
python cli.py summarize data.csv --col score --float-cols score
# フィルタリング（30歳以上）
python cli.py filter data.csv --col age --gte 30 --int-cols age --output out.csv
# ソート（降順）
python cli.py sort data.csv --col score --desc --float-cols score --output out.csv
ファイル構成
csv-analyzer/
├── csv_analyzer.py # CSVAnalyzerクラス
├── cli.py # CLIコマンド定義
├── test_csv_analyzer.py # ユニットテスト
└── test_cli.py # CLIテスト
動作環境
pip install click pytest
Sprint 1：CSVの読み込みと型変換
仕様定義
| 項目 | 内容 |
|---|---|
| 入力 | CSVファイルパス、整数変換する列名リスト、小数変換する列名リスト |
| 出力 | 辞書のリスト（各行が1つの辞書） |
| 制約 | ヘッダー行を自動認識する、数値列は明示的に型変換する |
テスト設計（RED）
まずテストを書きます。この時点では実装がないので、実行すると必ず失敗します。
テストケース一覧
| # | 観点 | 入力 | 期待結果 |
|---|---|---|---|
| L1 | 正常読み込み | 正常なCSV | 行数が正しい |
| L2 | 型変換なし | int_cols=[], float_cols=[] | 全列が文字列のまま |
| L3 | int変換 | int_cols=['age'] | age列がint型 |
| L4 | float変換 | float_cols=['score'] | score列がfloat型 |
| L5 | 空ファイル | ヘッダーのみ | data=[] |
| L6 | 異常系 | 存在しないファイル | FileNotFoundError |
# test_csv_analyzer.py
import pytest
import csv
from csv_analyzer import CSVAnalyzer
@pytest.fixture
def sample_csv(tmp_path):
"""基本的なサンプルCSVを作成する"""
filepath = tmp_path / "sample.csv"
rows = [
{"name": "Alice", "age": "32", "score": "88.5"},
{"name": "Bob", "age": "25", "score": "72.0"},
{"name": "Carol", "age": "30", "score": "95.3"},
{"name": "Dave", "age": "29", "score": "60.1"},
]
with open(filepath, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=["name", "age", "score"])
writer.writeheader()
writer.writerows(rows)
return str(filepath)
@pytest.fixture
def empty_csv(tmp_path):
"""ヘッダーのみ（データ行なし）のCSVを作成する"""
filepath = tmp_path / "empty.csv"
with open(filepath, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=["name", "age", "score"])
writer.writeheader()
return str(filepath)
class TestLoad:
def test_L1_正常読み込み_行数が正しい(self, sample_csv):
analyzer = CSVAnalyzer().load(sample_csv)
assert len(analyzer.data) == 4
def test_L2_型変換なし_全列が文字列(self, sample_csv):
analyzer = CSVAnalyzer().load(sample_csv)
assert isinstance(analyzer.data[0]["age"], str)
assert isinstance(analyzer.data[0]["score"], str)
def test_L3_int変換_age列がint型(self, sample_csv):
analyzer = CSVAnalyzer().load(sample_csv, int_cols=["age"])
assert isinstance(analyzer.data[0]["age"], int)
assert analyzer.data[0]["age"] == 32
def test_L4_float変換_score列がfloat型(self, sample_csv):
analyzer = CSVAnalyzer().load(sample_csv, float_cols=["score"])
assert isinstance(analyzer.data[0]["score"], float)
assert analyzer.data[0]["score"] == 88.5
def test_L5_空ファイル_dataが空リスト(self, empty_csv):
analyzer = CSVAnalyzer().load(empty_csv)
assert analyzer.data == []
def test_L6_存在しないファイル_例外が発生(self):
with pytest.raises(FileNotFoundError):
CSVAnalyzer().load("non_existent.csv")
この時点で pytest を実行すると…
______________ ERROR collecting test_csv_analyzer.py ______________
ImportError while importing test module '.../test_csv_analyzer.py'.
...
E ModuleNotFoundError: No module named 'csv_analyzer'
まだ実装がないので当然失敗します。これが RED の状態です。
実装（GREEN）
テストが通る最小限の実装を書きます。
# csv_analyzer.py
import csv
class CSVAnalyzer:
def __init__(self):
self.data = []
self.filepath = None
def load(self, filepath, int_cols=None, float_cols=None):
int_cols = int_cols or []
float_cols = float_cols or []
self.filepath = filepath
rows = []
with open(filepath, newline='', encoding='utf-8') as f:
reader = csv.DictReader(f)
for row in reader:
rows.append(row)
self.data = self._convert_types(rows, int_cols, float_cols)
return self
def _convert_types(self, rows, int_cols, float_cols):
converted = []
for row in rows:
new_row = dict(row) # 元データを壊さないようコピー
for col in int_cols:
new_row[col] = int(row[col])
for col in float_cols:
new_row[col] = float(row[col])
converted.append(new_row)
return converted
$ pytest test_csv_analyzer.py::TestLoad -v
test_csv_analyzer.py::TestLoad::test_L1_正常読み込み_行数が正しい PASSED
test_csv_analyzer.py::TestLoad::test_L2_型変換なし_全列が文字列 PASSED
test_csv_analyzer.py::TestLoad::test_L3_int変換_age列がint型 PASSED
test_csv_analyzer.py::TestLoad::test_L4_float変換_score列がfloat型 PASSED
test_csv_analyzer.py::TestLoad::test_L5_空ファイル_dataが空リスト PASSED
test_csv_analyzer.py::TestLoad::test_L6_存在しないファイル_例外が発生 PASSED
6 passed in 0.05s
GREEN になりました。
ポイント解説
csv.DictReader とは
1行目のヘッダーを自動でキーとして読み込み、各行を辞書として返します。
# CSVの内容
# name,age,score
# Alice,32,88.5
reader = csv.DictReader(f)
# → {'name': 'Alice', 'age': '32', 'score': '88.5'}
注意点として、すべての値が 文字列 として読み込まれます。数値計算をするには明示的な型変換が必要です。
dict(row) でコピーを作る理由
new_row = dict(row) # ← コピー
row をそのまま変更すると元データが壊れます。辞書は dict(元の辞書) でコピーできます。
return self の意味
メソッドチェーンを実現するためです。
# return self があるとこう書ける
analyzer = CSVAnalyzer().load('sample.csv', int_cols=['age'])
Sprint 2：集計・フィルタリング・ソート
仕様定義
| メソッド | 入力 | 出力・副作用 |
|---|---|---|
summarize(col) |
列名 | 合計・平均・最大・最小を表示 |
filter(col, condition) |
列名、条件関数 | 条件に合う行だけ残す |
sort(col, reverse) |
列名、昇順/降順 | 指定列でソート |
テスト設計（RED）
テストケース一覧
| # | 観点 | 入力 | 期待結果 |
|---|---|---|---|
| F1 | 境界値（以上） | age >= 30、age=29のデータ | 29は含まれない |
| F2 | 境界値（以上） | age >= 30、age=30のデータ | 30は含まれる |
| F3 | 全件一致 | 常にTrueの条件 | 全行残る |
| F4 | 全件不一致 | 常にFalseの条件 | data=[] |
| S1 | 昇順 | reverse=False | 小さい順 |
| S2 | 降順 | reverse=True | 大きい順 |
| S3 | 同値あり | 同じ値が複数 | 元の順序が保たれる（安定ソート） |
| M1 | 正常系 | 数値列 | 集計値が正しい |
| M2 | 1件のみ | 1行のデータ | 合計=平均=最大=最小 |
# test_csv_analyzer.py に追記
@pytest.fixture
def analyzer(sample_csv):
"""読み込み済みのCSVAnalyzerを返す"""
return CSVAnalyzer().load(sample_csv, int_cols=["age"], float_cols=["score"])
class TestFilter:
def test_F1_境界値_29歳は含まれない(self, analyzer):
analyzer.filter("age", lambda x: x >= 30)
names = [row["name"] for row in analyzer.data]
assert "Dave" not in names
def test_F2_境界値_30歳は含まれる(self, analyzer):
analyzer.filter("age", lambda x: x >= 30)
names = [row["name"] for row in analyzer.data]
assert "Carol" in names
def test_F3_全件一致_全行残る(self, analyzer):
original_count = len(analyzer.data)
analyzer.filter("age", lambda x: True)
assert len(analyzer.data) == original_count
def test_F4_全件不一致_dataが空(self, analyzer):
analyzer.filter("age", lambda x: False)
assert analyzer.data == []
class TestSort:
def test_S1_昇順ソート(self, analyzer):
analyzer.sort("score", reverse=False)
scores = [row["score"] for row in analyzer.data]
assert scores == sorted(scores)
def test_S2_降順ソート(self, analyzer):
analyzer.sort("score", reverse=True)
scores = [row["score"] for row in analyzer.data]
assert scores == sorted(scores, reverse=True)
def test_S3_同値あり_安定してソートされる(self, tmp_path):
filepath = tmp_path / "same_score.csv"
rows = [
{"name": "A", "score": "80.0"},
{"name": "B", "score": "80.0"},
{"name": "C", "score": "80.0"},
]
with open(filepath, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=["name", "score"])
writer.writeheader()
writer.writerows(rows)
analyzer = CSVAnalyzer().load(str(filepath), float_cols=["score"])
analyzer.sort("score")
names = [row["name"] for row in analyzer.data]
assert names == ["A", "B", "C"] # 同値のとき元の順序が保たれる
class TestSummarize:
def test_M1_集計値が正しい(self, analyzer, capsys):
analyzer.summarize("score")
output = capsys.readouterr().out
assert "件数: 4" in output
assert "合計: 315.90" in output
assert "平均: 78.98" in output # 315.9 / 4 = 78.975 → 丸めて 78.98
assert "最小: 60.10" in output
assert "最大: 95.30" in output
def test_M2_1件のみ_合計平均最大最小が等しい(self, tmp_path, capsys):
filepath = tmp_path / "one_row.csv"
with open(filepath, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=["name", "score"])
writer.writeheader()
writer.writerow({"name": "Solo", "score": "42.0"})
analyzer = CSVAnalyzer().load(str(filepath), float_cols=["score"])
analyzer.summarize("score")
output = capsys.readouterr().out
for label in ["合計", "平均", "最小", "最大"]:
assert f"{label}: 42.00" in output
この時点では filter などのメソッドがまだないので、すべて失敗します（RED）。
E AttributeError: 'CSVAnalyzer' object has no attribute 'filter'
実装（GREEN）
# csv_analyzer.py に追記
def filter(self, col, condition):
self.data = [row for row in self.data if condition(row[col])]
return self
def sort(self, col, reverse=False):
self.data = sorted(self.data, key=lambda row: row[col], reverse=reverse)
return self
def summarize(self, col):
values = [row[col] for row in self.data]
print(f"列 '{col}' の集計:")
print(f" 件数: {len(values)}")
print(f" 合計: {sum(values):.2f}")
print(f" 平均: {sum(values) / len(values):.2f}")
print(f" 最小: {min(values):.2f}")
print(f" 最大: {max(values):.2f}")
return self
$ pytest test_csv_analyzer.py::TestFilter test_csv_analyzer.py::TestSort \
test_csv_analyzer.py::TestSummarize -v
9 passed in 0.06s
表示はすべて :.2f（小数点以下2桁）にそろえています。そのまま sum(values) を表示すると、浮動小数点の誤差で 315.90000000000003 のように出てしまうためです。
ポイント解説
関数を引数として渡す（高階関数）
filter メソッドは条件を関数として受け取ります。
def filter(self, col, condition):
self.data = [row for row in self.data if condition(row[col])]
return self
呼び出し側では lambda でその場で関数を作って渡します。
# lambda x: x >= 30 は以下と同じ意味
def condition(x):
return x >= 30
こうすることで filter 自体を変更せずに、条件だけを自由に差し替えられます。
analyzer.filter("age", lambda x: x >= 30) # 30歳以上
analyzer.filter("age", lambda x: 30 <= x < 40) # 30歳以上40歳未満
analyzer.filter("name", lambda x: x.startswith("A")) # 名前がAから始まる
capsys フィクスチャ
print() の出力をキャプチャしてテストで検証できます。
def test_M1_集計値が正しい(self, analyzer, capsys):
analyzer.summarize("score")
output = capsys.readouterr().out # 標準出力を取得
assert "最大: 95.30" in output
Sprint 3：CSVへの書き出し
仕様定義
| 項目 | 内容 |
|---|---|
| 入力 | 出力ファイルパス、書き出す列名リスト（省略可） |
| 出力 | CSVファイル |
| 制約 | 列名を指定した場合はその列のみ書き出す、dataが空の場合はファイルを作らない |
テスト設計（RED）
| # | 観点 | 入力 | 期待結果 |
|---|---|---|---|
| W1 | 正常系 | 通常データ | ファイルが存在し内容が正しい |
| W2 | 列指定あり | fieldnames=['name'] | 指定列のみ出力 |
| W3 | 空データ | data=[] | ファイルを作らず終了 |
# test_csv_analyzer.py の先頭の import に追加
import os
# test_csv_analyzer.py に追記
class TestSave:
def test_W1_正常系_ファイルが保存される(self, analyzer, tmp_path):
output_path = str(tmp_path / "output.csv")
analyzer.save(output_path)
assert os.path.exists(output_path)
with open(output_path, newline="", encoding="utf-8") as f:
rows = list(csv.DictReader(f))
assert len(rows) == 4
assert rows[0]["name"] == "Alice"
def test_W2_列指定あり_指定列のみ出力(self, analyzer, tmp_path):
output_path = str(tmp_path / "output.csv")
analyzer.save(output_path, fieldnames=["name", "age"])
with open(output_path, newline="", encoding="utf-8") as f:
rows = list(csv.DictReader(f))
assert "score" not in rows[0]
assert "name" in rows[0]
def test_W3_空データ_ファイルを作らない(self, analyzer, tmp_path):
analyzer.filter("age", lambda x: False) # 全件除外
output_path = str(tmp_path / "output.csv")
analyzer.save(output_path)
assert not os.path.exists(output_path)
save はまだないので、ここでも最初は失敗します（RED）。
E AttributeError: 'CSVAnalyzer' object has no attribute 'save'
実装（GREEN）→ バグ発見
# csv_analyzer.py に追記（最初の実装）
def save(self, filepath, fieldnames=None):
if not self.data:
print("データが空です")
return self
if fieldnames is None:
fieldnames = list(self.data[0].keys())
with open(filepath, 'w', newline='', encoding='utf-8') as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(self.data) # ← ここにバグ
return self
$ pytest test_csv_analyzer.py::TestSave -v
PASSED test_W1_正常系_ファイルが保存される
FAILED test_W2_列指定あり_指定列のみ出力 ← バグ発見！
PASSED test_W3_空データ_ファイルを作らない
テストが バグを発見しました。W2 の失敗内容を見ると：
ValueError: dict contains fields not in fieldnames: 'score'
fieldnames=['name', 'age'] を指定しても、辞書には score が残ったままなので DictWriter がエラーを出します。テストを書いていなければリリース後に発覚していたバグです。
バグ修正（GREEN）
失敗したテストを通すための修正なので、これは GREEN の作業です。「リファクタリング」はテストが通った状態のまま振る舞いを変えずに構造を整えることを指すので、バグ修正とは区別します（リファクタリングは Sprint 4 で行います）。
def save(self, filepath, fieldnames=None):
if not self.data:
print("データが空です")
return self
if fieldnames is None:
fieldnames = list(self.data[0].keys())
with open(filepath, 'w', newline='', encoding='utf-8') as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
# 指定された列だけ抽出してから書き出す
filtered = [{k: row[k] for k in fieldnames} for row in self.data]
writer.writerows(filtered)
return self
$ pytest test_csv_analyzer.py -v
18 passed in 0.07s ✅
csv_analyzer.py 完成形
ここまでの実装をまとめたものです。CLIから使ったときに処理状況がわかるよう、load と save に件数を表示する print を追加しています。
csv_analyzer.py 全体（クリックで展開）
import csv
class CSVAnalyzer:
def __init__(self):
self.data = []
self.filepath = None
def load(self, filepath, int_cols=None, float_cols=None):
int_cols = int_cols or []
float_cols = float_cols or []
self.filepath = filepath
rows = []
with open(filepath, newline='', encoding='utf-8') as f:
reader = csv.DictReader(f)
for row in reader:
rows.append(row)
self.data = self._convert_types(rows, int_cols, float_cols)
print(f"{len(self.data)}行を読み込みました")
return self
def _convert_types(self, rows, int_cols, float_cols):
converted = []
for row in rows:
new_row = dict(row)
for col in int_cols:
new_row[col] = int(row[col])
for col in float_cols:
new_row[col] = float(row[col])
converted.append(new_row)
return converted
def filter(self, col, condition):
self.data = [row for row in self.data if condition(row[col])]
return self
def sort(self, col, reverse=False):
self.data = sorted(self.data, key=lambda row: row[col], reverse=reverse)
return self
def summarize(self, col):
values = [row[col] for row in self.data]
print(f"列 '{col}' の集計:")
print(f" 件数: {len(values)}")
print(f" 合計: {sum(values):.2f}")
print(f" 平均: {sum(values) / len(values):.2f}")
print(f" 最小: {min(values):.2f}")
print(f" 最大: {max(values):.2f}")
return self
def save(self, filepath, fieldnames=None):
if not self.data:
print("データが空です")
return self
if fieldnames is None:
fieldnames = list(self.data[0].keys())
with open(filepath, 'w', newline='', encoding='utf-8') as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
filtered = [{k: row[k] for k in fieldnames} for row in self.data]
writer.writerows(filtered)
print(f"{len(self.data)}行を '{filepath}' に保存しました")
return self
Sprint 4：CLIツール化
仕様定義（コマンド体系）
python cli.py [コマンド] [オプション]
| コマンド | 必須オプション | 任意オプション |
|---|---|---|
summarize |
--col |
--int-cols, --float-cols
|
filter |
--col, --gte/--gt/--lt/--lte/--eqのいずれか |
--int-cols, --float-cols, --output
|
sort |
--col |
--desc, --int-cols, --float-cols, --output
|
テスト設計（RED）
CLIのテストには click が提供する CliRunner を使います。ターミナルを使わずPythonコード内でコマンドを呼び出して結果を検証できます。
| # | コマンド | 観点 | 期待結果 |
|---|---|---|---|
| CS1 | summarize | 正常系 | exit_code=0、集計値が出力に含まれる |
| CS2 | summarize |
--col 省略 |
exit_code≠0 |
| CF1 | filter |
--gte 条件 |
該当行のみ保存 |
| CF2 | filter | 複数条件AND | 両条件を満たす行のみ保存 |
| CF3 | filter | 条件なし | exit_code≠0、ファイル未生成 |
| CF4 | filter | 小数列に --eq
|
72 で 72.0 の行に一致する |
| CK1 | sort | 昇順 | 出力ファイルが昇順 |
| CK2 | sort |
--desc 降順 |
出力ファイルが降順 |
# test_cli.py
import pytest
import csv
import os
from click.testing import CliRunner
from cli import cli
@pytest.fixture
def runner():
return CliRunner()
@pytest.fixture
def sample_csv(tmp_path):
"""test_csv_analyzer.py と同じデータに Eve を1行追加したもの"""
filepath = tmp_path / "sample.csv"
filepath.write_text(
"name,age,score\n"
"Alice,32,88.5\nBob,25,72.0\nCarol,30,95.3\nDave,29,60.1\nEve,35,91.2\n",
encoding="utf-8",
)
return str(filepath)
def read_csv(filepath):
with open(filepath, newline="", encoding="utf-8") as f:
return list(csv.DictReader(f))
class TestSummarize:
def test_CS1_正常系_集計値が出力される(self, runner, sample_csv):
result = runner.invoke(cli, [
"summarize", sample_csv,
"--col", "score",
"--float-cols", "score",
])
assert result.exit_code == 0
assert "件数: 5" in result.output
assert "最大: 95.30" in result.output
assert "最小: 60.10" in result.output
def test_CS2_異常系_col省略でエラー(self, runner, sample_csv):
result = runner.invoke(cli, ["summarize", sample_csv])
assert result.exit_code != 0
class TestFilter:
def test_CF1_正常系_gte条件で絞り込まれる(self, runner, sample_csv, tmp_path):
output_path = str(tmp_path / "out.csv")
result = runner.invoke(cli, [
"filter", sample_csv,
"--col", "age", "--gte", "30",
"--int-cols", "age",
"--output", output_path,
])
assert result.exit_code == 0
names = [row["name"] for row in read_csv(output_path)]
assert names == ["Alice", "Carol", "Eve"]
def test_CF2_正常系_複数条件ANDで絞り込まれる(self, runner, sample_csv, tmp_path):
output_path = str(tmp_path / "out.csv")
result = runner.invoke(cli, [
"filter", sample_csv,
"--col", "age", "--gte", "30", "--lt", "35",
"--int-cols", "age",
"--output", output_path,
])
assert result.exit_code == 0
names = [row["name"] for row in read_csv(output_path)]
assert names == ["Alice", "Carol"]
def test_CF3_異常系_条件なしでエラー(self, runner, sample_csv, tmp_path):
output_path = str(tmp_path / "out.csv")
result = runner.invoke(cli, [
"filter", sample_csv, "--col", "age", "--output", output_path,
])
assert result.exit_code != 0
assert "フィルタ条件" in result.output
assert not os.path.exists(output_path)
def test_CF4_正常系_小数列をeqで絞り込める(self, runner, sample_csv, tmp_path):
output_path = str(tmp_path / "out.csv")
result = runner.invoke(cli, [
"filter", sample_csv,
"--col", "score", "--eq", "72",
"--float-cols", "score",
"--output", output_path,
])
assert result.exit_code == 0
names = [row["name"] for row in read_csv(output_path)]
assert names == ["Bob"]
class TestSort:
def test_CK1_昇順ソートで保存される(self, runner, sample_csv, tmp_path):
output_path = str(tmp_path / "out.csv")
result = runner.invoke(cli, [
"sort", sample_csv,
"--col", "score", "--float-cols", "score",
"--output", output_path,
])
assert result.exit_code == 0
scores = [float(row["score"]) for row in read_csv(output_path)]
assert scores == sorted(scores)
def test_CK2_降順ソートで保存される(self, runner, sample_csv, tmp_path):
output_path = str(tmp_path / "out.csv")
result = runner.invoke(cli, [
"sort", sample_csv,
"--col", "score", "--desc", "--float-cols", "score",
"--output", output_path,
])
assert result.exit_code == 0
scores = [float(row["score"]) for row in read_csv(output_path)]
assert scores == sorted(scores, reverse=True)
CF1・CF2 では「条件を満たすか」だけでなく、残る行の名前そのものを検証しています。all(age >= 30 for age in ages) のような書き方だと、1行も残らなかった場合（空リスト）でも True になってしまい、バグを見逃すからです。
cli.py はまだないので、実行すると収集の段階で失敗します（RED）。
E ModuleNotFoundError: No module named 'cli'
実装（GREEN）
# cli.py
import click
from csv_analyzer import CSVAnalyzer
@click.group()
def cli():
"""CSVファイルを解析するCLIツール"""
pass
@cli.command()
@click.argument("filepath")
@click.option("--col", required=True, help="集計する列名")
@click.option("--int-cols", default="", help="整数変換する列名（カンマ区切り）")
@click.option("--float-cols", default="", help="小数変換する列名（カンマ区切り）")
def summarize(filepath, col, int_cols, float_cols):
"""列の集計（合計・平均・最大・最小）を表示する"""
int_list = [c for c in int_cols.split(",") if c]
float_list = [c for c in float_cols.split(",") if c]
CSVAnalyzer().load(filepath, int_cols=int_list, float_cols=float_list).summarize(col)
@cli.command("filter")
@click.argument("filepath")
@click.option("--col", required=True, help="フィルタする列名")
@click.option("--gt", type=float, default=None, help="この値より大きい（>）")
@click.option("--gte", type=float, default=None, help="この値以上（>=）")
@click.option("--lt", type=float, default=None, help="この値より小さい（<）")
@click.option("--lte", type=float, default=None, help="この値以下（<=）")
@click.option("--eq", default=None, help="この値と等しい（==）")
@click.option("--int-cols", default="", help="整数変換する列名（カンマ区切り）")
@click.option("--float-cols", default="", help="小数変換する列名（カンマ区切り）")
@click.option("--output", default="output.csv", help="出力ファイルパス")
def filter_rows(filepath, col, gt, gte, lt, lte, eq, int_cols, float_cols, output):
"""条件で行を絞り込んで保存する"""
int_list = [c for c in int_cols.split(",") if c]
float_list = [c for c in float_cols.split(",") if c]
conditions = []
if gt is not None: conditions.append(lambda x: x > gt)
if gte is not None: conditions.append(lambda x: x >= gte)
if lt is not None: conditions.append(lambda x: x < lt)
if lte is not None: conditions.append(lambda x: x <= lte)
if eq is not None: conditions.append(lambda x: equals(x, eq))
if not conditions:
raise click.UsageError("フィルタ条件を最低1つ指定してください（--gt, --gte, --lt, --lte, --eq）")
def combined(x):
return all(cond(x) for cond in conditions)
(CSVAnalyzer()
.load(filepath, int_cols=int_list, float_cols=float_list)
.filter(col, combined)
.save(output))
def equals(x, value):
"""数値列は数値として、文字列列は文字列として比較する"""
if isinstance(x, (int, float)):
return x == float(value)
return x == value
@cli.command("sort")
@click.argument("filepath")
@click.option("--col", required=True, help="ソートする列名")
@click.option("--desc", is_flag=True, help="降順にする")
@click.option("--int-cols", default="", help="整数変換する列名（カンマ区切り）")
@click.option("--float-cols", default="", help="小数変換する列名（カンマ区切り）")
@click.option("--output", default="output.csv", help="出力ファイルパス")
def sort_rows(filepath, col, desc, int_cols, float_cols, output):
"""列でソートして保存する"""
int_list = [c for c in int_cols.split(",") if c]
float_list = [c for c in float_cols.split(",") if c]
(CSVAnalyzer()
.load(filepath, int_cols=int_list, float_cols=float_list)
.sort(col, reverse=desc)
.save(output))
if __name__ == "__main__":
cli()
$ pytest test_cli.py -v
8 passed in 0.10s
GREEN になりました。
関数名を filter_rows / sort_rows にして、コマンド名は @cli.command("filter") のように明示しています。関数名をそのまま filter / sort にすると、Python 組み込みの filter() を上書きしてしまうためです。
--eq は equals() で比較します。単純に str(x) == value と比べると、小数列の 72.0 に対して --eq 72 が一致しません（CF4 で検証）。
リファクタリング（REFACTOR）
テストが通ったので、振る舞いを変えずにコードを整理します。気になるのは次の2点です。
-
int_list/float_listを作る処理が3つのコマンドに重複している -
filter_rowsの条件組み立てがifの羅列になっている
# cli.py（リファクタリング後の変更箇所）
import operator
import click
from csv_analyzer import CSVAnalyzer
def parse_cols(value):
"""カンマ区切りの列名をリストにする（空文字は除く）"""
return [c for c in value.split(",") if c]
def load(filepath, int_cols, float_cols):
return CSVAnalyzer().load(
filepath, int_cols=parse_cols(int_cols), float_cols=parse_cols(float_cols)
)
def build_conditions(gt, gte, lt, lte):
"""指定された比較条件を関数のリストにする"""
conditions = []
for op, value in [(operator.gt, gt), (operator.ge, gte),
(operator.lt, lt), (operator.le, lte)]:
if value is not None:
conditions.append(lambda x, op=op, value=value: op(x, value))
return conditions
# summarize
load(filepath, int_cols, float_cols).summarize(col)
# filter_rows
conditions = build_conditions(gt, gte, lt, lte)
if eq is not None:
conditions.append(lambda x: equals(x, eq))
...
load(filepath, int_cols, float_cols).filter(col, combined).save(output)
# sort_rows
load(filepath, int_cols, float_cols).sort(col, reverse=desc).save(output)
$ pytest test_cli.py -v
8 passed in 0.10s
リファクタリング後もテストはすべて通ったままです。テストがあるから安心して構造を変えられるのが、TDDでいちばん効いてくるポイントです。
ポイント解説
CliRunner の使い方
ターミナルで打つコマンドをそのままリストで渡します。
# ターミナルで打つ場合
# python cli.py summarize sample.csv --col score
# CliRunnerで書く場合
result = runner.invoke(cli, ["summarize", "sample.csv", "--col", "score"])
返ってくる result から検証できる主な情報です。
| プロパティ | 内容 |
|---|---|
result.exit_code |
終了コード（0=正常、それ以外=エラー） |
result.output |
標準出力に出たテキスト |
result.exception |
例外が発生した場合その内容 |
ループ内 lambda の遅延束縛トラップ
リファクタリングで条件をループで組み立てるようにしたとき、lambda に op=op, value=value を付けていました。これを付けずに次のように書くと、テストが失敗します。
# NG：ループ内の lambda は変数そのものを参照する
for op, value in [(operator.gt, gt), (operator.ge, gte),
(operator.lt, lt), (operator.le, lte)]:
if value is not None:
conditions.append(lambda x: op(x, value))
lambda の中の op と value は、作った時点の値ではなく、呼び出した時点の値を参照します（遅延束縛）。呼び出されるのはループが終わった後なので、すべての条件がループ最後の値（operator.le, lte）を見てしまいます。--gte 30 だけを指定しても、実際には x <= None という比較になって TypeError で落ちます。
# OK：デフォルト引数でその時点の値をキャプチャする
conditions.append(lambda x, op=op, value=value: op(x, value))
デフォルト引数は lambda を定義した時点で評価されるので、各ループの値が固定されます。NG版で書いてしまっても CF1・CF2 が失敗して誤りを教えてくれるので、テストがリファクタリングの安全網になります。
なお、GREEN 版の lambda x: x > gt はループではなく、gt などの変数も後から書き換わらないので、この問題は起きません。
CLIテストとユニットテストの役割分担
test_csv_analyzer.py |
test_cli.py |
|
|---|---|---|
| 対象 |
CSVAnalyzer クラス |
CLIコマンド |
| 観点 | ロジックの正しさ | 引数の受け渡し・終了コード |
| 細かさ | 境界値・異常系を細かく | コマンドが正しく動くかを大まかに |
CLIテストはロジックの細部には踏み込まず、「コマンドとして正しく動くか」だけを見ます。細かいロジックはユニットテストに任せるという役割分担です。
全テスト実行
$ pytest -v
...
test_cli.py::TestFilter::test_CF4_正常系_小数列をeqで絞り込める PASSED
test_cli.py::TestSort::test_CK1_昇順ソートで保存される PASSED
test_cli.py::TestSort::test_CK2_降順ソートで保存される PASSED
26 passed in 0.12s ✅
ユニットテスト18件とCLIテスト8件、合わせて26件がすべて通りました。
テストを tests/ ディレクトリに分けると、pytest 実行時に ModuleNotFoundError: No module named 'cli' が出ることがあります。その場合は pyproject.toml に [tool.pytest.ini_options] と pythonpath = ["."] を書くと解決します（pytest 7 以降）。
おわりに
各スプリントのまとめ
| Sprint | 追加機能 | テスト数 | 発見したバグ |
|---|---|---|---|
| 1 | 読み込み・型変換 | 6件 | — |
| 2 | 集計・フィルタ・ソート | 9件 | — |
| 3 | 書き出し | 3件 | W2（列指定時のValueError） |
| 4 | CLIツール化 | 8件 | — |
| 合計 | 26件 | 1件 |
TDD × アジャイルで開発してみた所感
よかった点
- 各スプリント終了時点で「動く・テストされている」状態が保たれる
- Sprint 3でW2のバグをリリース前に発見できた
- Sprint 4のリファクタリングを、テストを安全網にして安心して進められた
- テストケース一覧が仕様書の役割を果たす
気をつける点
- テストを書く時間が必要なため、初速は遅く感じる
- テスト自体にバグが入ることもある。本記事でも test_M1 の平均値を、最初は手計算で
78.97と書いていました。しかし315.9 / 4は浮動小数点では78.97500000000001になり、:.2fで丸めると78.98になります。期待値は頭の中の計算だけで決めず、実際に実行して確かめる必要があります - アサーションが弱いと、テストは通るのにバグを見逃すことがある（空リストに対する
all()がTrueになる、同値だけのデータでは並び順を検証できない、など）
次のステップ
この記事で作ったツールへの機能追加として、以下が考えられます。
- グループ集計（都市ごとの平均スコアなど）
- エラー処理の強化（存在しない列名を指定したときのわかりやすいメッセージ、0件のデータに
summarizeを実行したときのZeroDivisionError対策） -
pandasを使ったリファクタリング