The CSV Test Suite Nobody Writes

def test_line_endings_are_crlf(csv_content: bytes):
    """Every record must end with CRLF, not LF-only or CR-only."""
    lines = csv_content.split(b'\r\n')
    # Check no bare LF exists outside of quoted fields
    for i, line in enumerate(lines):
        assert b'\n' not in line, (
            f"Row {i+1}: Found bare LF (Unix line ending). "
            f"RFC 4180 requires CRLF."
        )
    # Check no bare CR exists outside of quoted fields
    for i, line in enumerate(lines):
        assert b'\r' not in line, (
            f"Row {i+1}: Found bare CR (old Mac line ending). "
            f"RFC 4180 requires CRLF."
        )

def test_consumer_handles_mixed_line_endings(parser):
    """Consumer should handle LF, CR, and CRLF gracefully."""
    test_cases = {
        "unix_lf": b"name,age\nAlice,30\nBob,25\n",
        "windows_crlf": b"name,age\r\nAlice,30\r\nBob,25\r\n",
        "old_mac_cr": b"name,age\rAlice,30\rBob,25\r",
        "mixed": b"name,age\r\nAlice,30\nBob,25\r",
    }
    for label, content in test_cases.items():
        result = parser(content)
        assert len(result) == 2, (
            f"Line ending style '{label}': Expected 2 data rows, "
            f"got {len(result)}"
        )

def test_trailing_newline_does_not_create_empty_row(parser):
    """A trailing CRLF should not produce an extra empty record."""
    with_trailing = b"name,age\r\nAlice,30\r\n"
    without_trailing = b"name,age\r\nAlice,30"
    assert len(parser(with_trailing)) == len(parser(without_trailing))

def test_header_row_present_and_matches_schema(csv_file, expected_columns):
    """When headers are expected, verify they match the schema exactly."""
    header = csv_file.readline().strip().split(',')
    assert header == expected_columns, (
        f"Header mismatch.\n"
        f"Expected: {expected_columns}\n"
        f"Got:      {header}"
    )

def test_no_duplicate_column_names(csv_file):
    """Duplicate column names cause silent column overwrites in pandas."""
    header = csv_file.readline().strip().split(',')
    duplicates = [h for h in header if header.count(h) > 1]
    assert not duplicates, (
        f"Duplicate column names found: {set(duplicates)}. "
        f"Pandas will silently rename these to name.1, name.2, etc."
    )

def test_headerless_file_matches_manifest_schema(csv_file, manifest):
    """When headers are intentionally omitted for security,
    the first data row must match the manifest's column count
    and the manifest must define column names and types."""
    first_row = csv_file.readline().strip().split(',')
    expected_cols = manifest['files'][0]['columns']
    assert len(first_row) == len(expected_cols), (
        f"Headerless file has {len(first_row)} fields but manifest "
        f"defines {len(expected_cols)} columns. Without headers, "
        f"column alignment is entirely dependent on the manifest."
    )
    # Verify manifest actually defines names (not just count)
    for col_def in expected_cols:
        assert 'name' in col_def, (
            "Manifest column definitions must include 'name' — "
            "headerless files are only safe if the schema is "
            "fully specified in the manifest."
        )

def test_consistent_column_count(csv_reader):
    """Every row must have the same number of fields as the header."""
    header = next(csv_reader)
    expected = len(header)
    ragged_rows = []
    for i, row in enumerate(csv_reader, start=2):
        if len(row) != expected:
            ragged_rows.append({
                'row': i,
                'expected': expected,
                'got': len(row),
                'content': row[:3]  # First 3 fields for debugging
            })
    assert not ragged_rows, (
        f"Found {len(ragged_rows)} ragged rows. "
        f"First: row {ragged_rows[0]['row']} has "
        f"{ragged_rows[0]['got']} fields (expected {expected}). "
        f"This usually means an unescaped comma or newline in the data."
    )

# -- Producer tests: Do we generate valid quoted fields? --

def test_fields_with_commas_are_quoted():
    """A field containing a comma must be wrapped in double-quotes."""
    record = generate_csv_row(address="456 Oak Ave, Apt 2B")
    assert record == 'id,"456 Oak Ave, Apt 2B",other_field'

