第157回　CSV品質ゲートを回帰テストする：unittestで終了コードとJSON出力を検証する

第156回では、更新前後のCSVを突合し、許可されていない変更を止める品質ゲートを作りました。しかし、品質ゲート自体を修正したときに判定が変わると、正しい更新を止めたり、契約違反の更新を通したりするおそれがあります。

今回はPython標準ライブラリだけで小さな品質ゲートを実装し、そのCLIをunittestから実行します。比較関数だけでなく、CSV入力、1行のJSON出力、標準エラー出力、終了コードまでテストします。コードはPython 3.10以上を前提とします。

今回テストする判定契約

• 更新前後でヘッダーの名前と順序が一致すること
• 主キーの集合が一致し、空欄や重複がないこと
• --allowで指定した列だけが変更されていること
• --expected-changesを指定した場合は、変更されたレコード数が一致すること
• 処理結果を1行のJSONとして標準出力へ返し、通常時は標準エラー出力を使わないこと

終了コード	意味	後続処理
0	品質ゲートを通過	次の処理へ進める
1	CSVまたは変更内容が契約違反	更新を止めて内容を確認する
2	引数またはファイル入出力の問題	実行設定やパスを確認する

手順1：品質ゲートを実装する

次のコードをcsv_gate.pyとして保存します。UTF-8およびUTF-8 BOM付きのカンマ区切りCSVを読み込み、各フィールドを文字列のまま比較します。

from __future__ import annotations

import argparse
import csv
import json
from collections import Counter
from dataclasses import asdict, dataclass
from pathlib import Path


class GateError(Exception):
    def __init__(self, code: str, **details: object) -> None:
        super().__init__(code)
        self.code = code
        self.details = details

    def payload(self) -> dict[str, object]:
        return {
            "ok": False,
            "error": {"code": self.code, **self.details},
        }


@dataclass(frozen=True)
class GateResult:
    ok: bool
    row_count: int
    changed_keys: list[str]
    changed_columns: list[str]


def emit(payload: dict[str, object]) -> None:
    print(json.dumps(payload, ensure_ascii=False, sort_keys=True))


class JsonArgumentParser(argparse.ArgumentParser):
    def error(self, message: str) -> None:
        emit({
            "ok": False,
            "error": {"code": "usage_error", "message": message},
        })
        raise SystemExit(2)


def non_negative_int(value: str) -> int:
    try:
        number = int(value)
    except ValueError as exc:
        raise argparse.ArgumentTypeError(
            "0以上の整数を指定してください"
        ) from exc
    if number < 0:
        raise argparse.ArgumentTypeError(
            "0以上の整数を指定してください"
        )
    return number


def read_indexed_csv(
    path: Path,
    key_column: str,
) -> tuple[list[str], dict[str, dict[str, str]]]:
    reader = None
    try:
        with path.open("r", encoding="utf-8-sig", newline="") as handle:
            reader = csv.DictReader(handle, strict=True)
            headers = reader.fieldnames

            if not headers:
                raise GateError("missing_header", path=str(path))

            blank_positions = [
                index + 1
                for index, name in enumerate(headers)
                if not name.strip()
            ]
            if blank_positions:
                raise GateError(
                    "blank_header",
                    path=str(path),
                    positions=blank_positions,
                )

            duplicates = sorted(
                name
                for name, count in Counter(headers).items()
                if count > 1
            )
            if duplicates:
                raise GateError(
                    "duplicate_header",
                    path=str(path),
                    columns=duplicates,
                )

            if key_column not in headers:
                raise GateError(
                    "missing_key_column",
                    path=str(path),
                    column=key_column,
                )

            rows: dict[str, dict[str, str]] = {}
            for row in reader:
                if None in row or any(
                    value is None for value in row.values()
                ):
                    raise GateError(
                        "row_width_mismatch",
                        path=str(path),
                        line=reader.line_num,
                    )

                record_id = row[key_column]
                if not record_id.strip():
                    raise GateError(
                        "blank_key",
                        path=str(path),
                        line=reader.line_num,
                    )
                if record_id in rows:
                    raise GateError(
                        "duplicate_key",
                        path=str(path),
                        key=record_id,
                    )
                rows[record_id] = row

            return headers, rows

    except UnicodeDecodeError as exc:
        raise GateError("decode_error", path=str(path)) from exc
    except csv.Error as exc:
        line = reader.line_num if reader is not None else None
        raise GateError(
            "csv_parse_error",
            path=str(path),
            line=line,
            message=str(exc),
        ) from exc


