Approach: build a configurable Python test-case generator that emits rows (dicts) for an input preprocessing function. It injects missing values, wrong types, unicode/encoding problems, extreme numerics, and very long strings. Seed both random and numpy for reproducibility. Provide small validation harness that runs preprocessing and reports failures/exceptions and summary statistics.python
import random
import string
import numpy as np
# seed for reproducibility
def set_seed(seed=42):
random.seed(seed)
np.random.seed(seed)
# helpers
def random_string(length):
return ''.join(random.choices(string.ascii_letters + string.digits, k=length))
def unicode_mess():
# returns strings with unusual unicode codepoints and invalid surrogates
return "normal\u2603\u00A9" + "".join(chr(0xD800 + (i % 0x400)) for i in range(3))
# generator
def generate_row(schema):
"""
schema: dict of field -> {'type':'int'/'float'/'str', 'nullable':bool}
returns: dict representing a row with possible corruptions
"""
row = {}
for field, meta in schema.items():
typ = meta.get('type', 'str')
# decide if corrupt
p = random.random()
if p < 0.08 and meta.get('nullable', True):
row[field] = None # missing
continue
if p < 0.18:
# invalid type: string where number expected or number where string expected
if typ in ('int','float'):
row[field] = random_string(random.randint(1,10))
else:
row[field] = random.randint(-1000, 1000)
continue
if p < 0.25:
# unicode/encoding error-ish string
row[field] = unicode_mess()
continue
if p < 0.35 and typ in ('int','float'):
# extreme numeric values
extremes = [np.inf, -np.inf, np.nan, 10**18, -10**18, np.finfo(float).max]
row[field] = random.choice(extremes)
continue
if p < 0.45 and typ == 'str':
# extremely long string
row[field] = random_string(10000)
continue
# normal valid generation
if typ == 'int':
row[field] = random.randint(-1000, 1000)
elif typ == 'float':
row[field] = random.uniform(-1e3, 1e3)
else:
row[field] = random_string(random.randint(1, 20))
return row
# batch generator
def generate_batch(schema, n=100, seed=42):
set_seed(seed)
return [generate_row(schema) for _ in range(n)]
# simple validation harness
def validate_preprocessor(preprocess_fn, batch):
results = {'success':0, 'exceptions':{}, 'invalid_outputs':0}
for i, row in enumerate(batch):
try:
out = preprocess_fn(row)
# basic output checks: not None, expected fields exist
if out is None or not isinstance(out, dict):
results['invalid_outputs'] += 1
else:
results['success'] += 1
except Exception as e:
k = type(e).__name__
results['exceptions'].setdefault(k, 0)
results['exceptions'][k] += 1
return results
# Example usage:
if __name__ == '__main__':
schema = {
'age': {'type':'int', 'nullable':False},
'income': {'type':'float', 'nullable':True},
'name': {'type':'str', 'nullable':False}
}
batch = generate_batch(schema, n=500, seed=123)
# define a dummy preprocess function for demonstration
def dummy_preprocess(row):
# simulate checks a real preprocessor would do
if row['age'] is None:
raise ValueError("age missing")
if isinstance(row['age'], str):
row['age'] = int(row['age']) # may raise
return row
print(validate_preprocessor(dummy_preprocess, batch))
