Files
automatic/test/test-generation-api.py
CalamitousFelicitousness 0c0d455c11 fix(api): resolve sampler names case-insensitively to canonical form
validate_sampler_name only matched the exact, case-sensitive name, so near-miss client names such as lowercase variants were rejected while an omitted name silently used the model scheduler via the Default sentinel. Fall back to find_sampler and return the canonical name so create_sampler applies the intended sampler; unknown names still return 404. Add an API test covering case-insensitive resolution and rejection of unknown names.
2026-06-07 06:57:07 +01:00

647 lines
27 KiB
Python

#!/usr/bin/env python
"""
API tests for generation with scheduler params, color grading, and latent corrections.
Tests:
- GET /sdapi/v1/samplers — sampler enumeration and config
- POST /sdapi/v1/txt2img — generation with various samplers
- POST /sdapi/v1/txt2img — generation with color grading params
- POST /sdapi/v1/txt2img — generation with latent correction params
Requires a running SD.Next instance with a model loaded.
Usage:
python test/test-generation-api.py [--url URL] [--steps STEPS]
"""
import io
import os
import sys
import json
import time
import base64
import argparse
import requests
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
class GenerationAPITest:
"""Test harness for generation API with scheduler and grading params."""
# Samplers to test — a representative subset covering different scheduler families
TEST_SAMPLERS = [
'Euler a',
'Euler',
'DPM++ 2M',
'UniPC',
'DDIM',
'DPM++ 2M SDE',
]
def __init__(self, base_url, steps=10, timeout=300):
self.base_url = base_url.rstrip('/')
self.steps = steps
self.timeout = timeout
self.results = {
'samplers': {'passed': 0, 'failed': 0, 'skipped': 0, 'tests': []},
'generation': {'passed': 0, 'failed': 0, 'skipped': 0, 'tests': []},
'grading': {'passed': 0, 'failed': 0, 'skipped': 0, 'tests': []},
'correction': {'passed': 0, 'failed': 0, 'skipped': 0, 'tests': []},
'param_validation': {'passed': 0, 'failed': 0, 'skipped': 0, 'tests': []},
}
self._category = 'samplers'
self._critical_error = None
def _get(self, endpoint):
try:
r = requests.get(f'{self.base_url}{endpoint}', timeout=self.timeout, verify=False)
if r.status_code != 200:
return {'error': r.status_code, 'reason': r.reason}
return r.json()
except requests.exceptions.ConnectionError:
return {'error': 'connection_refused', 'reason': 'Server not running'}
except Exception as e:
return {'error': 'exception', 'reason': str(e)}
def _post(self, endpoint, data):
try:
r = requests.post(f'{self.base_url}{endpoint}', json=data, timeout=self.timeout, verify=False)
if r.status_code != 200:
return {'error': r.status_code, 'reason': r.reason}
return r.json()
except requests.exceptions.ConnectionError:
return {'error': 'connection_refused', 'reason': 'Server not running'}
except Exception as e:
return {'error': 'exception', 'reason': str(e)}
def record(self, passed, name, detail=''):
status = 'PASS' if passed else 'FAIL'
self.results[self._category]['passed' if passed else 'failed'] += 1
self.results[self._category]['tests'].append((status, name))
msg = f' {status}: {name}'
if detail:
msg += f' ({detail})'
print(msg)
def skip(self, name, reason):
self.results[self._category]['skipped'] += 1
self.results[self._category]['tests'].append(('SKIP', name))
print(f' SKIP: {name} ({reason})')
def _txt2img(self, extra_params=None, prompt='a cat'):
"""Helper: run txt2img with base params + overrides. Returns (data, time)."""
payload = {
'prompt': prompt,
'steps': self.steps,
'width': 512,
'height': 512,
'seed': 42,
'save_images': False,
'send_images': True,
}
if extra_params:
payload.update(extra_params)
t0 = time.time()
data = self._post('/sdapi/v1/txt2img', payload)
return data, time.time() - t0
def _check_generation(self, data, test_name, elapsed):
"""Validate a generation response has images."""
if 'error' in data:
self.record(False, test_name, f"error: {data}")
return False
has_images = 'images' in data and len(data['images']) > 0
self.record(has_images, test_name, f"time={elapsed:.1f}s")
return has_images
def _get_info(self, data):
"""Extract info string from generation response."""
if 'info' not in data:
return ''
info = data['info']
return info if isinstance(info, str) else json.dumps(info)
def _decode_image(self, data):
"""Decode first image from generation response into numpy array."""
import numpy as np
from PIL import Image
if 'images' not in data or len(data['images']) == 0:
return None
img_data = data['images'][0].split(',', 1)[0]
img = Image.open(io.BytesIO(base64.b64decode(img_data))).convert('RGB')
return np.array(img, dtype=np.float32)
def _pixel_diff(self, arr_a, arr_b):
"""Mean absolute pixel difference between two images (0-255 scale)."""
import numpy as np
if arr_a is None or arr_b is None:
return -1.0
if arr_a.shape != arr_b.shape:
return -1.0
return float(np.abs(arr_a - arr_b).mean())
def _channel_means(self, arr):
"""Return per-channel means [R, G, B]."""
if arr is None:
return [0, 0, 0]
return [float(arr[:, :, c].mean()) for c in range(3)]
# =========================================================================
# Tests: Sampler Enumeration
# =========================================================================
def test_samplers_list(self):
"""GET /sdapi/v1/samplers returns available samplers with config."""
self._category = 'samplers'
print("\n--- Sampler Enumeration ---")
data = self._get('/sdapi/v1/samplers')
if 'error' in data:
self.record(False, 'samplers_list', f"error: {data}")
self._critical_error = f"Server error: {data}"
return []
if not isinstance(data, list):
self.record(False, 'samplers_list', f"expected list, got {type(data).__name__}")
return []
self.record(True, 'samplers_list', f"{len(data)} samplers available")
# Check that each sampler has a name
sampler_names = []
for s in data:
name = s.get('name', '')
if name:
sampler_names.append(name)
self.record(len(sampler_names) == len(data), 'samplers_have_names',
f"{len(sampler_names)}/{len(data)} have names")
# Check for our test samplers
for test_sampler in self.TEST_SAMPLERS:
found = test_sampler in sampler_names
if not found:
self.skip(f'sampler_available_{test_sampler}', 'not in server sampler list')
else:
self.record(True, f'sampler_available_{test_sampler}')
return sampler_names
# =========================================================================
# Tests: Generation with Different Samplers
# =========================================================================
def test_samplers_generate(self, available_samplers):
"""Generate with each test sampler and verify success."""
self._category = 'generation'
print("\n--- Generation with Different Samplers ---")
if self._critical_error:
for s in self.TEST_SAMPLERS:
self.skip(f'generate_{s}', self._critical_error)
return
for sampler in self.TEST_SAMPLERS:
if sampler not in available_samplers:
self.skip(f'generate_{sampler}', 'sampler not available')
continue
data, elapsed = self._txt2img({'sampler_name': sampler})
self._check_generation(data, f'generate_{sampler}', elapsed)
def test_sampler_name_resolution(self, available_samplers):
"""Sampler name resolution: a case-insensitive name resolves to the canonical sampler
(and is applied, not silently swapped for the model default), while an unknown name is
rejected rather than falling back to the default scheduler."""
self._category = 'samplers'
print("\n--- Sampler Name Resolution ---")
if self._critical_error:
self.skip('sampler_lenient_case', self._critical_error)
self.skip('sampler_unknown_rejected', self._critical_error)
return
canonical = next((s for s in ('Euler a', 'DPM++ 2M', 'UniPC') if s in available_samplers), None)
if canonical is None:
self.skip('sampler_lenient_case', 'no known sampler available')
else:
data, _ = self._txt2img({'sampler_name': canonical.lower()})
if 'error' in data:
self.record(False, 'sampler_lenient_case', f"lowercase '{canonical.lower()}' rejected: {data}")
else:
resolved = canonical in self._get_info(data)
self.record(resolved, 'sampler_lenient_case',
f"'{canonical.lower()}' -> '{canonical}'" if resolved
else f"generated but '{canonical}' not in info (model default used?)")
data, _ = self._txt2img({'sampler_name': 'ThisIsNotARealSampler'})
rejected = 'error' in data
self.record(rejected, 'sampler_unknown_rejected',
'unknown name rejected' if rejected else 'unknown name was NOT rejected')
# =========================================================================
# Tests: Color Grading Params
# =========================================================================
def test_grading_brightness_contrast(self):
"""Generate with grading brightness and contrast."""
data, elapsed = self._txt2img({
'grading_brightness': 0.2,
'grading_contrast': 0.3,
})
self._check_generation(data, 'grading_brightness_contrast', elapsed)
def test_grading_saturation_hue(self):
"""Generate with grading saturation and hue shift."""
data, elapsed = self._txt2img({
'grading_saturation': 0.5,
'grading_hue': 0.1,
})
self._check_generation(data, 'grading_saturation_hue', elapsed)
def test_grading_gamma_sharpness(self):
"""Generate with gamma correction and sharpness."""
data, elapsed = self._txt2img({
'grading_gamma': 0.8,
'grading_sharpness': 0.5,
})
self._check_generation(data, 'grading_gamma_sharpness', elapsed)
def test_grading_color_temp(self):
"""Generate with warm color temperature."""
data, elapsed = self._txt2img({
'grading_color_temp': 3500,
})
self._check_generation(data, 'grading_color_temp', elapsed)
def test_grading_tone(self):
"""Generate with shadows/midtones/highlights adjustments."""
data, elapsed = self._txt2img({
'grading_shadows': 0.3,
'grading_midtones': -0.1,
'grading_highlights': 0.2,
})
self._check_generation(data, 'grading_tone', elapsed)
def test_grading_effects(self):
"""Generate with vignette and grain."""
data, elapsed = self._txt2img({
'grading_vignette': 0.5,
'grading_grain': 0.3,
})
self._check_generation(data, 'grading_effects', elapsed)
def test_grading_split_toning(self):
"""Generate with split toning colors."""
data, elapsed = self._txt2img({
'grading_shadows_tint': '#003366',
'grading_highlights_tint': '#ffcc00',
'grading_split_tone_balance': 0.6,
})
self._check_generation(data, 'grading_split_toning', elapsed)
def test_grading_combined(self):
"""Generate with multiple grading params at once."""
data, elapsed = self._txt2img({
'grading_brightness': 0.1,
'grading_contrast': 0.2,
'grading_saturation': 0.3,
'grading_gamma': 0.9,
'grading_color_temp': 5000,
'grading_vignette': 0.3,
})
self._check_generation(data, 'grading_combined', elapsed)
def run_grading_tests(self):
"""Run all grading tests."""
self._category = 'grading'
print("\n--- Color Grading Tests ---")
if self._critical_error:
self.skip('grading_all', self._critical_error)
return
self.test_grading_brightness_contrast()
self.test_grading_saturation_hue()
self.test_grading_gamma_sharpness()
self.test_grading_color_temp()
self.test_grading_tone()
self.test_grading_effects()
self.test_grading_split_toning()
self.test_grading_combined()
# =========================================================================
# Tests: Latent Correction Params
# =========================================================================
def test_correction_brightness(self):
"""Generate with latent brightness correction."""
data, elapsed = self._txt2img({'hdr_brightness': 1.5})
ok = self._check_generation(data, 'correction_brightness', elapsed)
if ok:
info = self._get_info(data)
has_param = 'Latent brightness' in info
self.record(has_param, 'correction_brightness_metadata',
'found in info' if has_param else 'not found in info')
def test_correction_color(self):
"""Generate with latent color centering."""
data, elapsed = self._txt2img({'hdr_color': 0.5, 'hdr_mode': 1})
ok = self._check_generation(data, 'correction_color', elapsed)
if ok:
info = self._get_info(data)
has_param = 'Latent color' in info
self.record(has_param, 'correction_color_metadata',
'found in info' if has_param else 'not found in info')
def test_correction_clamp(self):
"""Generate with latent clamping."""
data, elapsed = self._txt2img({
'hdr_clamp': True,
'hdr_threshold': 0.8,
'hdr_boundary': 4.0,
})
ok = self._check_generation(data, 'correction_clamp', elapsed)
if ok:
info = self._get_info(data)
has_param = 'Latent clamp' in info
self.record(has_param, 'correction_clamp_metadata',
'found in info' if has_param else 'not found in info')
def test_correction_sharpen(self):
"""Generate with latent sharpening."""
data, elapsed = self._txt2img({'hdr_sharpen': 1.0})
ok = self._check_generation(data, 'correction_sharpen', elapsed)
if ok:
info = self._get_info(data)
has_param = 'Latent sharpen' in info
self.record(has_param, 'correction_sharpen_metadata',
'found in info' if has_param else 'not found in info')
def test_correction_maximize(self):
"""Generate with latent maximize/normalize."""
data, elapsed = self._txt2img({
'hdr_maximize': True,
'hdr_max_center': 0.6,
'hdr_max_boundary': 2.0,
})
ok = self._check_generation(data, 'correction_maximize', elapsed)
if ok:
info = self._get_info(data)
has_param = 'Latent max' in info
self.record(has_param, 'correction_maximize_metadata',
'found in info' if has_param else 'not found in info')
def test_correction_combined(self):
"""Generate with multiple correction params."""
data, elapsed = self._txt2img({
'hdr_brightness': 1.0,
'hdr_color': 0.3,
'hdr_sharpen': 0.5,
'hdr_clamp': True,
})
ok = self._check_generation(data, 'correction_combined', elapsed)
if ok:
info = self._get_info(data)
# At least some correction params should appear
found = [k for k in ['Latent brightness', 'Latent color', 'Latent sharpen', 'Latent clamp'] if k in info]
self.record(len(found) > 0, 'correction_combined_metadata', f"found: {found}")
def run_correction_tests(self):
"""Run all latent correction tests."""
self._category = 'correction'
print("\n--- Latent Correction Tests ---")
if self._critical_error:
self.skip('correction_all', self._critical_error)
return
self.test_correction_brightness()
self.test_correction_color()
self.test_correction_clamp()
self.test_correction_sharpen()
self.test_correction_maximize()
self.test_correction_combined()
# =========================================================================
# Tests: Per-Request Param Validation (baseline comparison)
# =========================================================================
def _generate_baseline(self):
"""Generate a baseline image with no grading/correction. Cache and reuse."""
if hasattr(self, '_baseline_arr') and self._baseline_arr is not None:
return self._baseline_arr, self._baseline_data
data, elapsed = self._txt2img()
if 'error' in data or 'images' not in data:
return None, data
self._baseline_arr = self._decode_image(data)
self._baseline_data = data
print(f' Baseline generated: time={elapsed:.1f}s mean={self._channel_means(self._baseline_arr)}')
return self._baseline_arr, data
def _compare_param(self, name, params, check_fn=None):
"""Generate with params and compare to baseline. Optionally run check_fn(baseline, result)."""
baseline, _ = self._generate_baseline()
if baseline is None:
self.skip(f'param_{name}', 'baseline generation failed')
return
data, _elapsed = self._txt2img(params)
if 'error' in data:
self.record(False, f'param_{name}', f"generation error: {data}")
return
result = self._decode_image(data)
if result is None:
self.record(False, f'param_{name}', 'no image in response')
return
diff = self._pixel_diff(baseline, result)
differs = diff > 0.5 # more than 0.5/255 mean difference
self.record(differs, f'param_{name}_differs',
f"mean_diff={diff:.2f}" if differs else f"images identical (diff={diff:.4f})")
if check_fn and differs:
try:
ok, detail = check_fn(baseline, result, data)
self.record(ok, f'param_{name}_direction', detail)
except Exception as e:
self.record(False, f'param_{name}_direction', f"check error: {e}")
def run_param_validation_tests(self):
"""Verify per-request grading/correction params actually change the output."""
self._category = 'param_validation'
print("\n--- Per-Request Param Validation ---")
if self._critical_error:
self.skip('param_validation_all', self._critical_error)
return
import numpy as np
# -- Grading params --
# Brightness: positive should increase mean pixel value
def check_brightness(base, result, _data):
base_mean = float(base.mean())
result_mean = float(result.mean())
return result_mean > base_mean, f"baseline={base_mean:.1f} graded={result_mean:.1f}"
self._compare_param('grading_brightness', {'grading_brightness': 0.3}, check_brightness)
# Contrast: should increase standard deviation
def check_contrast(base, result, _data):
return float(result.std()) > float(base.std()), \
f"baseline_std={float(base.std()):.1f} graded_std={float(result.std()):.1f}"
self._compare_param('grading_contrast', {'grading_contrast': 0.5}, check_contrast)
# Saturation: desaturation should reduce color channel spread
def check_desaturation(base, result, _data):
base_spread = max(self._channel_means(base)) - min(self._channel_means(base))
result_spread = max(self._channel_means(result)) - min(self._channel_means(result))
return result_spread < base_spread, \
f"baseline_spread={base_spread:.1f} graded_spread={result_spread:.1f}"
self._compare_param('grading_saturation_neg', {'grading_saturation': -0.5}, check_desaturation)
# Hue shift: just verify it changes
self._compare_param('grading_hue', {'grading_hue': 0.2})
# Gamma < 1: should brighten (raise values that are < 1)
def check_gamma(base, result, _data):
return float(result.mean()) > float(base.mean()), \
f"baseline={float(base.mean()):.1f} gamma={float(result.mean()):.1f}"
self._compare_param('grading_gamma', {'grading_gamma': 0.7}, check_gamma)
# Sharpness: just verify it changes
self._compare_param('grading_sharpness', {'grading_sharpness': 0.8})
# Color temperature warm: red channel mean should increase relative to blue
def check_warm(base, result, _data):
base_r, _, base_b = self._channel_means(base)
res_r, _, res_b = self._channel_means(result)
base_rb = base_r - base_b
res_rb = res_r - res_b
return res_rb > base_rb, f"baseline R-B={base_rb:.1f} warm R-B={res_rb:.1f}"
self._compare_param('grading_color_temp_warm', {'grading_color_temp': 3000}, check_warm)
# Color temperature cool: blue should increase relative to red
def check_cool(base, result, _data):
base_r, _, base_b = self._channel_means(base)
res_r, _, res_b = self._channel_means(result)
base_rb = base_r - base_b
res_rb = res_r - res_b
return res_rb < base_rb, f"baseline R-B={base_rb:.1f} cool R-B={res_rb:.1f}"
self._compare_param('grading_color_temp_cool', {'grading_color_temp': 10000}, check_cool)
# Vignette: corners should be darker than baseline corners
def check_vignette(base, result, _data):
h, _w = base.shape[:2]
corner_size = h // 8
base_corners = np.concatenate([
base[:corner_size, :corner_size].flatten(),
base[:corner_size, -corner_size:].flatten(),
base[-corner_size:, :corner_size].flatten(),
base[-corner_size:, -corner_size:].flatten(),
])
result_corners = np.concatenate([
result[:corner_size, :corner_size].flatten(),
result[:corner_size, -corner_size:].flatten(),
result[-corner_size:, :corner_size].flatten(),
result[-corner_size:, -corner_size:].flatten(),
])
return float(result_corners.mean()) < float(base_corners.mean()), \
f"baseline_corners={float(base_corners.mean()):.1f} vignette_corners={float(result_corners.mean()):.1f}"
self._compare_param('grading_vignette', {'grading_vignette': 0.8}, check_vignette)
# Grain: just verify it changes (stochastic)
self._compare_param('grading_grain', {'grading_grain': 0.5})
# Shadows/midtones/highlights: verify changes
self._compare_param('grading_shadows', {'grading_shadows': 0.5})
self._compare_param('grading_highlights', {'grading_highlights': -0.3})
# CLAHE: should increase local contrast
self._compare_param('grading_clahe', {'grading_clahe_clip': 2.0})
# Split toning: verify changes
self._compare_param('grading_split_toning', {
'grading_shadows_tint': '#003366',
'grading_highlights_tint': '#ffcc00',
})
# -- Correction params --
# Latent brightness: should change output and appear in metadata
def check_correction_meta(key):
def _check(_base, _result, data):
info = self._get_info(data)
return key in info, f"'{key}' {'found' if key in info else 'missing'} in info"
return _check
self._compare_param('hdr_brightness', {'hdr_brightness': 2.0}, check_correction_meta('Latent brightness'))
self._compare_param('hdr_color', {'hdr_color': 0.8, 'hdr_mode': 1}, check_correction_meta('Latent color'))
self._compare_param('hdr_sharpen', {'hdr_sharpen': 1.5}, check_correction_meta('Latent sharpen'))
self._compare_param('hdr_clamp', {'hdr_clamp': True, 'hdr_threshold': 0.7}, check_correction_meta('Latent clamp'))
# Isolation: verify params from one request don't leak to the next
data_after, _ = self._txt2img()
arr_after = self._decode_image(data_after)
baseline, _ = self._generate_baseline()
if baseline is not None and arr_after is not None:
leak_diff = self._pixel_diff(baseline, arr_after)
no_leak = leak_diff < 0.5
self.record(no_leak, 'param_isolation',
f"post-grading baseline diff={leak_diff:.4f}" if no_leak
else f"LEAK: baseline changed after grading requests (diff={leak_diff:.2f})")
# =========================================================================
# Runner
# =========================================================================
def run_all(self):
print("=" * 60)
print("Generation API Test Suite")
print(f"Server: {self.base_url}")
print(f"Steps: {self.steps}")
print("=" * 60)
# Samplers
available = self.test_samplers_list()
self.test_samplers_generate(available)
self.test_sampler_name_resolution(available)
# Grading
self.run_grading_tests()
# Corrections
self.run_correction_tests()
# Per-request param validation (baseline comparison)
self.run_param_validation_tests()
# Summary
print("\n" + "=" * 60)
print("Results")
print("=" * 60)
total_passed = 0
total_failed = 0
total_skipped = 0
for cat, data in self.results.items():
total_passed += data['passed']
total_failed += data['failed']
total_skipped += data['skipped']
status = 'PASS' if data['failed'] == 0 else 'FAIL'
print(f" {cat}: {data['passed']} passed, {data['failed']} failed, {data['skipped']} skipped [{status}]")
print(f" Total: {total_passed} passed, {total_failed} failed, {total_skipped} skipped")
print("=" * 60)
return total_failed == 0
if __name__ == "__main__":
parser = argparse.ArgumentParser(description='Generation API Tests (samplers, grading, correction)')
parser.add_argument('--url', default=os.environ.get('SDAPI_URL', 'http://127.0.0.1:7860'), help='server URL')
parser.add_argument('--steps', type=int, default=10, help='generation steps (lower = faster tests)')
args = parser.parse_args()
test = GenerationAPITest(args.url, args.steps)
success = test.run_all()
sys.exit(0 if success else 1)