def test_fields_with_quotes_are_escaped():
    """Double-quotes in data must be doubled, and field must be quoted."""
    record = generate_csv_row(company='Acme "Holdings" Ltd')
    assert '"Acme ""Holdings"" Ltd"' in record

def test_fields_with_newlines_are_quoted():
    """A field containing a line break must be wrapped in double-quotes."""
    record = generate_csv_row(notes="Line one\r\nLine two")
    # The entire field, including the newline, should be inside quotes
    assert '"Line one\r\nLine two"' in record

# -- Consumer tests: Can we parse valid-but-tricky fields? --

def test_parse_embedded_comma():
    content = b'name,address\r\n"Alice","456 Oak Ave, Apt 2B"\r\n'
    result = parse(content)
    assert result[0]['address'] == '456 Oak Ave, Apt 2B'

def test_parse_embedded_newline():
    content = b'name,notes\r\n"Alice","Line one\r\nLine two"\r\n'
    result = parse(content)
    assert result[0]['notes'] == 'Line one\r\nLine two'

def test_parse_escaped_quotes():
    content = b'name,company\r\n"Alice","Acme ""Holdings"" Ltd"\r\n'
    result = parse(content)
    assert result[0]['company'] == 'Acme "Holdings" Ltd'

def test_no_bom_prefix(csv_path):
    """UTF-8 BOM causes invisible characters in the first column name."""
    with open(csv_path, 'rb') as f:
        first_bytes = f.read(3)
    assert first_bytes != b'\xef\xbb\xbf', (
        "File contains UTF-8 BOM. This will prepend an invisible "
        "character to the first column name and break column lookups."
    )

def test_file_is_valid_utf8(csv_path):
    """Catch encoding mismatches before they become mojibake."""
    with open(csv_path, 'rb') as f:
        content = f.read()
    try:
        content.decode('utf-8')
    except UnicodeDecodeError as e:
        # Try to identify the actual encoding
        import chardet
        detected = chardet.detect(content)
        raise AssertionError(
            f"File is not valid UTF-8. Decode error at byte {e.start}. "
            f"Detected encoding: {detected['encoding']} "
            f"(confidence: {detected['confidence']:.0%}). "
            f"This is likely a Latin-1 or Windows-1252 file."
        )

def test_no_null_bytes(csv_path):
    """Null bytes (0x00) indicate binary data or a UTF-16 file
    being read as UTF-8."""
    with open(csv_path, 'rb') as f:
        content = f.read()
    null_positions = [i for i, b in enumerate(content) if b == 0]
    assert not null_positions, (
        f"Found {len(null_positions)} null bytes. "
        f"First at position {null_positions[0]}. "
        f"This might be a UTF-16 encoded file."
    )

# These are all real null representations I've seen in production
NULL_VARIANTS = [
    '',           # Empty string (most common)
    'NULL',       # MySQL Workbench default
    'null',       # JavaScript/JSON convention
    'None',       # Python convention
    r'\N',        # PostgreSQL COPY convention
    'N/A',        # Business user convention
    'n/a',        # Lowercase variant
    'NA',         # R convention
    '#N/A',       # Excel formula error
    '-',          # Dashboard export convention
    'NaN',        # Pandas/NumPy convention
    '(null)',     # Some BI tools
]

def test_null_representation_is_consistent(csv_path, null_convention=''):
    """All null values should use the agreed convention.
    Mixed conventions cause type inference failures."""
    import csv
    with open(csv_path, newline='') as f:
        reader = csv.reader(f)
        header = next(reader)
        violations = []
        for row_num, row in enumerate(reader, start=2):
            for col_idx, value in enumerate(row):
                if value in NULL_VARIANTS and value != null_convention:
                    violations.append({
                        'row': row_num,
                        'column': header[col_idx],
                        'value': repr(value),
                        'expected': repr(null_convention)
                    })
    assert not violations, (
        f"Found {len(violations)} non-standard null representations. "
        f"First: row {violations[0]['row']}, "
        f"column '{violations[0]['column']}' contains "
        f"{violations[0]['value']} (expected {violations[0]['expected']})"
    )

