paper-atlas/tests/test_public_openreview.py

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',
                      'Workshop', 'Poster Workshop', 'Oral Workshop'):
            self.assertIsNone(negative_status(value), value)
        self.assertEqual(negative_status('Reject'), 'rejected')
        self.assertEqual(negative_status('Desk rejected'), 'desk_rejected')
        self.assertEqual(negative_status('Withdraw'), 'withdrawn')
        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_missing_title_is_held_without_losing_valid_negative_records(self):
        with tempfile.TemporaryDirectory() as tmp:
            result = self.collect(tmp, {'venues/iclr/iclr2017.jsonl': [
                self.row(id='SJUdkecgx', title=''),
                self.row(id='NoTitle001', title=None),
                self.row(id='Whitespace', title='  '),
                self.row(id='ValidReject'),
                self.row(id='Withdraw01', status='Withdraw'),
                self.row(id='Workshop01', status='Workshop'),
            ]}, families=['iclr'], years=[2017])
            self.assertEqual(result['input_rows'], 6)
            self.assertEqual(result['normalized_records'], 2)
            self.assertEqual(result['source_files'][0]['excluded_or_held'],
                             {'missing_title': 3, 'not_explicit_negative': 1})
            held_path = Path(result['held_file'])
            self.assertEqual(hashlib.sha256(held_path.read_bytes()).hexdigest(),
                             result['held_file_sha256'])
            held = [json.loads(line) for line in held_path.read_text().splitlines()]
            self.assertEqual([row['id'] for row in held], ['SJUdkecgx', 'NoTitle001', 'Whitespace'])
            self.assertTrue(all(row['reason'] == 'missing_title' and row['status'] == 'Reject'
                                for row in held))
            self.assertEqual([row['provenance']['source_line'] for row in held], [1, 2, 3])
            self.assertNotIn('private@example.org', held_path.read_text())
            self.assertNotIn('private review', held_path.read_text())
            manifest = result['files'][0]
            self.assertEqual((manifest['parser_version'], manifest['status']), (3, 'partial'))
            rows = [json.loads(line) for line in Path(manifest['file']).read_text().splitlines()]
            self.assertEqual([(row['identifiers']['openreview'], row['status']) for row in rows],
                             [('ValidReject', 'rejected'), ('Withdraw01', 'withdrawn')])
            for row in held:
                self.assertEqual(row['provenance']['source_revision'], REVISION)

    def test_nontext_title_still_fails_collection(self):
        with tempfile.TemporaryDirectory() as tmp:
            with self.assertRaisesRegex(ValueError, 'lacks a title'):
                self.collect(tmp, {'venues/iclr/iclr2017.jsonl': [self.row(title=['invalid'])]},
                             families=['iclr'], years=[2017])
            self.assertFalse((Path(tmp) / 'public-openreview-inventory.json').exists())
            self.assertEqual(list(Path(tmp).glob('*.jsonl')), [])

    def workshop_cases(self):
        return [
            (2024, 'workshop_me_fomo', 'ICLR 2024 Workshop MEFoMo Withdrawn Submission', 'withdrawn', 'ME-FoMo'),
            (2024, 'workshop_genai4dm', 'ICLR 2024 Workshop GenAI4DM Withdrawn Submission', 'withdrawn', 'GenAI4DM'),
            (2025, 'workshop_agenticai', 'ICLR 2025 Workshop AgenticAI Reject', 'rejected', 'AgenticAI'),
            (2025, 'workshop_lmrl', 'ICLR 2025 Workshop LMRL Withdrawn Submission', 'withdrawn', 'LMRL'),
            (2026, 'workshop_llm_reasoning', 'ICLR 2026 Workshop LLM Reasoning Withdrawn Submission', 'withdrawn', 'LLM_Reasoning'),
        ]

    def test_iclr_workshop_labels_require_opt_in_and_retain_identity_evidence(self):
        for year, track, label, status, group in self.workshop_cases():
            with self.subTest(label=label):
                raw = self.row(track=track, status=label)
                self.assertEqual(normalize(raw, 'iclr', year, self.provenance()),
                                 (None, 'separate_workshop'))
                row, reason = normalize(raw, 'iclr', year, self.provenance(), True)
                self.assertIsNone(reason)
                self.assertEqual((row['venue'], row['year'], row['track'], row['status']),
                                 ('ICLR', year, track, status))
                self.assertEqual(row['raw_decision'], label)
                self.assertEqual(row['metadata']['workshop_venueid'],
                                 f'ICLR.cc/{year}/Workshop/{group}')
                self.assertFalse(row['metadata']['original_decision_signature_verified'])
                self.assertEqual(row['evidence_url'], 'https://example.org/pinned.jsonl#L7')
                self.assertNotIn('private review', json.dumps(row))

    def test_workshop_label_scope_conflicts_never_become_main_events(self):
        for year, track, label, _, _ in self.workshop_cases():
            for enabled in (False, True):
                for wrong_track in ('main', 'workshop_other'):
                    self.assertEqual(normalize(self.row(track=wrong_track, status=label),
                                               'iclr', year, self.provenance(), enabled),
                                     (None, 'conflicting_workshop_track'))
                for family, wrong_year in [('icml', year), ('iclr', year + 1)]:
                    with self.assertRaisesRegex(ValueError, 'conference year'):
                        normalize(self.row(track=track, status=label), family,
                                  wrong_year, self.provenance(), enabled)
        row, reason = normalize(self.row(track='workshop_unknown',
                                        status='ICLR 2026 Workshop Unknown Reject'),
                                'iclr', 2026, self.provenance(), True)
        self.assertIsNone(row)
        self.assertEqual(reason, 'unmapped_negative_label')

    def test_workshop_collection_and_import_keep_five_tracks_separate_from_main(self):
        files = {}
        for index, (year, track, label, _, _) in enumerate(self.workshop_cases()):
            files.setdefault(f'venues/iclr/iclr{year}.jsonl', []).append(
                self.row(id=f'Workshop{index:02}', track=track, status=label))
        files['venues/iclr/iclr2024.jsonl'].append(self.row())
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            default = self.collect(root / 'default', files, families=['iclr'])
            self.assertEqual(default['normalized_records'], 1)
            held = [json.loads(line) for line in Path(default['held_file']).read_bytes().split(b'\n') if line]
            self.assertEqual(len(held), 5)
            self.assertTrue(all(row['reason'] == 'separate_workshop' for row in held))
            selected = self.collect(root / 'selected', files, families=['iclr'], include_workshops=True)
            self.assertEqual(selected['normalized_records'], 6)
            self.assertEqual(len(selected['files']), 6)
            self.assertEqual(Path(selected['held_file']).read_bytes(), b'')
            db = root / 'db.sqlite'
            for manifest in selected['files']:
                self.assertEqual(manifest['parser_version'], 3)
                ingest_file(db, manifest['file'])
            store = Store(db)
            try:
                scopes = set(store.conn.execute('SELECT e.year,e.track,d.status FROM submissions s '
                                               'JOIN editions e ON e.id=s.edition_id '
                                               'JOIN decisions d ON d.submission_id=s.id'))
                expected = {(year, track, status) for year, track, _, status, _ in self.workshop_cases()}
                self.assertEqual(scopes, expected | {(2024, 'main', 'rejected')})
                self.assertEqual(store.conn.execute('PRAGMA foreign_key_check').fetchall(), [])
            finally:
                store.conn.close()

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