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