def compare_csv(
    before_path: Path,
    after_path: Path,
    key_column: str,
    allowed_columns: list[str],
    expected_changes: int | None,
) -> GateResult:
    allowed = set(allowed_columns)
    if key_column in allowed:
        raise GateError(
            "key_column_cannot_be_allowed",
            column=key_column,
        )

    before_headers, before_rows = read_indexed_csv(
        before_path,
        key_column,
    )
    after_headers, after_rows = read_indexed_csv(
        after_path,
        key_column,
    )

    if before_headers != after_headers:
        raise GateError(
            "header_mismatch",
            before=before_headers,
            after=after_headers,
        )

    unknown_allowed = sorted(allowed - set(before_headers))
    if unknown_allowed:
        raise GateError(
            "unknown_allowed_columns",
            columns=unknown_allowed,
        )

    before_keys = set(before_rows)
    after_keys = set(after_rows)
    if before_keys != after_keys:
        raise GateError(
            "key_set_mismatch",
            added=sorted(after_keys - before_keys),
            removed=sorted(before_keys - after_keys),
        )

    changed_keys: list[str] = []
    changed_columns: set[str] = set()

    for record_id in sorted(before_rows):
        differing = [
            column
            for column in before_headers
            if column != key_column
            and before_rows[record_id][column]
            != after_rows[record_id][column]
        ]
        forbidden = sorted(set(differing) - allowed)
        if forbidden:
            raise GateError(
                "forbidden_change",
                key=record_id,
                columns=forbidden,
            )
        if differing:
            changed_keys.append(record_id)
            changed_columns.update(differing)

    if (
        expected_changes is not None
        and len(changed_keys) != expected_changes
    ):
        raise GateError(
            "unexpected_change_count",
            expected=expected_changes,
            actual=len(changed_keys),
        )

    return GateResult(
        ok=True,
        row_count=len(before_rows),
        changed_keys=changed_keys,
        changed_columns=sorted(changed_columns),
    )


def build_parser() -> argparse.ArgumentParser:
    parser = JsonArgumentParser()
    parser.add_argument("--before", required=True, type=Path)
    parser.add_argument("--after", required=True, type=Path)
    parser.add_argument("--key", required=True)
    parser.add_argument(
        "--allow",
        action="append",
        default=[],
        metavar="COLUMN",
    )
    parser.add_argument(
        "--expected-changes",
        type=non_negative_int,
    )
    return parser


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    try:
        result = compare_csv(
            before_path=args.before,
            after_path=args.after,
            key_column=args.key,
            allowed_columns=args.allow,
            expected_changes=args.expected_changes,
        )
    except GateError as exc:
        emit(exc.payload())
        return 1
    except OSError as exc:
        emit({
            "ok": False,
            "error": {
                "code": "io_error",
                "message": str(exc),
            },
        })
        return 2

    emit(asdict(result))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

CSVの形式や変更内容に関する不一致はGateErrorへ変換し、終了コード1で返します。ファイルが存在しない場合などのOSErrorは終了コード2です。引数の不足や不正な値も、JsonArgumentParserがJSONを出力して終了コード2にします。

想定外の不具合をexcept Exceptionで隠してはいません。未分類の例外は、実装を修正すべき問題としてテスト結果や実行ログに残します。

手順2：CLI全体を回帰テストする

次のコードをcsv_gate.pyと同じディレクトリへtest_csv_gate.pyとして保存します。各テストは一時ディレクトリ内にCSVを作るため、既存の業務ファイルには触れません。

from __future__ import annotations

import csv
import json
import os
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path


SCRIPT = Path(__file__).with_name("csv_gate.py").resolve()
FIELDS = ["id", "status", "note"]


