File size: 6,471 Bytes
e9190ff | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 | import json
import tempfile
import unittest
from pathlib import Path
from PIL import Image
from scripts.editing_protocol import DEFAULT_SEEDS, image_diagnostics
from scripts.editing_report import validate_suite
from scripts.integrity import sha256
class EditingReleaseGateTests(unittest.TestCase):
def fixture(self, directory, case_count=5, seeds=DEFAULT_SEEDS):
root = Path(directory)
suite = root/'evaluation'/'editing-v2'
inputs = suite/'inputs'
inputs.mkdir(parents=True)
source_root = root/'scripts'
source_root.mkdir()
hashes = {}
for name in ('benchmark_edits_v2.py', 'editing_protocol.py', 'benchmark.py', 'runtime.py'):
(source_root/name).write_text('# CPU release validation fixture\n')
hashes[name] = sha256(source_root/name)
definitions = Path(__file__).resolve().parents[1]/'benchmarks'/'editing-v2.json'
cases = json.loads(definitions.read_text(encoding='utf-8'))
for case in cases:
Image.new('RGBA', (32, 32), 'white').save(inputs/case['input_file'])
case['input_sha256'] = sha256(inputs/case['input_file'])
(root/'benchmarks').mkdir()
(root/'benchmarks'/'editing-v2.json').write_text(json.dumps(cases), encoding='utf-8')
selected = cases[:case_count]
image = Image.new('RGBA', (1024, 1024), 'white')
diagnostics = image_diagnostics(image)
isolation = []
for index, precision in enumerate(('bf16', 'int8')):
run = suite/precision
run.mkdir()
image.save(run/'result.png')
(run/'cases.json').write_text(json.dumps(selected), encoding='utf-8')
env = dict(
protocol='editing-v2', seeds=list(seeds), steps=40, resolution=1024,
kv_cache=True, vae_tiling=False, offload='model', offload_aux_fix=True,
warmup=True, generator_device='cpu', pid=index+100,
before_load_memory={'allocated_bytes': 0},
cases_sha256=sha256(run/'cases.json'),
model_identity={'kind': precision, 'base_revision': 'fixture',
'fingerprint': precision},
gpu='fixture', cuda='fixture', packages={}, source_sha256=hashes,
upstream_source_sha256={}, scheduler_config={}, vae_dtype='torch.bfloat16',
vae_tiles={})
(run/'environment.json').write_text(json.dumps(env), encoding='utf-8')
rows = []
for case in selected:
for seed in seeds:
rows.append(dict(
protocol='editing-v2', case_id=case['id'], prompt=case['prompt'],
source_seed=case['source_seed'], seed=seed,
input_file=case['input_file'], input_sha256=case['input_sha256'],
width=1024, height=1024, output_resolution=1024, steps=40, cfg=1.,
kv_cache=True, vae_tiling=False, offload='model',
image='result.png', image_sha256=sha256(run/'result.png'),
sigmas=[1., 0.], **diagnostics))
(run/'records.jsonl').write_text(
''.join(json.dumps(row)+'\n' for row in rows), encoding='utf-8')
(run/'COMPLETE.json').write_text(json.dumps(dict(
records=len(rows), records_sha256=sha256(run/'records.jsonl'))))
isolation.append(dict(precision=precision, pid=env['pid'], exit_code=0,
started=index*2, ended=index*2+1))
(suite/'process-isolation.jsonl').write_text(
''.join(json.dumps(row)+'\n' for row in isolation), encoding='utf-8')
return suite, source_root
def test_accepts_complete_release_suite(self):
with tempfile.TemporaryDirectory() as directory:
suite, source_root = self.fixture(directory)
runs = validate_suite(suite, source_root=source_root)
self.assertEqual(len(runs['bf16'][1]), 10)
self.assertEqual(len(runs['int8'][1]), 10)
def test_windows_launcher_pid_can_differ_from_recorded_inference_pid(self):
with tempfile.TemporaryDirectory() as directory:
suite, source_root = self.fixture(directory)
path = suite/'process-isolation.jsonl'
rows = [json.loads(line) for line in path.read_text().splitlines()]
for row in rows:
row['inference_pid'] = row['pid']
row['pid'] += 1000
path.write_text(''.join(json.dumps(row)+'\n' for row in rows))
self.assertEqual(len(validate_suite(suite, source_root)['int8'][1]), 10)
rows[0]['inference_pid'] = 99999
path.write_text(''.join(json.dumps(row)+'\n' for row in rows))
with self.assertRaises(ValueError): validate_suite(suite, source_root)
def test_rejects_partial_case_matrix_for_release(self):
with tempfile.TemporaryDirectory() as directory:
suite, source_root = self.fixture(directory, case_count=1)
with self.assertRaises(ValueError):
validate_suite(suite, source_root=source_root)
def test_rejects_single_seed_matrix_for_release(self):
with tempfile.TemporaryDirectory() as directory:
suite, source_root = self.fixture(directory, seeds=DEFAULT_SEEDS[:1])
with self.assertRaises(ValueError):
validate_suite(suite, source_root=source_root)
def test_partial_suite_remains_usable_for_diagnostics(self):
with tempfile.TemporaryDirectory() as directory:
suite, _ = self.fixture(directory, case_count=1, seeds=DEFAULT_SEEDS[:1])
runs = validate_suite(suite)
self.assertEqual(len(runs['bf16'][1]), 1)
def test_rejects_empty_measured_source_inventory_for_release(self):
with tempfile.TemporaryDirectory() as directory:
suite, source_root = self.fixture(directory)
for precision in ('bf16', 'int8'):
path = suite/precision/'environment.json'
env = json.loads(path.read_text(encoding='utf-8'))
env['source_sha256'] = {}
path.write_text(json.dumps(env), encoding='utf-8')
with self.assertRaises(ValueError):
validate_suite(suite, source_root=source_root)
if __name__ == '__main__':
unittest.main()
|