175 lines
9.5 KiB
Python
175 lines
9.5 KiB
Python
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'])
|