CSV 처리 스크립트의 성공·실패 종료와 출력 파일을 자동 시험하기

"""P55: five independent subprocess contract tests; standard library only."""
import csv
import hashlib
import io
import subprocess
import sys
import tempfile
import unittest
from dataclasses import dataclass
from pathlib import Path

ROOT = Path(__file__).resolve().parent
TARGET = ROOT / "validate_orders.py"
TIMEOUT_SECONDS = 10
LOG_LIMIT = 4096
INPUT_LIMIT = 65536
OUTPUT_LIMIT = 32768


class OutputLimitExceeded(AssertionError):
    """The completed process wrote more diagnostic bytes than allowed."""


def read_bounded(path, limit):
    with Path(path).open("rb") as stream:
        data = stream.read(limit + 1)
    if len(data) > limit:
        raise OutputLimitExceeded(f"file exceeds {limit} bytes")
    return data


def sha256(path, limit=INPUT_LIMIT):
    return hashlib.sha256(read_bounded(path, limit)).hexdigest()


def read_log(stream, name):
    stream.seek(0, 2)
    size = stream.tell()
    if size > LOG_LIMIT:
        raise OutputLimitExceeded(f"{name} exceeds {LOG_LIMIT} bytes: {size}")
    stream.seek(0)
    decoded = stream.read(LOG_LIMIT + 1).decode("utf-8", errors="strict")
    # Normalize diagnostics only; the limit and byte counts remain raw bytes.
    return decoded.replace("\r\n", "\n"), size


@dataclass(frozen=True)
class RunResult:
    returncode: int
    stdout: str
    stderr: str
    stdout_bytes: int
    stderr_bytes: int


def run_program(argv, cwd, timeout=TIMEOUT_SECONDS):
    # Files avoid capture_output's in-memory buffering. Their disk growth is
    # checked AFTER completion, not actively capped while the child is running.
    with tempfile.TemporaryFile(dir=cwd) as out_log, \
            tempfile.TemporaryFile(dir=cwd) as err_log:
        process = subprocess.run(
            argv, cwd=cwd, stdin=subprocess.DEVNULL,
            stdout=out_log, stderr=err_log, shell=False,
            timeout=timeout, check=False,
        )
        stdout, stdout_bytes = read_log(out_log, "stdout")
        stderr, stderr_bytes = read_log(err_log, "stderr")
    return RunResult(process.returncode, stdout, stderr,
                     stdout_bytes, stderr_bytes)


def run_cli(source, destination, cwd):
    return run_program(
        [sys.executable, "-I", str(TARGET), str(source), str(destination)], cwd
    )


class CliRegressionTests(unittest.TestCase):
    def setUp(self):
        temporary = tempfile.TemporaryDirectory(prefix="p55-case-")
        self.addCleanup(temporary.cleanup)
        self.work = Path(temporary.name)
        # Spaces in names also verify argument-list handling without a shell.
        self.source = self.work / "orders input.csv"
        self.output = self.work / "orders output.csv"
        self.observations = {}

    def exercise(self, fixture, expected_code, stdout, stderr, collision=False):
        original = ROOT / "fixtures" / f"{fixture}.csv"
        original_hash = sha256(original)
        self.source.write_bytes(read_bounded(original, INPUT_LIMIT))
        before = sha256(self.source)
        sentinel_hash = None
        if collision:
            self.output.write_bytes(b"DO NOT OVERWRITE\n")
            sentinel_hash = sha256(self.output)
        try:
            result = run_cli(self.source, self.output, self.work)
        finally:
            after = sha256(self.source)
            self.observations.update(
                fixture=fixture, input_sha256_before=before,
                input_sha256_after=after, input_unchanged=before == after,
                bundled_fixture_unchanged=sha256(original) == original_hash,
            )
            self.assertEqual(after, before, "input bytes changed")
            self.assertEqual(sha256(original), original_hash)
        self.observations.update(
            expected_returncode=expected_code, actual_returncode=result.returncode,
            stdout=result.stdout, stderr=result.stderr,
            stdout_bytes=result.stdout_bytes, stderr_bytes=result.stderr_bytes,
            final_output_exists=self.output.exists() or self.output.is_symlink(),
        )
        self.assertEqual(result.returncode, expected_code)
        self.assertEqual(result.stdout, stdout)
        self.assertEqual(result.stderr, stderr)
        if collision:
            after_sentinel = sha256(self.output)
            self.observations.update(
                sentinel_sha256_before=sentinel_hash,
                sentinel_sha256_after=after_sentinel,
                sentinel_unchanged=sentinel_hash == after_sentinel,
            )
            self.assertEqual(after_sentinel, sentinel_hash)
        elif expected_code != 0:
            self.assertFalse(self.output.exists() or self.output.is_symlink(),
                             "failed run left final output")
        return result

    def test_normal(self):
        self.exercise("normal", 0, "OK rows=2\n", "")
        self.assertTrue(self.output.is_file(), "success did not create output")
        self.assertFalse(self.output.is_symlink(), "success output is a symlink")
        output_bytes = read_bounded(self.output, OUTPUT_LIMIT)
        parsed = list(csv.reader(io.StringIO(output_bytes.decode("utf-8"),
                                             newline=""), strict=True))
        # Expectations deliberately do not import the target's validation code.
        expected = [
            ["order_id", "order_date", "amount"],
            ["ORD-001", "2026-10-01", "1200.00"],
            ["ORD-002", "2026-10-02", "3400.50"],
        ]
        self.assertEqual(parsed, expected)
        self.observations.update(output_rows=parsed, output_bytes=len(output_bytes),
                                 output_sha256=sha256(self.output, OUTPUT_LIMIT))

    def test_header_missing(self):
        self.exercise("header_missing", 2, "", "ERROR data: header\n")

    def test_duplicate_key(self):
        self.exercise("duplicate_key", 2, "", "ERROR data: duplicate_key\n")

    def test_impossible_date(self):
        self.exercise("impossible_date", 2, "", "ERROR data: date\n")

    def test_output_collision(self):
        self.exercise("output_collision", 3, "", "ERROR conflict\n", collision=True)


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