class GateCliTests(unittest.TestCase):
    def setUp(self) -> None:
        self.temp_dir = tempfile.TemporaryDirectory()
        self.root = Path(self.temp_dir.name)
        self.before_rows = [
            {
                "id": "A001",
                "status": "pending",
                "note": "call, email",
            },
            {
                "id": "A002",
                "status": "pending",
                "note": "",
            },
        ]

    def tearDown(self) -> None:
        self.temp_dir.cleanup()

    def write_csv(
        self,
        name: str,
        rows: list[dict[str, str]],
        fieldnames: list[str] = FIELDS,
    ) -> Path:
        path = self.root / name
        with path.open(
            "w",
            encoding="utf-8",
            newline="",
        ) as handle:
            writer = csv.DictWriter(
                handle,
                fieldnames=fieldnames,
                lineterminator=chr(10),
            )
            writer.writeheader()
            writer.writerows(rows)
        return path

    def run_gate(
        self,
        before: Path,
        after: Path,
        allowed: tuple[str, ...] = ("status",),
        expected: int | None = None,
    ) -> tuple[
        subprocess.CompletedProcess[str],
        dict[str, object],
    ]:
        command = [
            sys.executable,
            str(SCRIPT),
            "--before",
            str(before),
            "--after",
            str(after),
            "--key",
            "id",
        ]
        for column in allowed:
            command.extend(["--allow", column])
        if expected is not None:
            command.extend([
                "--expected-changes",
                str(expected),
            ])

        child_env = os.environ.copy()
        child_env["PYTHONIOENCODING"] = "utf-8"

        completed = subprocess.run(
            command,
            capture_output=True,
            text=True,
            encoding="utf-8",
            check=False,
            timeout=10,
            env=child_env,
        )

        self.assertEqual(completed.stderr, "", completed.stderr)
        stdout_lines = completed.stdout.splitlines()
        self.assertEqual(
            len(stdout_lines),
            1,
            completed.stdout,
        )
        return completed, json.loads(stdout_lines[0])

    def test_allowed_change_passes(self) -> None:
        before = self.write_csv(
            "before.csv",
            self.before_rows,
        )
        after = self.write_csv("after.csv", [
            {
                "id": "A001",
                "status": "approved",
                "note": "call, email",
            },
            {
                "id": "A002",
                "status": "pending",
                "note": "",
            },
        ])

        completed, payload = self.run_gate(
            before,
            after,
            expected=1,
        )

        self.assertEqual(completed.returncode, 0)
        self.assertEqual(payload, {
            "ok": True,
            "row_count": 2,
            "changed_keys": ["A001"],
            "changed_columns": ["status"],
        })

    def test_header_mismatch_is_rejected(self) -> None:
        before = self.write_csv(
            "header_before.csv",
            self.before_rows,
        )
        after = self.write_csv(
            "header_after.csv",
            self.before_rows,
            fieldnames=["id", "note", "status"],
        )

        completed, payload = self.run_gate(before, after)

        self.assertEqual(completed.returncode, 1)
        self.assertEqual(
            payload["error"]["code"],
            "header_mismatch",
        )

    def test_rejected_cases(self) -> None:
        cases = [
            (
                "forbidden",
                [
                    {
                        "id": "A001",
                        "status": "approved",
                        "note": "call only",
                    },
                    {
                        "id": "A002",
                        "status": "pending",
                        "note": "",
                    },
                ],
                "forbidden_change",
            ),
            (
                "duplicate",
                [
                    {
                        "id": "A001",
                        "status": "approved",
                        "note": "call, email",
                    },
                    {
                        "id": "A001",
                        "status": "pending",
                        "note": "",
                    },
                ],
                "duplicate_key",
            ),
            (
                "blank",
                [
                    {
                        "id": " ",
                        "status": "approved",
                        "note": "call, email",
                    },
                    {
                        "id": "A002",
                        "status": "pending",
                        "note": "",
                    },
                ],
                "blank_key",
            ),
            (
                "removed",
                [
                    {
                        "id": "A001",
                        "status": "approved",
                        "note": "call, email",
                    },
                ],
                "key_set_mismatch",
            ),
        ]

        for name, after_rows, expected_code in cases:
            with self.subTest(name=name):
                before = self.write_csv(
                    f"{name}_before.csv",
                    self.before_rows,
                )
                after = self.write_csv(
                    f"{name}_after.csv",
                    after_rows,
                )

                completed, payload = self.run_gate(
                    before,
                    after,
                )

                self.assertEqual(completed.returncode, 1)
                self.assertEqual(
                    payload["error"]["code"],
                    expected_code,
                )

    def test_wrong_expected_change_count_is_rejected(
        self,
    ) -> None:
        before = self.write_csv(
            "count_before.csv",
            self.before_rows,
        )
        after = self.write_csv("count_after.csv", [
            {
                "id": "A001",
                "status": "approved",
                "note": "call, email",
            },
            {
                "id": "A002",
                "status": "pending",
                "note": "",
            },
        ])

        completed, payload = self.run_gate(
            before,
            after,
            expected=2,
        )

        self.assertEqual(completed.returncode, 1)
        self.assertEqual(
            payload["error"]["code"],
            "unexpected_change_count",
        )

    def test_missing_file_uses_exit_code_2(self) -> None:
        missing = self.root / "missing.csv"
        after = self.write_csv(
            "io_after.csv",
            self.before_rows,
        )

        completed, payload = self.run_gate(missing, after)

        self.assertEqual(completed.returncode, 2)
        self.assertEqual(
            payload["error"]["code"],
            "io_error",
        )

    def test_invalid_argument_uses_exit_code_2(self) -> None:
        before = self.write_csv(
            "argument_before.csv",
            self.before_rows,
        )
        after = self.write_csv(
            "argument_after.csv",
            self.before_rows,
        )

        completed, payload = self.run_gate(
            before,
            after,
            expected=-1,
        )

        self.assertEqual(completed.returncode, 2)
        self.assertEqual(
            payload["error"]["code"],
            "usage_error",
        )