def test_leading_zeros_preserved(csv_path, zero_padded_columns):
    """Fields that should retain leading zeros still have them.
    Excel strips these silently on open-and-save."""
    import csv
    with open(csv_path, newline='') as f:
        reader = csv.DictReader(f)
        for row_num, row in enumerate(reader, start=2):
            for col in zero_padded_columns:
                value = row[col]
                if value and not value[0].isdigit():
                    continue  # Skip non-numeric values
                if value and value[0] == '0' and len(value) > 1:
                    pass  # Good — leading zero preserved
                elif value and col in zero_padded_columns:
                    # Check against expected length
                    # e.g., ZIP should be 5 digits
                    pass

def test_no_scientific_notation(csv_path, numeric_columns):
    """Numbers should not be in scientific notation (1.23E+11).
    This happens when Excel converts long numbers."""
    import csv, re
    sci_notation = re.compile(r'^-?\d+\.?\d*[eE][+-]?\d+$')
    with open(csv_path, newline='') as f:
        reader = csv.DictReader(f)
        violations = []
        for row_num, row in enumerate(reader, start=2):
            for col in numeric_columns:
                if sci_notation.match(row[col] or ''):
                    violations.append((row_num, col, row[col]))
    assert not violations, (
        f"Scientific notation found — data may have been "
        f"corrupted by Excel. First: row {violations[0][0]}, "
        f"column '{violations[0][1]}', value '{violations[0][2]}'"
    )

def test_dates_are_iso8601(csv_path, date_columns):
    """Date fields should use ISO 8601 (YYYY-MM-DD) format.
    Ambiguous formats like MM/DD/YYYY vs DD/MM/YYYY
    are programmatically unresolvable."""
    import csv, re
    iso_pattern = re.compile(r'^\d{4}-\d{2}-\d{2}')
    with open(csv_path, newline='') as f:
        reader = csv.DictReader(f)
        violations = []
        for row_num, row in enumerate(reader, start=2):
            for col in date_columns:
                val = row.get(col, '')
                if val and not iso_pattern.match(val):
                    violations.append((row_num, col, val))
    assert not violations, (
        f"Non-ISO date format found. First: row {violations[0][0]}, "
        f"column '{violations[0][1]}', value '{violations[0][2]}'. "
        f"Ambiguous date formats cannot be safely parsed."
    )

INVISIBLE_CHARS = {
    '\u00a0': 'Non-breaking space',
    '\u200b': 'Zero-width space',
    '\u200c': 'Zero-width non-joiner',
    '\u200d': 'Zero-width joiner',
    '\ufeff': 'BOM / Zero-width no-break space',
    '\u2028': 'Line separator',
    '\u2029': 'Paragraph separator',
    '\u00ad': 'Soft hyphen',
}

def test_no_invisible_characters(csv_path):
    """Invisible Unicode characters break joins, type casting,
    and string comparisons while being impossible to spot visually."""
    with open(csv_path, encoding='utf-8') as f:
        content = f.read()
    found = []
    for char, name in INVISIBLE_CHARS.items():
        positions = [i for i, c in enumerate(content) if c == char]
        if positions:
            found.append(f"{name} (U+{ord(char):04X}): "
                        f"{len(positions)} occurrences")
    assert not found, (
        "Invisible characters detected:\n" +
        "\n".join(f"  - {f}" for f in found)
    )

{
  "manifest_version": "1.0",
  "generated_at": "2026-03-04T08:00:00Z",
  "source_system": "vendor-crm",
  "batch_id": "batch-20260304-001",
  "files": [
    {
      "filename": "customers_20260304.csv",
      "sha256": "a3f2b8c9d1e4f5a6b7c8d9e0f1a2b3c4d5e6f7...",
      "row_count": 50432,
      "row_count_includes_header": true,
      "file_size_bytes": 8234567,
      "column_count": 12,
      "encoding": "UTF-8",
      "delimiter": ",",
      "quote_char": "\"",
      "has_header": true,
      "columns": [
        { "name": "id", "type": "integer" },
        { "name": "name", "type": "string" },
        { "name": "email", "type": "string" },
        { "name": "created_date", "type": "date", "format": "YYYY-MM-DD" }
      ]
    }
  ]
}

import hashlib
import json
import os