"""Small trusted CLI target for P55, not a production CSV framework."""
import argparse
import csv
import io
import re
import sys
from datetime import date
from decimal import Decimal
from pathlib import Path

HEADER = ["order_id", "order_date", "amount"]
MAX_INPUT_BYTES = 65536
MAX_OUTPUT_BYTES = 32768
MAX_ROWS = 1000
MAX_FIELD_CHARS = 128


class DataError(ValueError):
    pass


def validated_payload(source):
    if not source.is_file():
        raise OSError("input is not a regular file")
    with source.open("rb") as stream:
        data = stream.read(MAX_INPUT_BYTES + 1)
    if len(data) > MAX_INPUT_BYTES:
        raise DataError("input_limit")
    try:
        text = data.decode("utf-8-sig")
    except UnicodeError as exc:
        raise DataError("encoding") from exc
    csv.field_size_limit(MAX_FIELD_CHARS)
    rows, seen = [], set()
    try:
        reader = csv.reader(io.StringIO(text, newline=""), strict=True)
        if next(reader, None) != HEADER:
            raise DataError("header")
        for number, row in enumerate(reader, start=1):
            if number > MAX_ROWS:
                raise DataError("row_limit")
            if len(row) != 3:
                raise DataError("columns")
            key, day, amount = row
            if not re.fullmatch(r"[A-Z0-9_-]{1,32}", key):
                raise DataError("key")
            if key in seen:
                raise DataError("duplicate_key")
            if not re.fullmatch(r"[0-9]{4}-[0-9]{2}-[0-9]{2}", day):
                raise DataError("date")
            try:
                date.fromisoformat(day)
            except ValueError as exc:
                raise DataError("date") from exc
            if not re.fullmatch(r"[0-9]{1,9}(?:\.[0-9]{1,2})?", amount):
                raise DataError("amount")
            seen.add(key)
            rows.append([key, day, format(Decimal(amount), ".2f")])
    except csv.Error as exc:
        raise DataError("csv") from exc
    if not rows:
        raise DataError("empty")
    buffer = io.StringIO(newline="")
    writer = csv.writer(buffer, lineterminator="\n")
    writer.writerow(HEADER)
    writer.writerows(rows)
    payload = buffer.getvalue().encode("utf-8")
    if len(payload) > MAX_OUTPUT_BYTES:
        raise DataError("output_limit")
    return payload, len(rows)


def main(argv=None):
    parser = argparse.ArgumentParser(description="Validate synthetic order CSV")
    parser.add_argument("input", type=Path)
    parser.add_argument("output", type=Path)
    args = parser.parse_args(argv)
    created = False
    try:
        if (args.output.is_symlink() or args.output.exists()
                or args.input.resolve() == args.output.resolve()):
            print("ERROR conflict", file=sys.stderr)
            return 3
        payload, count = validated_payload(args.input)
        # 'xb' fails if the destination exists; it never overwrites a sentinel.
        stream = args.output.open("xb")
        created = True
        with stream:
            stream.write(payload)
    except FileExistsError:
        print("ERROR conflict", file=sys.stderr)
        return 3
    except DataError as exc:
        print(f"ERROR data: {exc}", file=sys.stderr)
        return 2
    except (OSError, RuntimeError):  # Python 3.12 resolve() may flag a symlink loop.
        if created:
            try:
                args.output.unlink()
            except OSError:
                print("ERROR io: cleanup_failed", file=sys.stderr)
                return 4
        print("ERROR io", file=sys.stderr)
        return 4
    print(f"OK rows={count}")
    return 0


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