if __name__ == "__main__":
    unittest.main()

run_gateは、標準出力が1行だけであることと、標準エラー出力が空であることを確認してからJSONを解析します。CLIが例外終了した場合や、JSON以外の文字列を出した場合もテスト失敗になります。

子プロセスにはPYTHONIOENCODING=utf-8を渡しています。これにより、OSの既定文字コードに左右されず、日本語を含むJSONをUTF-8として読み取れます。

追加ライブラリは不要です。2つのファイルを保存したディレクトリで、次のコマンドを実行します。

python -m unittest -v test_csv_gate.py

6件のテストがすべてokになり、最後にOKと表示されれば、正常系、主要な契約違反、変更件数の不一致、入出力エラー、不正な引数を確認できています。実行時間などの表示は環境によって異なります。

手順3：実際のCSVで入出力を確認する

更新前の内容をbefore.csvとして保存します。

id,status,note
A001,pending,"call, email"
A002,pending,

更新後のafter.csvでは、許可対象のstatusだけを変更します。

id,status,note
A001,approved,"call, email"
A002,pending,

変更可能列と予定変更件数を指定して実行します。

python csv_gate.py --before before.csv --after after.csv --key id --allow status --expected-changes 1

通過時の標準出力は次の1行で、終了コードは0です。

{"changed_columns": ["status"], "changed_keys": ["A001"], "ok": true, "row_count": 2}

A001のnoteもcall onlyへ変更すると、許可されていない列の変更として終了コード1になります。

{"error": {"code": "forbidden_change", "columns": ["note"], "key": "A001"}, "ok": false}

このテストで守れる範囲

subprocess.runでCLIを別プロセスとして呼び出すため、引数の解釈、CSVの読み込み、JSONの形式、標準出力と標準エラー出力、終了コードをまとめて検査できます。後続の処理が依存するCLIの境界を確認するテストです。

一方、次の制約は残ります。

• 値は文字列として厳密に比較します。01と1、末尾空白の有無は別の値です。
• 引用符や改行コードだけが異なり、CSVから読み取ったフィールド値が同じ場合は変更に数えません。
• 列の追加、削除、並べ替えと、レコードの追加、削除はすべて拒否します。
• 更新前後のCSVを辞書としてメモリへ保持します。大容量データでは、キー順に整列した逐次比較やデータベースを使う別の設計が必要です。
• --expected-changesが数えるのはセル数ではなく、1つ以上の値が変わったレコード数です。
• --allowの指定が業務上正しいかどうかは判定できません。承認済みの変更内容に基づき、人または管理された実行設定から明示的に渡す必要があります。
• timeout=10は小さなテスト用CSVを前提とした停止策です。実データの処理時間を保証する設定ではありません。

次の一歩

実際の列名に合わせて--keyと--allowを変更し、正常な更新、許可外の列変更、空の主キー、重複キー、レコード欠落のテストデータを用意します。品質ゲートの仕様を変更するときは、期待する動作を先にテストへ反映し、そのテストが意図どおり失敗することを確かめてから実装を変更すると、判定の変化を追いやすくなります。