File size: 14,285 Bytes
ae36cfb | 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 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 | """Success-only showcase checks; no Godot/model execution or network requests."""
import hashlib
import importlib.util
import io
import json
import os
import shutil
import statistics
import subprocess
import sys
import tempfile
import unittest
from contextlib import redirect_stderr, redirect_stdout
from pathlib import Path
from unittest.mock import patch
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / 'scripts'))
import build_demo as demo
import publish
import record_game
import showcase
FILES = {'project.godot': 'config_version=5\n[application]\nrun/main_scene="res://main.tscn"\n',
'main.tscn': '[gd_scene load_steps=2 format=3]\n[ext_resource type="Script" path="res://main.gd" id="1"]\n[node name="Main" type="Node2D"]\nscript = ExtResource("1")\n',
'main.gd': 'extends Node2D\nfunc _draw():\n draw_circle(Vector2(100, 100), 25, Color.CYAN)\n'}
def write(path, value):
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(value))
class ShowcaseTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.addCleanup(self.temp.cleanup)
self.root = Path(self.temp.name)
self.evaluation = self.root / 'evaluation'
self.godot = self.root / 'not-an-executable'
self.godot.write_bytes(b'unit-fixture-engine-not-executable')
self.clean = [2, 7, 11, 23]
self.manifest = {'expected_count': 500, 'run_seconds': 60, 'smoke_test': False,
'sample_indices': [0, 200, 400], 'runtime_sha256': showcase.sha256(self.godot),
'generation': {'backend': 'unit-test-only', 'gpu': 'fixture GPU'}}
write(self.evaluation / 'manifest.json', self.manifest)
project_hash = showcase.project_sha256(FILES)
for index in range(500):
item = self.evaluation / 'items' / f'{index:04d}'
write(item / 'attempt.json', {'index': index, 'prompt_id': f'prompt-{index}', 'instruction': f'Fixture prompt {index}'})
result = {'index': index, 'prompt_id': f'prompt-{index}', 'compile_success': index in self.clean,
'clean_run_60s': index in self.clean, 'run_seconds': 60, 'infrastructure_failure': False,
'project_sha256': project_hash, 'prompt_to_launch_seconds': 700 + index if index in self.clean else None,
'validation_evidence': 'check'}
write(item / 'result.json', result)
if index in self.clean:
output = json.dumps({'files': FILES})
write(item / 'generation.json', {'output': output, 'output_sha256': hashlib.sha256(output.encode()).hexdigest(),
'generation_seconds': 650 + index, 'generation_call_seconds': 660 + index})
check = item / 'check'; check.mkdir()
for phase in ('import', 'runtime'):
(check / (phase + '.log')).write_text('Godot unit fixture, not an actual engine run')
write(check / 'validation.json', {'compile_success': True, 'clean_run_60s': True, 'run_seconds': 60,
'project_sha256': project_hash, 'godot_binary_sha256': self.manifest['runtime_sha256'],
'import': {'log_sha256': showcase.sha256(check / 'import.log')},
'runtime': {'wall_seconds': 60.1, 'log_sha256': showcase.sha256(check / 'runtime.log')}})
self.summary = {'status': 'complete', 'expected_prompts': 500, 'recorded_prompts': 500,
'infrastructure_failures': 0, 'run_seconds': 60, 'pass_k': 1, 'retries': 0,
'sample_indices': [0, 200, 400], 'compile_success_count': 4, 'compile_success_rate': .008,
'clean_run_60s_count': 4, 'clean_run_60s_rate': .008,
'prompt_to_launch_median_seconds': statistics.median(700 + i for i in self.clean),
'under_60s_launch_count': 0}
write(self.evaluation / 'summary.json', self.summary)
def captures(self):
selection = showcase.ensure_selection(self.evaluation)
for index in self.clean[:3]:
video = self.evaluation / 'showcase/recordings' / f'sample-{index:03d}.mp4'
video.parent.mkdir(parents=True, exist_ok=True)
video.write_bytes(b'unit fixture only; ffprobe is mocked')
write(video.with_suffix('.recording.json'), {
'kind': 'model-success-showcase', 'showcase_policy': showcase.POLICY,
'showcase_selection_sha256': showcase.sha256(selection), 'godot_binary_sha256': self.manifest['runtime_sha256'],
'capture_success': True, 'duration_seconds': 15, 'fixed_timestep': True, 'player_input': False,
'sample_index': index, 'prompt_id': f'prompt-{index}', 'project_sha256': showcase.project_sha256(FILES),
'evaluation_manifest_sha256': showcase.sha256(self.evaluation / 'manifest.json'),
'generation_sha256': showcase.sha256(self.evaluation / 'items' / f'{index:04d}' / 'generation.json'),
'output_sha256': showcase.sha256(video)})
return selection
def test_selects_first_three_in_frozen_order_and_keeps_benchmark_slots(self):
selection = showcase.ensure_selection(self.evaluation)
value = showcase.load_selection(self.evaluation, selection)
self.assertEqual(value['showcase_indices'], [2, 7, 11])
self.assertEqual(value['benchmark_sample_indices'], [0, 200, 400])
self.assertEqual(value['metrics']['clean_run_60s_rate'], .008)
self.assertEqual(value['metrics']['denominator'], 500)
self.assertIn('after the A100 experiment began', value['policy_timing'])
before = selection.read_bytes()
showcase.ensure_selection(self.evaluation)
self.assertEqual(selection.read_bytes(), before)
value['showcase_indices'] = [2, 11, 23]
write(selection, value)
with self.assertRaisesRegex(ValueError, 'selection or its frozen inputs changed'):
showcase.load_selection(self.evaluation, selection)
def test_partial_infrastructure_and_fewer_than_three_block_selection(self):
path = self.evaluation / 'summary.json'
for changed in [{'recorded_prompts': 499}, {'infrastructure_failures': 1}]:
write(path, {**self.summary, **changed})
with self.assertRaisesRegex(ValueError, 'complete unchanged 500'):
showcase.ensure_selection(self.evaluation)
write(path, self.summary)
(self.evaluation / 'items/0499/result.json').unlink()
with self.assertRaisesRegex(ValueError, 'exactly all 500'):
showcase.ensure_selection(self.evaluation)
def test_changed_nonselected_result_or_selected_log_invalidates_receipt(self):
receipt = showcase.ensure_selection(self.evaluation)
path = self.evaluation / 'items/0100/result.json'
result = json.loads(path.read_text()); result['error'] = 'changed after selection'; write(path, result)
with self.assertRaisesRegex(ValueError, 'frozen inputs changed'):
showcase.load_selection(self.evaluation, receipt)
(self.evaluation / 'items/0002/check/runtime.log').write_text('tampered log')
with self.assertRaisesRegex(ValueError, 'log hash differs'):
showcase.selection_for(self.evaluation)
def test_fewer_than_three_successes_never_substitutes_seed_games(self):
for index in (11, 23):
path = self.evaluation / 'items' / f'{index:04d}' / 'result.json'
result = json.loads(path.read_text()); result.update(compile_success=False, clean_run_60s=False)
write(path, result)
write(self.evaluation / 'summary.json', {**self.summary, 'compile_success_count': 2, 'clean_run_60s_count': 2,
'compile_success_rate': .004, 'clean_run_60s_rate': .004})
with self.assertRaisesRegex(ValueError, 'Fewer than three clean'):
showcase.ensure_selection(self.evaluation)
self.assertFalse(showcase.selection_path(self.evaluation).exists())
@patch.object(demo, 'probe_video', return_value={'duration_seconds': 15, 'width': 960, 'height': 540})
def test_all_three_original_captures_required_and_no_later_replacement(self, probe):
receipt = self.captures()
plan = demo.evaluation_plan(self.evaluation, showcase_selection=receipt)
self.assertEqual(plan['kind'], 'model-success-showcase')
self.assertEqual([row['index'] for row in plan['samples']], [2, 7, 11])
self.assertEqual(plan['metrics']['clean_run_60s_rate'], .008)
self.assertIn('662.0s', demo.latency_text(plan['samples'][0]))
self.assertFalse(plan['includes_failed_or_unrecorded_samples'])
self.assertTrue(all(row['capture'] for row in plan['samples']))
(self.evaluation / 'showcase/recordings/sample-011.mp4').unlink()
with self.assertRaisesRegex(ValueError, 'All three selected successes'):
demo.evaluation_plan(self.evaluation, showcase_selection=receipt)
with self.assertRaisesRegex(ValueError, 'At least one real recording'):
demo.evaluation_plan(self.evaluation)
def test_recorder_only_accepts_selected_index_and_adds_selection_identity(self):
receipt = showcase.ensure_selection(self.evaluation)
argv = ['record_game.py', '--evaluation', str(self.evaluation), '--showcase-selection', str(receipt),
'--index', '2', '--godot', str(self.godot), '--output', str(self.evaluation / 'showcase/recordings/sample-002.mp4')]
with patch.object(sys, 'argv', argv), patch.object(record_game, 'record', return_value={}) as record, redirect_stdout(io.StringIO()):
record_game.main()
metadata = record.call_args.kwargs['generation_metadata']
self.assertEqual(metadata['kind'], 'model-success-showcase')
self.assertEqual(metadata['showcase_selection_sha256'], showcase.sha256(receipt))
argv[argv.index('--index') + 1] = '23'
with patch.object(sys, 'argv', argv), patch.object(record_game, 'record') as record, redirect_stderr(io.StringIO()):
with self.assertRaises(SystemExit):
record_game.main()
record.assert_not_called()
@patch.object(demo, 'probe_video', return_value={'duration_seconds': 15, 'width': 960, 'height': 540})
def test_publisher_recomputes_selection_and_all_capture_evidence(self, probe):
receipt = self.captures()
plan = demo.evaluation_plan(self.evaluation, showcase_selection=receipt)
video = self.root / 'VibeGame-launch-60s.mp4'; video.write_bytes(b'unit final-video placeholder')
evidence = {'kind': 'model-success-showcase', 'scaffold_preview': False, 'output_sha256': showcase.sha256(video),
'total_seconds': 60, 'source_plan': plan, 'watermark': showcase.DISCLOSURE}
sidecar = video.with_suffix('.demo.json'); write(sidecar, evidence)
self.assertEqual(publish.inspect_demo(self.evaluation / 'summary.json'), (video, sidecar))
evidence['source_plan']['samples'][2]['index'] = 23; write(sidecar, evidence)
with self.assertRaisesRegex(ValueError, 'exact first three'):
publish.inspect_demo(self.evaluation / 'summary.json')
uploads = publish.evaluation_uploads(self.evaluation / 'summary.json', self.root, self.root)
self.assertIn('showcase/selection.json', uploads[0]['files'])
self.assertIn('showcase/recordings/sample-002.mp4', uploads[0]['files'])
@unittest.skipUnless(importlib.util.find_spec('PIL'), 'Pillow is required for rendering')
@patch.object(demo, 'probe_video', return_value={'duration_seconds': 15, 'width': 960, 'height': 540})
def test_every_frame_asset_has_selection_warning_full_rates_and_duration_caveat(self, probe):
receipt = self.captures()
plan = demo.evaluation_plan(self.evaluation, showcase_selection=receipt)
with patch.object(demo, 'block', wraps=demo.block) as block:
assets = demo.render_assets(plan, self.root / 'fixture-assets')
texts = [call.args[1] for call in block.call_args_list]
self.assertEqual(sum('SELECTED SUCCESSES' in text for text in texts), 6)
self.assertEqual(sum('All 500: compile 4/500 (0.8%); clean 60s 4/500 (0.8%)' in text for text in texts), 6)
self.assertEqual(sum('60-second edited video, not generation time' in text for text in texts), 6)
self.assertEqual(sum(row['seconds'] for row in assets), 60)
@unittest.skipUnless(os.environ.get('VIBEGAME_VIDEO_INTEGRATION') == '1' and importlib.util.find_spec('PIL')
and shutil.which('ffmpeg') and shutil.which('ffprobe'), 'Optional CPU-only fixture encoding check')
def test_showcase_fixture_encodes_actual_60_second_video(self):
receipt = self.captures()
color = self.root / 'synthetic-blue-test-source.mp4'
subprocess.run([shutil.which('ffmpeg'), '-nostdin', '-v', 'error', '-f', 'lavfi', '-i',
'color=c=blue:s=960x540:r=30:d=15', '-c:v', 'libx264', '-pix_fmt', 'yuv420p', str(color)], check=True)
for index in self.clean[:3]:
video = self.evaluation / 'showcase/recordings' / f'sample-{index:03d}.mp4'
shutil.copyfile(color, video)
sidecar = video.with_suffix('.recording.json')
metadata = json.loads(sidecar.read_text()); metadata['output_sha256'] = showcase.sha256(video)
write(sidecar, metadata)
plan = demo.evaluation_plan(self.evaluation, showcase_selection=receipt)
result = demo.encode_video(plan, self.root / 'synthetic-test-only-not-a-model-demo.mp4')
self.assertAlmostEqual(result['seconds'], 60, places=2)
receipt = json.loads(Path(result['evidence']).read_text())
self.assertEqual(receipt['watermark'], showcase.DISCLOSURE)
self.assertEqual(receipt['probe']['r_frame_rate'], '30/1')
self.assertEqual((receipt['probe']['width'], receipt['probe']['height']), (1280, 720))
if __name__ == '__main__':
unittest.main()
|