import copy from collections import Counter import gzip import hashlib import json from pathlib import Path import tempfile import unittest from unittest.mock import patch from urllib.parse import parse_qs, urlsplit from paper_atlas.collectors import neurips_program as program FIXTURES = Path(__file__).parent/'fixtures' RAW = gzip.decompress((FIXTURES/'neurips2026-program.json.gz').read_bytes()) RECEIPTS = json.loads((FIXTURES/'neurips2026-receipts.json').read_text()) EVIDENCE = {'url': program.URL, 'observed_at': RECEIPTS['receipts'][0]['finished_at']} class NeuripsProgramTests(unittest.TestCase): def sample(self): rows = json.loads(RAW)['results'] oral = next(r for r in rows if r['id'] == 154866) poster = next(r for r in rows if r['paper_url'] == oral['paper_url'] and r['event_type'] == 'Poster') return copy.deepcopy(poster), copy.deepcopy(oral) def payload(self, rows): return json.dumps({'count': len(rows), 'next': None, 'previous': None, 'results': rows}) def snapshots(self, folder): folder.mkdir() (folder/'program-json.response').write_bytes(RAW) (folder/'receipts.json').write_text(json.dumps(RECEIPTS)) return folder def test_full_snapshot_tracks_and_all_titles_authors_affiliations(self): rows, held, duplicates, folded = program.parse_program(RAW, EVIDENCE) self.assertEqual((len(rows), len(held), len(duplicates), folded), (9004, 139, 1, 132)) self.assertEqual(Counter(r['track'] for r in rows), {'main': 7896, 'evaluations_datasets': 971, 'position': 137}) expected = {} for e in json.loads(RAW)['results']: if e['sourceurl'] in program.GROUPS and e['event_type'] == 'Poster': forum = parse_qs(urlsplit(e['paper_url']).query)['id'][0] expected[forum] = e self.assertEqual(len(expected), 9004) for r in rows: e = expected[r['identifiers']['openreview']] self.assertEqual(r['title'], e['name']) self.assertEqual(r['authors'], [a['fullname'] for a in e['authors']]) self.assertEqual(r['metadata']['author_affiliations'], [a.get('institution') for a in e['authors']]) self.assertEqual(r['raw_decision'], e['decision']) self.assertEqual(r['status'], 'accepted') self.assertNotIn('decision_date', r) self.assertIsNone(r['abstract']) self.assertEqual(r['year'], 2026) self.assertEqual(duplicates[0]['event_id'], 154866) def test_same_forum_folds_presentations_and_preserves_signed_source_ids(self): poster, oral = self.sample() rows, held, duplicates, folded = program.parse_program(self.payload([oral, poster, oral]), EVIDENCE) row, = rows self.assertEqual((len(held), len(duplicates), folded), (0, 1, 1)) self.assertEqual({p['sourceid'] for p in row['metadata']['presentations']}, {poster['sourceid'], oral['sourceid']}) self.assertLess(oral['sourceid'], 0) self.assertEqual({t['value'] for t in row['tags']}, {'Poster', 'Oral'}) self.assertTrue(all(t['namespace'] == 'neurips_program:presentation_format' for t in row['tags'])) def test_identical_title_and_authors_in_distinct_forums_remain_distinct(self): poster, _ = self.sample(); other = copy.deepcopy(poster) other.update(id=999999, uid='0'*32, sourceid=999999, paper_url='https://openreview.net/forum?id=otherForum7', virtualsite_url='/virtual/2026/poster/999999') rows, _, _, _ = program.parse_program(self.payload([poster, other]), EVIDENCE) self.assertEqual(len(rows), 2) self.assertEqual(len({r['external_id'] for r in rows}), 2) def test_conflicting_event_ids_or_same_forum_bibliography_are_rejected(self): poster, oral = self.sample() for field in ('name', 'authors', 'sourceurl', 'decision'): changed = copy.deepcopy(oral) if field == 'authors': changed['authors'][0]['fullname'] = 'Different Author' elif field == 'sourceurl': changed[field] = 'https://openreview.net/group?id=NeurIPS.cc/2026/Position_Paper_Track' elif field == 'decision': changed[field] = 'Accept (poster)' else: changed[field] += ' changed' with self.subTest(field=field), self.assertRaises(ValueError): program.parse_program(self.payload([poster, changed]), EVIDENCE) changed = copy.deepcopy(poster); changed['name'] += ' changed' with self.assertRaises(ValueError): program.parse_program(self.payload([poster, changed]), EVIDENCE) def test_unreviewed_origin_decision_visibility_and_link_are_held(self): poster, _ = self.sample() mutations = [{'sourceurl': 'journal-track/NeurIPS_2026-Posters_TMLR.xlsx'}, {'sourceurl': 'neurips2026-bridging-the-future'}, {'sourceurl': None}, {'sourceurl': 'https://openreview.net/group?id=NeurIPS.cc/2027/Conference'}, {'decision': 'Reject'}, {'visible': False}, {'paper_url': 'https://other.example/forum?id=anotherID7'}] for mutation in mutations: changed = copy.deepcopy(poster); changed.update(id=999999, virtualsite_url='/virtual/2026/poster/999999') changed.update(mutation) rows, held, _, _ = program.parse_program(self.payload([poster, changed]), EVIDENCE) self.assertEqual((len(rows), len(held)), (1, 1)) self.assertEqual(held[0]['raw_event'], changed) def test_rejects_truncation_pagination_invalid_url_and_missing_authors(self): poster, _ = self.sample() raw = json.loads(self.payload([poster])) bad = [dict(raw, count=2), dict(raw, next='https://example.com/next'), {k: v for k, v in raw.items() if k != 'previous'}] for payload in bad: with self.assertRaises(ValueError): program.parse_program(json.dumps(payload), EVIDENCE) for mutation in [{'authors': []}, {'virtualsite_url': '/virtual/2025/poster/154865'}, {'id': True}]: changed = copy.deepcopy(poster); changed.update(mutation) with self.assertRaises(ValueError): program.parse_program(self.payload([changed]), EVIDENCE) def test_frozen_collection_is_partial_and_no_http_or_database_write(self): with tempfile.TemporaryDirectory() as tmp, patch.object(program, 'fetch', side_effect=AssertionError('No HTTP')): root = Path(tmp); source = self.snapshots(root/'source') report = program.collect(root/'out', snapshots=source) self.assertEqual(report['records'], 9004) self.assertEqual(len(report['files']), 3) self.assertFalse(report['database_written']) self.assertFalse(report['pdfs_downloaded']) for m in report['files']: self.assertEqual(m['status'], 'partial') self.assertEqual(m['missing_abstract_count'], m['count']) self.assertEqual(hashlib.sha256(Path(m['file']).read_bytes()).hexdigest(), m['file_sha256']) with self.assertRaises(ValueError): program.collect(root/'out', snapshots=source) def test_frozen_receipt_tamper_path_symlink_status_and_duplicate(self): for mode in ('tamper', 'path', 'symlink', 'status', 'duplicate'): with self.subTest(mode=mode), tempfile.TemporaryDirectory() as tmp: root = Path(tmp); source = self.snapshots(root/'source') receipts = copy.deepcopy(RECEIPTS) if mode == 'tamper': (source/'program-json.response').write_text('bad') elif mode == 'path': receipts['receipts'][0]['file'] = '../outside.json' elif mode == 'symlink': (root/'outside.json').write_bytes(RAW) (source/'program-json.response').unlink() (source/'program-json.response').symlink_to(root/'outside.json') elif mode == 'status': receipts['receipts'][0]['status'] = 403 else: receipts['receipts'].append(copy.deepcopy(receipts['receipts'][0])) (source/'receipts.json').write_text(json.dumps(receipts)) with self.assertRaises(ValueError): program.collect(root/'out', snapshots=source) self.assertFalse((root/'out/neurips-program-2026.json').exists()) def test_online_mode_only_requests_fixed_program_metadata(self): poster, _ = self.sample(); raw = self.payload([poster]).encode() with tempfile.TemporaryDirectory() as tmp: root = Path(tmp); cache = root/'cache'; cache.mkdir() receipt = {'retrieved_at': EVIDENCE['observed_at'], 'url': program.URL} (cache/hashlib.sha256(program.URL.encode()).hexdigest()).with_suffix('.json').write_text(json.dumps(receipt)) with patch.object(program, 'fetch', return_value=raw) as fetch: report = program.collect(root/'out', cache=cache) fetch.assert_called_once_with(program.URL, cache, program.LIMIT, delay=1) self.assertEqual(report['records'], 1) def test_frozen_fixture_preserves_actual_response_bytes(self): receipt, = RECEIPTS['receipts'] self.assertEqual(hashlib.sha256(RAW).hexdigest(), receipt['sha256']) self.assertEqual(len(RAW), receipt['bytes'])