def validate_delivery(manifest_path, data_dir):
    """Validate a CSV delivery against its manifest.
    Returns list of errors. Empty list = valid delivery."""
    errors = []

    # 1. Parse manifest
    with open(manifest_path) as f:
        manifest = json.load(f)

    for file_spec in manifest['files']:
        filepath = os.path.join(data_dir, file_spec['filename'])

        # 2. File existence
        if not os.path.exists(filepath):
            errors.append(f"MISSING: {file_spec['filename']} not found")
            continue

        # 3. File size
        actual_size = os.path.getsize(filepath)
        if actual_size == 0:
            errors.append(f"EMPTY: {file_spec['filename']} is zero bytes")
            continue
        if actual_size != file_spec['file_size_bytes']:
            errors.append(
                f"SIZE MISMATCH: {file_spec['filename']} "
                f"expected {file_spec['file_size_bytes']} bytes, "
                f"got {actual_size}"
            )

        # 4. Checksum
        sha256 = hashlib.sha256()
        with open(filepath, 'rb') as f:
            for chunk in iter(lambda: f.read(8192), b''):
                sha256.update(chunk)
        actual_hash = sha256.hexdigest()
        if actual_hash != file_spec['sha256']:
            errors.append(
                f"CHECKSUM MISMATCH: {file_spec['filename']} "
                f"has been modified or corrupted in transit"
            )

        # 5. Row count
        with open(filepath, encoding=file_spec.get('encoding', 'utf-8')) as f:
            actual_rows = sum(1 for _ in f)
        expected = file_spec['row_count']
        if actual_rows != expected:
            errors.append(
                f"ROW COUNT MISMATCH: {file_spec['filename']} "
                f"expected {expected} rows, got {actual_rows}"
            )

        # 6. Column count and names
        with open(filepath, encoding=file_spec.get('encoding', 'utf-8')) as f:
            import csv
            reader = csv.reader(f)
            header = next(reader)
            if len(header) != file_spec['column_count']:
                errors.append(
                    f"COLUMN COUNT MISMATCH: {file_spec['filename']} "
                    f"expected {file_spec['column_count']} columns, "
                    f"got {len(header)}"
                )
            if 'columns' in file_spec:
                expected_names = [c['name'] for c in file_spec['columns']]
                if header != expected_names:
                    errors.append(
                        f"COLUMN NAME MISMATCH: {file_spec['filename']}\n"
                        f"  Expected: {expected_names}\n"
                        f"  Got:      {header}"
                    )

    return errors

# Airflow DAG pattern: manifest-gated processing
wait_for_manifest = FileSensor(
    task_id="wait_for_manifest",
    filepath="/landing/{{ ds_nodash }}/manifest.json",
    poke_interval=120,
    timeout=7200,
)

validate = BranchPythonOperator(
    task_id="validate_delivery",
    python_callable=validate_and_route,
    # Routes to either 'load_data' or 'quarantine_and_alert'
)

wait_for_manifest >> validate >> [load_data, quarantine_and_alert]

def test_detects_truncated_file():
    """Simulate a file that was cut short during transfer."""
    # Write a file, then truncate it
    ...

def test_detects_missing_file():
    """Manifest references a file that doesn't exist."""
    ...

def test_detects_extra_rows():
    """File has more rows than manifest claims —
    could indicate duplicate data or appended garbage."""
    ...

def test_detects_wrong_checksum():
    """File was modified after manifest was generated."""
    ...

def test_rejects_zero_byte_file():
    """Empty files should never pass validation."""
    ...

