python
# pytest-style pseudocode
import numpy as np
import cv2
import time
import tempfile
import os
from my_pipeline import load_image, preprocess, augment, normalize, Pipeline
# Helpers: create images/files
def write_jpg(path, arr): cv2.imwrite(path, cv2.cvtColor(arr, cv2.COLOR_RGB2BGR))
def write_corrupt(path): open(path,'wb').write(b'not an image')
# 1) Corrupted file handling (unit)
def test_load_corrupted_file_raises_or_skips(tmp_path):
p = tmp_path/"bad.jpg"
write_corrupt(str(p))
with pytest.raises(ValueError) or assert pipeline.skip_corrupt:
load_image(str(p)) # expected: raise specific exception OR returns None if configured to skip
# 2) Variable image sizes (unit)
def test_resize_preserves_aspect_and_outputs_target_size():
img = np.zeros((800,200,3), dtype=np.uint8) # tall vs wide variations
out = preprocess.resize_and_center_crop(img, target=(224,224))
assert out.shape == (224,224,3)
# check no extreme stretching: aspect ratio preserved before crop
assert abs((img.shape[1]/img.shape[0]) - (out.shape[1]/out.shape[0])) > 0 or check padding logic
# 3) Color space handling (unit)
def test_color_space_conversion_rgb_to_grayscale_and_back():
rgb = np.random.randint(0,255,(300,300,3),dtype=np.uint8)
gray = preprocess.to_grayscale(rgb)
assert gray.ndim == 2
recon = preprocess.to_rgb(gray)
assert recon.shape == (300,300,3)
# 4) Deterministic augmentations with seed (unit)
def test_augment_deterministic_with_seed():
img = np.random.randint(0,255,(224,224,3),dtype=np.uint8)
a1 = augment(img, seed=42)
a2 = augment(img, seed=42)
assert np.array_equal(a1, a2)
a3 = augment(img, seed=43)
assert not np.array_equal(a1, a3)
# 5) Correctness of normalization (unit)
def test_normalize_mean_std():
img = np.full((2,2,3), 128, dtype=np.float32) # constant image
norm = normalize(img, mean=[0.5,0.5,0.5], std=[0.5,0.5,0.5]) # assume input scaled [0,1]
expected = (0.5 - 0.5)/0.5 # zero
assert np.allclose(norm, 0.0, atol=1e-6)
# 6) Integration: pipeline end-to-end produces correct dtype and shape
def test_pipeline_integration_outputs_tensor_shape_and_dtype():
files = [create_temp_image((300,400)), create_temp_image((100,100))]
pipe = Pipeline(target_size=(224,224), mean=[0.485,0.456,0.406], std=[0.229,0.224,0.225])
batch = pipe.process_batch(files)
assert batch.shape == (len(files), 3, 224, 224) # channels-first for model
assert batch.dtype == np.float32
# 7) Performance benchmark (integration)
def test_pipeline_throughput_under_threshold():
files = [create_temp_image((512,512)) for _ in range(100)]
pipe = Pipeline(num_workers=4)
start = time.time()
pipe.process_batch(files) # maybe streams internally
elapsed = time.time() - start
assert elapsed < 5.0 # assert configurable threshold for CI (example)
# 8) Edge-case: extremely small image upscaling (unit/integration)
def test_small_image_upscale_and_no_nan_values():
img = np.random.randint(0,255,(1,1,3),dtype=np.uint8)
out = preprocess.resize_and_center_crop(img, (224,224))
assert out.shape == (224,224,3)
assert not np.isnan(out).any()