| """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() |
|
|
