import hashlib import json from pathlib import Path import tempfile import unittest from unittest.mock import patch from paper_atlas.collectors.public_openreview import (REVISION, SOURCE, collect, negative_status, normalize) from paper_atlas.db.store import Store from paper_atlas.ingestion.records import ingest_file from paper_atlas.audits.public_openreview import audit class PublicOpenReviewTests(unittest.TestCase): def row(self, **overrides): return {'id': 'RealForum01', 'title': 'A public paper', 'abstract': 'An abstract', 'author': 'Anonymous;Jane Doe;private@example.org', 'track': 'main', 'status': 'Reject', 'keywords': 'planning;reasoning', 'profiles': {'private': 'secret'}, 'reviews': ['private review'], 'gender': 'irrelevant', **overrides} def provenance(self): return {'source_file': 'https://example.org/pinned.jsonl', 'source_line': 7, 'source_revision': REVISION, 'source_blob_sha1': 'a' * 40} def fetcher(self, files): tree = [] payloads = {} for name, rows in files.items(): data = ''.join(json.dumps(row) + '\n' for row in rows).encode() sha = hashlib.sha1(b'blob ' + str(len(data)).encode() + b'\0' + data).hexdigest() tree.append({'path': name, 'type': 'blob', 'size': len(data), 'sha': sha}) payloads[name] = data def fetch(url, limit): if '?recursive=1' in url: return json.dumps({'sha': REVISION, 'truncated': False, 'tree': tree}).encode() return payloads[url.split('/' + REVISION + '/')[1]] return fetch def collect(self, out, files, **kwargs): with patch('paper_atlas.collectors.public_openreview.shutil.disk_usage', return_value=type('Space', (), {'free': 20 * 1024**3})()): return collect(out, fetcher=self.fetcher(files), **kwargs) def test_only_explicit_decisions_and_verified_scope_become_events(self): for value in ('Pending', 'Active', 'Submitted to NeurIPS 2025', 'TBD', '', 'Low score', 'ICML 2025 Retracted by Authors', 'Do not reject'): self.assertIsNone(negative_status(value), value) self.assertEqual(negative_status('Reject'), 'rejected') self.assertEqual(negative_status('Desk rejected'), 'desk_rejected') self.assertEqual(negative_status('NeurIPS 2023 Conference Withdrawn Submission'), 'withdrawn') for raw, expected in [('main', 'main'), ('Position', 'position'), ('Datasets & Benchmarks', 'datasets_benchmarks')]: row, reason = normalize(self.row(track=raw), 'nips', 2025, self.provenance()) self.assertIsNone(reason) self.assertEqual(row['track'], expected) for raw in ('workshop', 'workshop_mlsys', ''): self.assertIsNone(normalize(self.row(track=raw), 'icml', 2025, self.provenance())[0]) with self.assertRaisesRegex(ValueError, 'conference year'): normalize(self.row(status='NeurIPS 2023 Conference Withdrawn Submission'), 'icml', 2025, self.provenance()) def test_metadata_whitelist_retains_curated_provenance_without_profiles_or_reviews(self): row, _ = normalize(self.row(), 'icml', 2025, self.provenance()) self.assertEqual(row['authors'], ['Jane Doe']) self.assertEqual(row['identifiers'], {'openreview': 'RealForum01'}) self.assertEqual(row['metadata']['evidence_kind'], 'curated_public_status_assertion') self.assertFalse(row['metadata']['original_decision_signature_verified']) self.assertEqual(row['evidence_url'], 'https://example.org/pinned.jsonl#L7') for private in ('secret', 'private review', 'gender', 'private@example.org'): self.assertNotIn(private, json.dumps(row)) def test_pinned_collection_separates_tracks_and_holds_retraction(self): with tempfile.TemporaryDirectory() as tmp: files = {'venues/icml/icml2025.jsonl': [self.row(), self.row(id='Position01', track='Position'), self.row(id='Retracted1', status='ICML 2025 Retracted by Authors'), self.row(id='Pending001', status='Submitted to ICML 2025')]} result = self.collect(tmp, files) self.assertEqual(result['input_rows'], 4) self.assertEqual(result['normalized_records'], 2) self.assertEqual({f['track'] for f in result['files']}, {'main', 'position'}) held = [json.loads(line) for line in Path(result['held_file']).read_text().splitlines()] self.assertEqual([(r['id'], r['reason']) for r in held], [('Retracted1', 'retracted')]) self.assertTrue(all(f['status'] == 'partial' for f in result['files'])) with self.assertRaisesRegex(ValueError, 'Completed inventory'): self.collect(tmp, files) def test_hash_failure_or_duplicate_forum_publishes_no_ingestible_files(self): files = {'venues/icml/icml2025.jsonl': [self.row()]} fetch = self.fetcher(files) def corrupted(url, limit): result = fetch(url, limit) return result if '?recursive' in url else result.replace(b'Reject', b'Accept') for fetcher in (corrupted, self.fetcher({'venues/icml/icml2025.jsonl': [self.row(), self.row()]})): with tempfile.TemporaryDirectory() as tmp, patch('paper_atlas.collectors.public_openreview.shutil.disk_usage', return_value=type('Space', (), {'free': 20 * 1024**3})()): with self.assertRaises(ValueError): collect(tmp, fetcher=fetcher) self.assertEqual(list(Path(tmp).glob('*.jsonl')), []) self.assertFalse((Path(tmp) / 'public-openreview-inventory.json').exists()) def test_mirror_reimport_preserves_one_attempt_and_later_acceptance_elsewhere(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) result = self.collect(root / 'source', {'venues/icml/icml2025.jsonl': [self.row(author='Jane Doe')]}) db = root / 'db.sqlite' path = result['files'][0]['file'] ingest_file(db, path) ingest_file(db, path) store = Store(db) try: paper = store.conn.execute('SELECT paper_id FROM source_records').fetchone()[0] store.conn.execute('INSERT INTO identifiers VALUES(?,?,?)', ('arxiv', '2501.00001', paper)) same, record = store.record('official_program', 'neurips-later', 'https://example.org/accepted', 'A revised public paper', 'An abstract', ['Jane Doe'], {'openreview': 'LaterForum1', 'arxiv': '2501.00001'}) self.assertEqual(same, paper) store.decision(paper, record, 'NeurIPS', 2026, 'main', 'neurips-later', 'https://example.org/accepted', 'accepted', 'Accept', identities=[('openreview', 'LaterForum1')]) store.conn.commit() self.assertEqual(store.conn.execute('SELECT count(*) FROM submissions').fetchone()[0], 2) self.assertEqual(store.conn.execute('SELECT count(*) FROM decisions').fetchone()[0], 2) self.assertEqual(set(store.conn.execute('SELECT status FROM decisions')), {('accepted',), ('rejected',)}) self.assertEqual(store.conn.execute('SELECT count(*) FROM papers').fetchone()[0], 1) self.assertEqual(store.conn.execute('PRAGMA foreign_key_check').fetchall(), []) finally: store.conn.close() def test_preimport_audit_blocks_same_attempt_conflict_but_allows_later_other_venue_acceptance(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) self.collect(root / 'source', {'venues/icml/icml2025.jsonl': [self.row(author='Jane Doe')]}) db = root / 'db.sqlite' store = Store(db) paper, record = store.record('program', 'later', 'https://example.org/later', 'A public paper', 'An abstract', ['Jane Doe'], {'openreview': 'LaterForum1'}) store.decision(paper, record, 'NeurIPS', 2026, 'main', 'later', 'https://example.org/later', 'accepted', 'Accept', identities=[('openreview', 'LaterForum1')]) store.conn.commit() inventory = root / 'source/public-openreview-inventory.json' first = audit(db, inventory, root / 'audit.json') self.assertEqual(first['status'], 'ready_for_serial_import') self.assertEqual(len(first['files']), 1) store.decision(paper, record, 'ICML', 2025, 'main', 'same-attempt', 'https://example.org/same', 'accepted', 'Accept', identities=[('openreview', 'RealForum01')]) store.conn.commit() try: second = audit(db, inventory, root / 'audit.json') self.assertEqual(second['status'], 'blocked') self.assertEqual(second['files'], []) self.assertEqual(second['conflicts'][0]['existing_status'], 'accepted') finally: store.conn.close() def test_unicode_separators_in_abstract_survive_audit_and_import(self): for separator in ('\u0085', '\u2028', '\u2029'): with self.subTest(separator=repr(separator)), tempfile.TemporaryDirectory() as tmp: root = Path(tmp) abstract = 'First paragraph' + separator + 'Second paragraph' result = self.collect(root / 'source', { 'venues/icml/icml2025.jsonl': [self.row(abstract=abstract)]}) path = Path(result['files'][0]['file']) data = path.read_bytes() self.assertIn(separator.encode(), data) self.assertEqual(data.count(b'\n'), 1) db = root / 'db.sqlite' store = Store(db) store.conn.close() inventory = root / 'source/public-openreview-inventory.json' before = audit(db, inventory, root / 'before.json') self.assertEqual(before['status'], 'ready_for_serial_import') self.assertEqual(before['records_checked'], 1) ingest_file(db, path) after = audit(db, inventory, root / 'after.json') self.assertEqual(after['status'], 'ready_for_serial_import') self.assertEqual(after['existing_same_edition_attempts'], 1) store = Store(db) try: self.assertEqual(store.conn.execute( 'SELECT abstract FROM source_records').fetchone()[0], abstract) finally: store.conn.close() if __name__ == '__main__': unittest.main()