importcsvimporthashlibimportosimportrefromdataclassesimportdataclass, fieldfromtypingimportOptional@dataclassclassCSVValidationResult:"""Results from validating a CSV file."""filepath: stris_valid: bool=Trueerrors: list=field(default_factory=list)warnings: list=field(default_factory=list)stats: dict=field(default_factory=dict)defadd_error(self, message: str):self.errors.append(message)self.is_valid=Falsedefadd_warning(self, message: str):self.warnings.append(message)defvalidate_csv(filepath: str,expected_columns: Optional[list]=None,expected_encoding: str='utf-8',date_columns: Optional[list]=None,zero_padded_columns: Optional[list]=None,null_convention: str='',max_row_count: Optional[int]=None,min_row_count: Optional[int]=None,)->CSVValidationResult:"""Validate a CSV file against common quality checks.Works for both producers (checking your own output)and consumers (checking vendor deliveries)."""result=CSVValidationResult(filepath=filepath)# --- Check 1: File exists and isn't empty ---ifnotos.path.exists(filepath):result.add_error(f"File not found:{filepath}")returnresultfile_size=os.path.getsize(filepath)iffile_size==0:result.add_error("File is empty (zero bytes)")returnresultresult.stats['file_size_bytes']=file_size# --- Check 2: Encoding ---withopen(filepath,'rb')asf:raw_head=f.read(10000)# BOM checkifraw_head[:3]==b'\xef\xbb\xbf':result.add_warning("File contains UTF-8 BOM — this will corrupt ""the first column name in most parsers")# Null byte checkifb'\x00'inraw_head:result.add_error("File contains null bytes — likely wrong encoding ""(e.g., UTF-16 read as UTF-8) or binary data")returnresult# Encoding validitytry:raw_full=open(filepath,'rb').read()raw_full.decode(expected_encoding)exceptUnicodeDecodeErrorase:result.add_error(f"File is not valid{expected_encoding}. "f"Decode error at byte position{e.start}.")returnresult# --- Check 3: Structural integrity ---encoding='utf-8-sig'ifraw_head[:3]==b'\xef\xbb\xbf'elseexpected_encodingwithopen(filepath, newline='', encoding=encoding)asf:reader=csv.reader(f)# Header checktry:header=next(reader)exceptStopIteration:result.add_error("File has no rows (not even a header)")returnresultresult.stats['column_count']=len(header)result.stats['columns']=header# Duplicate column checkseen={}forcolinheader:seen[col]=seen.get(col,0)+1dupes={k: vfork, vinseen.items()ifv>1}ifdupes:result.add_error(f"Duplicate column names:{dupes}")# Expected columns checkifexpected_columnsandheader!=expected_columns:result.add_error(f"Column mismatch.\n"f" Expected:{expected_columns}\n"f" Got:{header}")# Row-level checksrow_count=0ragged_rows=[]iso_date=re.compile(r'^\d{4}-\d{2}-\d{2}')forrow_num, rowinenumerate(reader, start=2):row_count+=1# Column count consistencyiflen(row)!=len(header):ragged_rows.append((row_num, len(row)))# Date format checkifdate_columns:forcolindate_columns:idx=header.index(col)ifcolinheaderelseNoneifidxisnotNoneandrow[idx]:ifnotiso_date.match(row[idx]):result.add_warning(f"Row{row_num}, column '{col}': "f"Non-ISO date '{row[idx]}'")result.stats['row_count']=row_countifragged_rows:result.add_error(f"{len(ragged_rows)}rows have wrong column count. "f"Expected{len(header)}columns. "f"First bad row: line{ragged_rows[0][0]}"f"has{ragged_rows[0][1]}fields.")# Row count boundsifmin_row_countandrow_count<min_row_count:result.add_error(f"Row count{row_count}is below minimum "f"threshold of{min_row_count}")ifmax_row_countandrow_count>max_row_count:result.add_error(f"Row count{row_count}exceeds maximum "f"threshold of{max_row_count}")# --- Check 4: Invisible characters ---invisible={'\u00a0':'Non-breaking space','\u200b':'Zero-width space','\ufeff':'BOM (mid-file)',}text=raw_full.decode(encoding, errors='replace')forchar, nameininvisible.items():count=text.count(char)ifcount>0:result.add_warning(f"Found{count}'{name}' characters (U+{ord(char):04X})")# --- Compute checksum for manifest comparison ---sha256=hashlib.sha256(raw_full).hexdigest()result.stats['sha256']=sha256returnresult

result=validate_csv('vendor_delivery/customers_20260304.csv',expected_columns=['id','name','email','city','state','zip'],date_columns=['created_date','updated_date'],min_row_count=1000,max_row_count=100000,)ifnotresult.is_valid:forerrorinresult.errors:print(f"ERROR:{error}")# Route to quarantineelse:print(f"Valid.{result.stats['row_count']}rows, "f"SHA-256:{result.stats['sha256'][:16]}...")# Proceed to loading