| """Publication gates should reject incomplete or mismatched measurements.""" |
| import importlib.util |
| import contextlib |
| import io |
| import json |
| import sys |
| import tempfile |
| import types |
| import unittest |
| from unittest import mock |
| from pathlib import Path |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| spec = importlib.util.spec_from_file_location('publish', ROOT / 'scripts/publish.py') |
| publish = importlib.util.module_from_spec(spec) |
| spec.loader.exec_module(publish) |
|
|
| class PublishTests(unittest.TestCase): |
| def fixture(self, root): |
| data, adapter, evaluation = root/'data', root/'adapter', root/'eval' |
| for d in (data, adapter, evaluation): d.mkdir() |
| prompts = [{'id': str(i), 'instruction': f'independent instruction {i}'} for i in range(500)] |
| (data/'eval_prompts.jsonl').write_text(''.join(json.dumps(x)+'\n' for x in prompts)) |
| (adapter/'adapter_model.safetensors').write_bytes(b'fixture-only') |
| manifest = {'smoke_test': False, 'expected_count': 500, 'run_seconds': 60, 'sample_indices': [0, 200, 400], |
| 'prompts_sha256': publish.sha256(data/'eval_prompts.jsonl'), |
| 'generation': {'adapter': {'files': {'adapter_model.safetensors': publish.sha256(adapter/'adapter_model.safetensors')}}}} |
| (evaluation/'manifest.json').write_text(json.dumps(manifest)) |
| 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': 0, 'compile_success_rate': 0, |
| 'clean_run_60s_count': 0, 'clean_run_60s_rate': 0} |
| path = evaluation/'summary.json'; path.write_text(json.dumps(summary)) |
| for i in range(500): |
| case = evaluation/'items'/f'{i:04d}';case.mkdir(parents=True) |
| (case/'result.json').write_text(json.dumps({'index':i, 'compile_success':False, 'clean_run_60s':False})) |
| (case/'attempt.json').write_text(json.dumps({'index':i, 'instruction':prompts[i]['instruction']})) |
| return path, data, adapter, summary |
|
|
| def test_complete_zero_score_is_honest_and_permitted(self): |
| with tempfile.TemporaryDirectory() as temp: |
| path, data, adapter, _ = self.fixture(Path(temp)) |
| summary, _ = publish.inspect_evaluation(path, data, adapter) |
| self.assertEqual(summary['compile_success_rate'], 0) |
|
|
| def test_incomplete_summary_and_changed_adapter_block(self): |
| with tempfile.TemporaryDirectory() as temp: |
| path, data, adapter, summary = self.fixture(Path(temp)) |
| summary['recorded_prompts'] = 499 |
| path.write_text(json.dumps(summary)) |
| with self.assertRaisesRegex(ValueError, 'complete 500'): |
| publish.inspect_evaluation(path, data, adapter) |
| summary['recorded_prompts'] = 500 |
| path.write_text(json.dumps(summary)) |
| (adapter/'adapter_model.safetensors').write_bytes(b'different checkpoint') |
| with self.assertRaisesRegex(ValueError, 'does not match'): |
| publish.inspect_evaluation(path, data, adapter) |
|
|
| def test_cherry_picked_summary_or_prompt_order_blocks(self): |
| with tempfile.TemporaryDirectory() as temp: |
| path, data, adapter, summary = self.fixture(Path(temp)) |
| summary['compile_success_rate'] = 0.9 |
| path.write_text(json.dumps(summary)) |
| with self.assertRaisesRegex(ValueError, 'disagree'): |
| publish.inspect_evaluation(path, data, adapter) |
| summary['compile_success_rate'] = 0 |
| path.write_text(json.dumps(summary)) |
| attempt = path.parent/'items/0000/attempt.json' |
| attempt.write_text(json.dumps({'index':0, 'instruction':'swapped prompt'})) |
| with self.assertRaisesRegex(ValueError, 'ordering'): |
| publish.inspect_evaluation(path, data, adapter) |
|
|
| def test_resumed_check_is_allowed_but_traversal_pointer_is_not(self): |
| with tempfile.TemporaryDirectory() as temp: |
| path, data, adapter, summary = self.fixture(Path(temp)) |
| item = path.parent/'items/0000' |
| check = item/'check-resume-001' |
| check.mkdir() |
| (item/'check').mkdir() |
| (check/'import.log').write_text('Godot imported successfully') |
| (check/'runtime.log').write_text('Godot running') |
| proof = {'compile_success':True, 'clean_run_60s':True, |
| 'runtime': {'wall_seconds':60.1, 'log_sha256':publish.sha256(check/'runtime.log')}} |
| (check/'validation.json').write_text(json.dumps(proof)) |
| result = {'index':0, 'compile_success':True, 'clean_run_60s':True, 'validation_evidence':'check-resume-001'} |
| (item/'result.json').write_text(json.dumps(result)) |
| summary.update({'compile_success_count':1, 'compile_success_rate':.002, |
| 'clean_run_60s_count':1, 'clean_run_60s_rate':.002}) |
| path.write_text(json.dumps(summary)) |
| publish.inspect_evaluation(path, data, adapter) |
| result['validation_evidence'] = '../check-resume-001' |
| (item/'result.json').write_text(json.dumps(result)) |
| with self.assertRaisesRegex(ValueError, 'Unsafe'): |
| publish.inspect_evaluation(path, data, adapter) |
|
|
| def test_changing_predeclared_samples_blocks_publication(self): |
| with tempfile.TemporaryDirectory() as temp: |
| path, data, adapter, summary = self.fixture(Path(temp)) |
| summary['sample_indices'] = [0, 1, 2] |
| path.write_text(json.dumps(summary)) |
| with self.assertRaisesRegex(ValueError, 'predeclared'): |
| publish.inspect_evaluation(path, data, adapter) |
|
|
| def test_missing_mapping_is_not_empty(self): |
| class API: |
| def __init__(self, info): self.info = info |
| def model_info(self, **kwargs): return self.info |
| self.assertEqual(publish.provider_mapping(API(types.SimpleNamespace(inference_provider_mapping={})), 'u/repo'), {}) |
| with self.assertRaisesRegex(RuntimeError, 'cannot verify'): |
| publish.provider_mapping(API(types.SimpleNamespace()), 'u/repo') |
|
|
| def demo_fixture(self, root): |
| evaluation = root / 'evaluation' |
| evaluation.mkdir() |
| (evaluation / 'manifest.json').write_text(json.dumps({'run': 'current frozen evaluation'})) |
| summary = evaluation / 'summary.json' |
| video = root / 'VibeGame-launch-60s.mp4' |
| video.write_bytes(b'test fixture; not a real launch asset') |
| sidecar = video.with_suffix('.demo.json') |
| evidence = {'kind': 'model-evaluation', 'scaffold_preview': False, |
| 'output_sha256': publish.sha256(video), 'total_seconds': 60, |
| 'source_plan': {'kind': 'model-evaluation', 'scaffold_preview': False, |
| 'evaluation_manifest_sha256': publish.sha256(evaluation / 'manifest.json'), |
| 'samples': [{'index': 0, 'capture': {'verified': True}}, |
| {'index': 200, 'capture': None, 'diagnostic': 'Parse failure'}, |
| {'index': 400, 'capture': None, 'diagnostic': 'Runtime failure'}]}} |
| sidecar.write_text(json.dumps(evidence)) |
| return summary, video, sidecar, evidence |
|
|
| def test_demo_allows_fixed_sample_failure_cards_but_requires_files(self): |
| with tempfile.TemporaryDirectory() as temp: |
| summary, video, sidecar, _ = self.demo_fixture(Path(temp)) |
| self.assertEqual(publish.inspect_demo(summary), (video, sidecar)) |
| sidecar.unlink() |
| with self.assertRaisesRegex(ValueError, 'requires the launch video'): |
| publish.inspect_demo(summary) |
| video.unlink() |
| with self.assertRaisesRegex(ValueError, 'requires the launch video'): |
| publish.inspect_demo(summary) |
|
|
| def test_stale_demo_and_changed_video_block_publication(self): |
| with tempfile.TemporaryDirectory() as temp: |
| summary, video, sidecar, evidence = self.demo_fixture(Path(temp)) |
| evidence['source_plan']['evaluation_manifest_sha256'] = 'previous evaluation' |
| sidecar.write_text(json.dumps(evidence)) |
| with self.assertRaisesRegex(ValueError, 'different evaluation'): |
| publish.inspect_demo(summary) |
| evidence['source_plan']['evaluation_manifest_sha256'] = publish.sha256(summary.parent / 'manifest.json') |
| sidecar.write_text(json.dumps(evidence)) |
| video.write_bytes(b'changed movie') |
| with self.assertRaisesRegex(ValueError, 'matching final-model evidence'): |
| publish.inspect_demo(summary) |
|
|
| def test_demo_rejects_substitution_preview_and_diagnostic_only_video(self): |
| with tempfile.TemporaryDirectory() as temp: |
| summary, _, sidecar, evidence = self.demo_fixture(Path(temp)) |
| for replacement, message in [ |
| ({'samples': [{'index': index, 'capture': {'verified': True}} for index in [0, 1, 2]]}, 'predeclared'), |
| ({'scaffold_preview': True}, 'scaffold preview'), |
| ({'samples': [{'index': index, 'capture': None} for index in [0, 200, 400]]}, 'actual gameplay')]: |
| original = dict(evidence['source_plan']) |
| evidence['source_plan'].update(replacement) |
| sidecar.write_text(json.dumps(evidence)) |
| with self.subTest(replacement=replacement), self.assertRaisesRegex(ValueError, message): |
| publish.inspect_demo(summary) |
| evidence['source_plan'] = original |
|
|
|
|
| class ReleaseEvidenceTests(unittest.TestCase): |
| def fixture(self): |
| fixture_spec = importlib.util.spec_from_file_location('pro_fixture', ROOT / 'tests/test_evaluate_pro.py') |
| fixture_module = importlib.util.module_from_spec(fixture_spec) |
| fixture_spec.loader.exec_module(fixture_module) |
| fixture = fixture_module.ProEvaluationTests() |
| fixture.setUp() |
| self.addCleanup(fixture.tearDown) |
| fixture.output.mkdir() |
| protocol = fixture.protocol() |
| (fixture.output / 'pro-protocol.json').write_text(json.dumps({ |
| 'declared_utc': '2026-10-04T14:00:00+00:00', 'protocol': protocol})) |
| manifest = {**fixture.prior_manifest, 'generation': {**fixture.prior, 'gpu': fixture.runtime['gpu']}} |
| (fixture.output / 'manifest.json').write_text(json.dumps(manifest)) |
| (fixture.output / 'summary.json').write_text(json.dumps({'status': 'complete'})) |
| (fixture.t4 / 'summary.json').write_text(json.dumps({'status': 'incomplete', 'recorded_prompts': 72})) |
| for directory in (fixture.output, fixture.t4): |
| item = directory / 'items/0000' |
| item.mkdir(parents=True) |
| (item / 'attempt.json').write_text(json.dumps({'index': 0, 'started_utc': '2026-10-04T14:10:00+00:00'})) |
| (item / 'result.json').write_text(json.dumps({'index': 0, 'compile_success': False})) |
| capacity_path = fixture.output / 'capacity-0000.json' |
| capacity_path.write_text(json.dumps(fixture.capacity())) |
| budget_path = fixture.output / 'budget-0000.json' |
| budget_path.write_text(json.dumps({'observed_utc': '2026-10-04T14:05:00+00:00', |
| 'capacity_report_sha256': publish.sha256(capacity_path), 'fits_observed_balance': True})) |
| return fixture |
|
|
| def test_a100_receipts_and_t4_evidence_are_separate_and_unchanged(self): |
| fixture = self.fixture() |
| old = {str(path.relative_to(fixture.t4)): path.read_bytes() for path in fixture.t4.rglob('*') if path.is_file()} |
| (fixture.output / 'unrelated.json').write_text('{"not_for_release": true}') |
| uploads = publish.evaluation_uploads(fixture.output / 'summary.json', fixture.data, fixture.adapter) |
| self.assertEqual([entry['prefix'] for entry in uploads], ['evaluation', 'evaluation-t4-partial']) |
| self.assertTrue({'pro-protocol.json', 'capacity-0000.json', 'budget-0000.json'} <= uploads[0]['files'].keys()) |
| self.assertNotIn('unrelated.json', uploads[0]['files']) |
| with tempfile.TemporaryDirectory() as temp: |
| destination = Path(temp) |
| publish.stage_evaluation_upload(uploads[0], destination) |
| declaration = json.loads((destination / 'pro-protocol.json').read_text()) |
| self.assertEqual(declaration['protocol']['source_t4_manifest_sha256'], publish.sha256(fixture.t4 / 'manifest.json')) |
| self.assertTrue((destination / 'REDACTIONS.json').is_file()) |
| self.assertEqual(old, {str(path.relative_to(fixture.t4)): path.read_bytes() for path in fixture.t4.rglob('*') if path.is_file()}) |
|
|
| def test_changed_source_or_capacity_and_late_declaration_are_rejected(self): |
| fixture = self.fixture() |
| summary = fixture.output / 'summary.json' |
| budget_path = fixture.output / 'budget-0000.json' |
| original = budget_path.read_bytes() |
| budget = json.loads(original); budget['capacity_report_sha256'] = 'wrong-report' |
| budget_path.write_text(json.dumps(budget)) |
| with self.assertRaisesRegex(ValueError, 'matching valid capacity'): |
| publish.evaluation_uploads(summary, fixture.data, fixture.adapter) |
| budget_path.write_bytes(original) |
| declaration_path = fixture.output / 'pro-protocol.json' |
| declaration = json.loads(declaration_path.read_text()) |
| declaration['declared_utc'] = '2026-10-04T15:00:00+00:00' |
| declaration_path.write_text(json.dumps(declaration)) |
| with self.assertRaisesRegex(ValueError, 'after blind'): |
| publish.evaluation_uploads(summary, fixture.data, fixture.adapter) |
| declaration['declared_utc'] = '2026-10-04T14:00:00+00:00' |
| declaration_path.write_text(json.dumps(declaration)) |
| (fixture.t4 / 'manifest.json').write_text('{}') |
| with self.assertRaisesRegex(ValueError, 'selected experiment or T4'): |
| publish.evaluation_uploads(summary, fixture.data, fixture.adapter) |
|
|
| def test_credentials_and_changed_preflight_files_stop_before_upload(self): |
| fixture = self.fixture() |
| log = fixture.output / 'items/0000/import.log' |
| log.write_text('auth=' + 'hf_' + 'x' * 30) |
| with self.assertRaisesRegex(ValueError, 'possible credential'): |
| publish.evaluation_uploads(fixture.output / 'summary.json', fixture.data, fixture.adapter) |
| log.write_text('Godot normal output /content/drive/MyDrive/example\n') |
| uploads = publish.evaluation_uploads(fixture.output / 'summary.json', fixture.data, fixture.adapter) |
| with tempfile.TemporaryDirectory() as temp: |
| destination = Path(temp) |
| publish.stage_evaluation_upload(uploads[0], destination) |
| self.assertIn('/persistent-drive/example', (destination / 'items/0000/import.log').read_text()) |
| self.assertIn('items/0000/import.log', json.loads((destination / 'REDACTIONS.json').read_text())) |
| log.write_text('changed after inspection') |
| with tempfile.TemporaryDirectory() as temp, self.assertRaisesRegex(ValueError, 'changed after'): |
| publish.stage_evaluation_upload(uploads[0], Path(temp)) |
| with self.assertRaisesRegex(ValueError, 'credential field'): |
| publish.public_evidence_text('{"nested": {"access_token": "test-secret"}}') |
|
|
| def test_regular_t4_upload_keeps_existing_prefix_without_pro_receipts(self): |
| fixture = self.fixture() |
| uploads = publish.evaluation_uploads(fixture.t4 / 'summary.json', fixture.data, fixture.adapter) |
| self.assertEqual([entry['prefix'] for entry in uploads], ['evaluation']) |
| self.assertNotIn('pro-protocol.json', uploads[0]['files']) |
|
|
| def test_model_card_uses_selected_hardware_and_discloses_separate_run(self): |
| fixture = self.fixture() |
| evidence = {'dataset': {'train': 97, 'dev': 11}, |
| 'evaluation_manifest': json.loads((fixture.output / 'manifest.json').read_text()), |
| 'evaluation': {'compile_success_rate': .002, 'clean_run_60s_rate': .002, |
| 'compile_success_count': 1, 'clean_run_60s_count': 1, |
| 'under_60s_launch_count': 0, 'prompt_to_launch_median_seconds': 90}} |
| repos = {key: 'fixture/' + key for key in ('merged', 'gguf', 'adapter', 'dataset')} |
| card = publish.model_card(repos, evidence, ROOT / 'docs/model_card.md') |
| self.assertIn('NVIDIA A100-SXM4-40GB', card) |
| self.assertIn('separate full A100 pass on Colab Pro', card) |
| self.assertIn('evaluation-t4-partial/', card) |
| self.assertNotIn('{{', card) |
|
|
|
|
| class PublicationFlowTests(unittest.TestCase): |
| def test_all_four_folder_uploads_resume_after_augmented_merged_release(self): |
| fixture_spec = importlib.util.spec_from_file_location('stage_fixture', ROOT / 'tests/test_stage_exports.py') |
| fixture_module = importlib.util.module_from_spec(fixture_spec) |
| fixture_spec.loader.exec_module(fixture_module) |
| fixture = fixture_module.StagingTests() |
| fixture.setUp() |
| self.addCleanup(fixture.tearDown) |
| fixture.stage_all() |
| root, api, stage = fixture.root, fixture.api, fixture_module.stage |
| api.whoami = lambda: {'name': 'khursheed', 'orgs': []} |
| api.list_inference_endpoints = lambda **kwargs: [] |
| folder_calls, fail_once = [], [True] |
| upload_folder = api.upload_folder |
| def record_folder(**kwargs): |
| if not kwargs.get('path_in_repo'): |
| folder_calls.append(kwargs['repo_id']) |
| if kwargs['repo_id'].endswith('-GGUF') and fail_once[0]: |
| fail_once[0] = False |
| raise RuntimeError('simulated interruption before second artifact') |
| return upload_folder(**kwargs) |
| api.upload_folder = record_folder |
| adapter, data = root / 'adapter', root / 'data' |
| adapter.mkdir(); data.mkdir() |
| (adapter / 'adapter_config.json').write_text('{}') |
| (data / 'train.jsonl').write_text('{}\n') |
| (root / 'README.md').write_text('Release source fixture') |
| video, sidecar = root / 'video.mp4', root / 'video.demo.json' |
| video.write_bytes(b'fixture footage') |
| sidecar.write_text(json.dumps({'kind': 'model-evaluation'})) |
| output = root / 'publish-report.json' |
| argv = ['publish.py', '--adapter', str(adapter), '--merged', str(fixture.roots['merged']), |
| '--gguf', str(fixture.roots['gguf']), '--data', str(data), '--evaluation', str(root / 'summary.json'), |
| '--output', str(output), '--namespace', 'khursheed'] |
| repos = ['khursheed/VibeGame-8B', 'khursheed/VibeGame-8B-GGUF', |
| 'khursheed/VibeGame-8B-LoRA', 'khursheed/VibeGame-8B-data'] |
| with mock.patch.object(sys, 'argv', argv), \ |
| mock.patch.dict(sys.modules, {'huggingface_hub': types.SimpleNamespace(HfApi=lambda **kwargs: api), 'stage_exports': stage}), \ |
| mock.patch.dict(publish.os.environ, {'HF_TOKEN': 'not-a-real-test-token'}), \ |
| mock.patch.object(publish, 'ROOT', root), \ |
| mock.patch.object(publish, 'inspect_artifacts', return_value={'adapter_sha256': 'fixture-adapter', 'gguf_sha256': 'fixture-gguf'}), \ |
| mock.patch.object(publish, 'prepare_gguf_licenses', return_value={}), \ |
| mock.patch.object(publish, 'license_files', return_value={}), \ |
| mock.patch.object(publish, 'inspect_demo', return_value=(video, sidecar)), \ |
| mock.patch.object(publish, 'evaluation_uploads', return_value=[]), \ |
| mock.patch.object(publish, 'model_card', return_value='Actual model card fixture'), \ |
| mock.patch.object(publish, 'sha256', return_value='fixture-hash'), \ |
| mock.patch.object(stage, 'build_manifests', return_value=fixture.manifests), \ |
| mock.patch('subprocess.run'), contextlib.redirect_stdout(io.StringIO()): |
| with self.assertRaisesRegex(RuntimeError, 'Publication stopped'): |
| publish.main() |
| failed = json.loads(output.read_text()) |
| self.assertEqual(failed['failure_type'], 'RuntimeError') |
| self.assertEqual(failed['uploaded'], repos[:1]) |
| self.assertEqual(folder_calls, repos[:2]) |
| self.assertTrue(api.repos[repos[0]]['private']) |
| self.assertIn('code/README.md', api.repos[repos[0]]['files']) |
| for repo in repos[2:]: |
| self.assertEqual(set(api.repos[repo]['files']), {'.gitattributes'}) |
| self.assertIn(repo, failed['publication_ownership']) |
| folder_calls.clear() |
| publish.main() |
| self.assertEqual(folder_calls, repos) |
| completed = json.loads(output.read_text()) |
| self.assertEqual(completed['status'], 'artifacts_uploaded_account_reverification_required') |
| self.assertEqual(completed['uploaded'], repos) |
| self.assertTrue(all(not api.repos[repo]['private'] for repo in repos)) |
| for repo in repos: |
| self.assertEqual(api.repos[repo]['mapping'], {}) |
|
|
| if __name__ == '__main__': unittest.main() |
|
|
