| """Lightweight synthetic tests: no data, GPU, Slurm, network, or model imports.""" |
| import importlib.util |
| import json |
| import hashlib |
| import sys |
| import tempfile |
| import types |
| import unittest |
| from unittest.mock import patch |
| from pathlib import Path |
|
|
| SPEC=importlib.util.spec_from_file_location('dyn24_builder',Path(__file__).with_name('build.py')) |
| builder=importlib.util.module_from_spec(SPEC) |
| SPEC.loader.exec_module(builder) |
| PUB_SPEC=importlib.util.spec_from_file_location('dyn24_publisher',Path(__file__).with_name('publish.py')) |
| publisher=importlib.util.module_from_spec(PUB_SPEC) |
| PUB_SPEC.loader.exec_module(publisher) |
| REPACK_SPEC=importlib.util.spec_from_file_location('dyn24_repackage',Path(__file__).with_name('repackage.py')) |
| repacker=importlib.util.module_from_spec(REPACK_SPEC) |
| with patch.dict(sys.modules,{'build':builder}): |
| REPACK_SPEC.loader.exec_module(repacker) |
|
|
|
|
| def dense(xy=(500,500),death=24,visible=True): |
| return dict(image_positions=[xy]*24,vis=[visible]*24,is_in_frame=[visible]*24,death=death) |
|
|
|
|
| class VisualizerTests(unittest.TestCase): |
| def test_score_is_twelve_term_mean(self): |
| metrics={key:{t:(i+1)/4 for i,t in enumerate(builder.THRESHOLDS)} for key in builder.FAMILIES.values()} |
| score,detail=builder.score_sample({'interaction_metrics':metrics}) |
| self.assertEqual(score,.625) |
| self.assertEqual(len(detail),3) |
|
|
| def test_selection_excludes_zero_and_starts_at_linear_quartile(self): |
| rows=[dict(key=str(i),scene_id=f's{i:03d}',split='val',score=i/144,parseable=True) for i in range(145)] |
| rows.append(dict(key='broken',scene_id='broken',split='val',score=.99,parseable=False)) |
| selected,rule=builder.select_cases(rows) |
| self.assertEqual(rule['nonzero_parseable'],144) |
| self.assertEqual(rule['failure_start_rank'],37) |
| self.assertEqual([r['key'] for r in selected[:10]],[str(i) for i in range(37,47)]) |
| self.assertEqual([r['key'] for r in selected[10:]],[str(i) for i in range(144,134,-1)]) |
| self.assertEqual(len({r['key'] for r in selected}),20) |
|
|
| def test_ties_use_scene_id(self): |
| rows=[dict(key=str(i),scene_id=f's{i:03d}',split='val',score=.5 if i<30 else .9,parseable=True) for i in reversed(range(50))] |
| selected,_=builder.select_cases(rows) |
| self.assertEqual(selected[0]['scene_id'],'s000') |
|
|
| def test_pairing_preserves_duplicate_predicted_ids(self): |
| gt=[{'id':0},{'id':1}];pred=[{'id':7},{'id':7},{'id':8}] |
| matching={'matches':[dict(target_index=0,pred_index=1,target_id=0,pred_id=7)]} |
| rows=builder.pair_objects(gt,pred,matching) |
| self.assertEqual([(r['gt_index'],r['pred_index']) for r in rows],[(0,1),(1,None),(None,0),(None,2)]) |
|
|
| def test_mismatched_id_is_rejected(self): |
| with self.assertRaises(ValueError): |
| builder.pair_objects([{'id':0}],[{'id':1}],{'matches':[dict(target_index=0,pred_index=0,target_id=5,pred_id=1)]}) |
|
|
| def test_corrected_caption_requires_hashed_source_evidence(self): |
| with tempfile.TemporaryDirectory() as tmp: |
| path=Path(tmp)/'instance.json' |
| path.write_text(json.dumps({'asset_id':'yellow mountain'})) |
| item=dict(sha256=builder.digest(path),old='yellow mountain',new='yellow mountain bicycle') |
| evidence={str(path.resolve()):item} |
| obj={'type':'yellow mountain bicycle'} |
| builder.validate_source_caption(path,obj,evidence) |
| self.assertEqual(obj['type'],'yellow mountain bicycle') |
| for caption in ['yellow mountain','yellow bike','yellow mountain bicycle bicycle']: |
| with self.subTest(caption=caption),self.assertRaises(ValueError): |
| builder.validate_source_caption(path,{'type':caption},evidence) |
| with self.assertRaises(ValueError): |
| builder.validate_source_caption(path,obj,{}) |
| path.write_text(json.dumps({'asset_id':'red mountain'})) |
| with self.assertRaises(ValueError): |
| builder.validate_source_caption(path,obj,evidence) |
|
|
| def test_legacy_caption_validation_unchanged(self): |
| with tempfile.TemporaryDirectory() as tmp: |
| path=Path(tmp)/'instance.json';path.write_text('{"asset_id":"yellow mountain"}') |
| builder.validate_source_caption(path,{'type':'yellow mountain'}) |
| with self.assertRaises(ValueError): |
| builder.validate_source_caption(path,{'type':'yellow mountain bicycle'}) |
|
|
| def test_caption_manifest_binds_exact_inputs_and_report(self): |
| with tempfile.TemporaryDirectory() as tmp: |
| root=Path(tmp) |
| for name in ['data.json','predictions.jsonl']: |
| (root/name).write_text('immutable input') |
| cell=dict(split='val',horizon=24,dataset='data.json',predictions='predictions.jsonl',output='report.json') |
| m=dict(cells=[cell],source_instances=[],input_sha256={name:builder.digest(root/name) for name in ['data.json','predictions.jsonl']}) |
| (root/'manifest.json').write_text(json.dumps(m)) |
| c=dict(caption_manifest=str(root/'manifest.json'),cells=[dict(split='val',dataset_json=str(root/'data.json'),predictions_jsonl=str(root/'predictions.jsonl'),report_json=str(root/'report.json'))]) |
| self.assertEqual(builder.caption_evidence(c),{}) |
| c['cells'][0]['report_json']=str(root/'wrong-report.json') |
| with self.assertRaises(ValueError):builder.caption_evidence(c) |
| c['cells'][0]['report_json']=str(root/'report.json') |
| (root/'data.json').write_text('changed input') |
| with self.assertRaises(ValueError):builder.caption_evidence(c) |
|
|
| def test_incomplete_report_cannot_change_selection_pool(self): |
| with tempfile.TemporaryDirectory() as tmp: |
| root=Path(tmp) |
| (root/'data.json').write_text('[{},{}]') |
| (root/'predictions.jsonl').write_text('{}\n{}\n') |
| (root/'report.json').write_text(json.dumps(dict(num_samples=1,successful_line_numbers=[1],samples=[{}]))) |
| c=dict(cells=[dict(dataset_json=str(root/'data.json'),predictions_jsonl=str(root/'predictions.jsonl'),report_json=str(root/'report.json'))]) |
| with self.assertRaisesRegex(ValueError,'complete evaluation'): |
| builder.load_cells(c,lambda _:None) |
|
|
| def test_occluded_gt_remains_and_pred_circle_disappears(self): |
| f=builder.frame_payload(dense((None,None)),dense(),[[.2,.3]]*24,[False]*24,5) |
| self.assertTrue(f['render_gt']) |
| self.assertFalse(f['render_pred']) |
| self.assertTrue(f['pred_visible']) |
| self.assertTrue(f['pred_hidden_by_gt']) |
| self.assertEqual(f['gt_xy'],[.2,.3]) |
| self.assertEqual(f['gt_state'],'occluded') |
| self.assertEqual(f['gt_position_source'],'source projection') |
|
|
| def test_visible_gt_keeps_label_position_not_projection(self): |
| f=builder.frame_payload(dense((600,400)),dense(),[[.2,.3]]*24,[True]*24,0) |
| self.assertEqual(f['gt_xy'],[.6,.4]) |
| self.assertTrue(f['render_pred']) |
|
|
| def test_death_mistake_is_warned_not_hidden(self): |
| f=builder.frame_payload(dense(),dense(death=12),[[.5,.5]]*24,[True]*24,23) |
| self.assertTrue(f['render_pred']) |
| self.assertTrue(f['pred_after_declared_death']) |
|
|
| def test_unmatched_prediction_uses_own_visibility(self): |
| self.assertTrue(builder.frame_payload(None,dense(),None,None,0)['render_pred']) |
| self.assertFalse(builder.frame_payload(None,dense(visible=False),None,None,0)['render_pred']) |
|
|
| def test_offscreen_gt_not_clamped(self): |
| f=builder.frame_payload(dense((None,None)),dense(),[[1.5,.3]]*24,[False]*24,0) |
| self.assertFalse(f['render_gt']) |
| self.assertEqual(f['gt_state'],'out of frame') |
|
|
| def test_create_only_output(self): |
| with tempfile.TemporaryDirectory() as existing: |
| with self.assertRaises(FileExistsError): |
| builder.build('unused.json',existing,'unused-repo') |
|
|
| def test_viewer_renders_untrusted_text_safely(self): |
| js=Path(__file__).with_name('viewer.js').read_text() |
| self.assertNotIn('.innerHTML',js) |
| self.assertIn('textContent',js) |
| self.assertIn('paintToken',js) |
|
|
| def test_pure_loader_refuses_missing_helpers(self): |
| with tempfile.TemporaryDirectory() as tmp: |
| path=Path(tmp)/'eval/utils' |
| path.mkdir(parents=True) |
| (path/'metrics_utils.py').write_text('raise RuntimeError("must not execute module body")\n') |
| with self.assertRaises(ValueError): |
| builder.load_dense_helper(tmp) |
|
|
| def test_dataset_readme_has_portable_instructions_and_scope(self): |
| card=publisher.dataset_readme('owner/data','owner/viewer') |
| self.assertIn('https://huggingface.co/spaces/owner/viewer',card) |
| self.assertIn('https://huggingface.co/datasets/owner/data/resolve/main/visualizer.zip',card) |
| self.assertIn('python3 serve.py',card) |
| self.assertIn('not the full benchmark',card) |
| self.assertNotIn('sdk:',card) |
|
|
| def _fake_hub(self, existing=None): |
| calls=[] |
| class Missing(Exception): |
| pass |
| class API: |
| def __init__(self,**kwargs):calls.append(('auth',kwargs)) |
| def whoami(self):return {'name':'owner'} |
| def repo_info(self,name,repo_type): |
| targets=existing if isinstance(existing,set) else {existing} |
| if (name,repo_type) in targets: |
| return types.SimpleNamespace(private=False,sdk='static',sha=repo_type+'-parent') |
| raise Missing() |
| def create_repo(self,**kwargs):calls.append(('create',kwargs)) |
| def upload_folder(self,**kwargs):calls.append(('upload',kwargs));return 'space-commit' |
| def create_commit(self,**kwargs):calls.append(('commit',kwargs));return 'dataset-commit' |
| module=types.ModuleType('huggingface_hub') |
| module.HfApi=API |
| module.CommitOperationAdd=lambda **kwargs:kwargs |
| errors=types.ModuleType('huggingface_hub.errors') |
| errors.RepositoryNotFoundError=Missing |
| return calls,{'huggingface_hub':module,'huggingface_hub.errors':errors} |
|
|
| def _tiny_bundle(self,root): |
| for name in ['manifest.json','index.html','README.md','visualizer.zip']: |
| (root/name).write_bytes(b'synthetic test data') |
| (root/'checksums.sha256').write_text(''.join( |
| f'{hashlib.sha256(p.read_bytes()).hexdigest()} {p.name}\n' for p in sorted(root.iterdir()))) |
|
|
| def test_public_publication_uses_selected_token_and_both_repos(self): |
| calls,modules=self._fake_hub() |
| with tempfile.TemporaryDirectory() as tmp: |
| root=Path(tmp)/'bundle';root.mkdir();self._tiny_bundle(root) |
| token=Path(tmp)/'credential';token.write_text('synthetic selected credential') |
| with patch.dict(sys.modules,modules),patch('builtins.print'): |
| result=publisher.publish(root,'owner/viewer',True,token,'owner/data') |
| self.assertEqual(calls[0],('auth',{'token':'synthetic selected credential'})) |
| creates=[args for name,args in calls if name=='create'] |
| self.assertEqual({c['repo_type'] for c in creates},{'space','dataset'}) |
| self.assertTrue(all(c['private'] is False and c['exist_ok'] is False for c in creates)) |
| self.assertFalse(result['dataset']['private']) |
|
|
| def test_existing_dataset_blocks_all_remote_writes(self): |
| calls,modules=self._fake_hub(existing=('owner/data','dataset')) |
| with tempfile.TemporaryDirectory() as tmp: |
| root=Path(tmp);self._tiny_bundle(root) |
| with patch.dict(sys.modules,modules),self.assertRaises(FileExistsError): |
| publisher.publish(root,'owner/viewer',dataset_repo_id='owner/data') |
| self.assertFalse(any(name in ['create','upload','commit'] for name,_ in calls)) |
|
|
| def test_other_namespace_blocks_all_remote_writes(self): |
| calls,modules=self._fake_hub() |
| with tempfile.TemporaryDirectory() as tmp: |
| root=Path(tmp);self._tiny_bundle(root) |
| with patch.dict(sys.modules,modules),self.assertRaises(PermissionError): |
| publisher.publish(root,'owner/viewer',dataset_repo_id='someone-else/data') |
| self.assertFalse(any(name in ['create','upload','commit'] for name,_ in calls)) |
|
|
| def test_explicit_update_keeps_repos_and_pins_parent_commits(self): |
| calls,modules=self._fake_hub(existing={('owner/viewer','space'),('owner/data','dataset')}) |
| with tempfile.TemporaryDirectory() as tmp: |
| root=Path(tmp);self._tiny_bundle(root) |
| with patch.dict(sys.modules,modules),patch('builtins.print'): |
| publisher.publish(root,'owner/viewer',public=True,dataset_repo_id='owner/data',update_existing=True) |
| self.assertFalse(any(name=='create' for name,_ in calls)) |
| upload=next(args for name,args in calls if name=='upload') |
| commit=next(args for name,args in calls if name=='commit') |
| self.assertEqual(upload['parent_commit'],'space-parent') |
| self.assertEqual(commit['parent_commit'],'dataset-parent') |
| self.assertNotIn('delete_patterns',upload) |
| self.assertIn('repackage.py',upload['allow_patterns']) |
|
|
| def test_update_missing_dataset_blocks_all_writes(self): |
| calls,modules=self._fake_hub(existing=('owner/viewer','space')) |
| with tempfile.TemporaryDirectory() as tmp: |
| root=Path(tmp);self._tiny_bundle(root) |
| with patch.dict(sys.modules,modules),self.assertRaises(FileNotFoundError): |
| publisher.publish(root,'owner/viewer',public=True,dataset_repo_id='owner/data',update_existing=True) |
| self.assertFalse(any(name in ['create','upload','commit'] for name,_ in calls)) |
|
|
| def test_update_cannot_silently_change_visibility(self): |
| calls,modules=self._fake_hub(existing=('owner/viewer','space')) |
| with tempfile.TemporaryDirectory() as tmp: |
| root=Path(tmp);self._tiny_bundle(root) |
| with patch.dict(sys.modules,modules),self.assertRaisesRegex(ValueError,'Visibility mismatch'): |
| publisher.publish(root,'owner/viewer',public=False,update_existing=True) |
| self.assertFalse(any(name in ['create','upload','commit'] for name,_ in calls)) |
|
|
| def test_repackage_keeps_original_archive_and_cases(self): |
| with tempfile.TemporaryDirectory() as tmp: |
| source=Path(tmp)/'old';source.mkdir() |
| (source/'cases').mkdir() |
| (source/'cases/case.json').write_text('original prediction') |
| (source/'manifest.json').write_text('{}') |
| (source/'visualizer.zip').write_bytes(b'original archive stays intact') |
| (source/'checksums.sha256').write_text(''.join( |
| f'{builder.digest(p)} {p.relative_to(source)}\n' |
| for p in sorted(source.rglob('*')) if p.is_file())) |
| output=Path(tmp)/'new' |
| with patch('builtins.print'): |
| repacker.repackage(source,output) |
| self.assertEqual((source/'visualizer.zip').read_bytes(),b'original archive stays intact') |
| self.assertEqual((source/'cases/case.json').read_bytes(),(output/'cases/case.json').read_bytes()) |
| self.assertEqual((source/'manifest.json').read_bytes(),(output/'manifest.json').read_bytes()) |
| with self.assertRaises(FileExistsError): |
| repacker.repackage(source,output) |
|
|
| def test_repackage_refuses_unverified_asset(self): |
| with tempfile.TemporaryDirectory() as tmp: |
| source=Path(tmp)/'old';source.mkdir() |
| (source/'cases').mkdir() |
| (source/'manifest.json').write_text('{}') |
| (source/'checksums.sha256').write_text('') |
| with self.assertRaises(ValueError): |
| repacker.repackage(source,Path(tmp)/'new') |
| self.assertFalse((Path(tmp)/'new').exists()) |
|
|
|
|
| if __name__=='__main__': |
| unittest.main() |
|
|
