import copy import hashlib import json from pathlib import Path import tempfile import unittest from unittest.mock import patch from paper_atlas.audits import iclr_legacy from paper_atlas.ingestion.backfill import backfill, queue_files from paper_atlas.ingestion.records import ingest_file from paper_atlas.db.store import Store class ICLRLegacyTests(unittest.TestCase): def program(self): return {'source': 'iclr_program', 'external_id': '2020/program-42', 'venue': 'ICLR', 'year': 2020, 'track': 'main', 'status': 'accepted', 'url': 'https://iclr.cc/virtual/2020/papers.html', 'title': 'Historical paper title', 'abstract': 'Historical abstract', 'authors': ['Jane Doe', 'John Smith'], 'identifiers': {}, 'raw_decision': 'listed in official program', 'evidence_url': 'https://iclr.cc/virtual/2020/papers.html', 'metadata': {'program_identity_type': 'submission', 'program_source_id': 42}} def snapshot(self): return {'source': 'openreview_snapshot', 'external_id': '2020/realForum', 'venue': 'ICLR', 'year': 2020, 'track': 'main', 'status': 'accepted', 'url': 'https://openreview.net/forum?id=realForum', 'title': 'Historical paper title', 'abstract': 'Historical abstract', 'authors': ['Jane Doe', 'John Smith'], 'identifiers': {'openreview': 'realForum'}, 'submission_identifiers': [{'scheme': 'openreview', 'value': 'realForum'}, {'scheme': 'iclr_number', 'value': '42'}], 'decision_events': [{'status': 'accepted', 'raw_value': 'Accept', 'source_decision_id': 'decision1', 'evidence_url': 'https://openreview.net/forum?id=realForum¬eId=decision1', 'authority': {'public_readers_verified': True}}], 'metadata': {'source_submission_number': 42, 'public_readers_verified': True}} def legacy(self): return {'id': 'realForum', 'forum': 'realForum', 'content': { 'iclr_id': 'realForum', 'title': 'Historical paper title', 'authors': ['Jane Doe', 'John Smith']}} def test_independent_legacy_identity_preserves_existing_source_id_and_text(self): program, submission = self.program(), self.snapshot() submission.update(title='A later title', abstract='A later abstract', authors=['Jane Doe']) rows, checks = iclr_legacy.reconcile([program], {'realForum': submission}, [self.legacy()]) self.assertEqual(rows[0]['external_id'], '2020/program-42') self.assertEqual(rows[0]['url'], 'https://iclr.cc/virtual_2020/poster_realForum.html') for field in ('title', 'abstract', 'authors'): self.assertEqual(rows[0][field], program[field]) self.assertEqual(rows[0]['identifiers']['openreview'], 'realForum') self.assertTrue(rows[0]['metadata']['program_submission_number_verified']) self.assertEqual(checks['independent_legacy'], 1) self.assertEqual(program['identifiers'], {}) def test_program_number_and_shared_authors_alone_do_not_assign_a_forum(self): program = self.program() program.update(title='A different paper', abstract='An unrelated abstract') with self.assertRaisesRegex(ValueError, 'number alone'): iclr_legacy.reconcile([program], {'realForum': self.snapshot()}, [self.legacy()]) def test_independent_snapshot_title_and_abstract_handle_legacy_name_variants(self): legacy = self.legacy() legacy['content']['authors'] = ['J. Doe', 'J. Smith'] rows, checks = iclr_legacy.reconcile([self.program()], {'realForum': self.snapshot()}, [legacy]) self.assertEqual(checks['independent_snapshot'], 1) self.assertEqual(rows[0]['identifiers']['openreview'], 'realForum') def test_reviewed_title_variant_is_exact_and_requires_public_author_corroboration(self): variant = iclr_legacy.REVIEWED_TITLE_VARIANT program, submission, legacy = self.program(), self.snapshot(), self.legacy() program['title'] = variant['program'] submission.update(title=variant['snapshot'], abstract='A different revision') submission['identifiers']['openreview'] = variant['forum'] legacy.update(id=variant['forum'], forum=variant['forum']) legacy['content'].update(iclr_id=variant['forum'], title=variant['legacy']) rows, checks = iclr_legacy.reconcile([program], {variant['forum']: submission}, [legacy]) self.assertEqual(checks['reviewed_title_variant'], 1) self.assertTrue(rows[0]['metadata']['program_identity_verification']['reviewed_title_variant']) program['authors'] = ['Jane Doe'] with self.assertRaisesRegex(ValueError, 'number alone'): iclr_legacy.reconcile([program], {variant['forum']: submission}, [legacy]) def test_conflicting_forum_ambiguous_number_and_repeated_legacy_identity_fail(self): p = self.program() p['identifiers']['openreview'] = 'anotherForum' with self.assertRaisesRegex(ValueError, 'conflicts'): iclr_legacy.reconcile([p], {'realForum': self.snapshot()}, [self.legacy()]) duplicate = copy.deepcopy(self.snapshot()) duplicate['identifiers']['openreview'] = 'otherForum' with self.assertRaisesRegex(ValueError, 'ambiguous'): iclr_legacy.reconcile([self.program()], {'realForum': self.snapshot(), 'otherForum': duplicate}, [self.legacy()]) with self.assertRaisesRegex(ValueError, 'repeated'): iclr_legacy.reconcile([self.program()], {'realForum': self.snapshot()}, [self.legacy(), self.legacy()]) def write_file(self, path, rows, source): path.write_text(''.join(json.dumps(row) + '\n' for row in rows)) manifest = {'source': source, 'venue': 'ICLR', 'year': 2020, 'track': 'main', 'parser_version': 2, 'count': len(rows), 'missing_abstract_count': 0, 'list_url': 'https://iclr.cc/virtual/2020/papers.html', 'status': 'collected', 'submission_status_counts': {status: sum(r['status'] == status for r in rows) for status in {r['status'] for r in rows}}} path.with_suffix('.manifest.json').write_text(json.dumps(manifest)) def test_audited_bridge_then_snapshot_import_keeps_one_accepted_attempt(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) program_path, snapshot_path, legacy_path = (root / name for name in ('program.jsonl', 'snapshot.jsonl', 'legacy.json')) self.write_file(program_path, [self.program()], 'iclr_program') self.write_file(snapshot_path, [self.snapshot()], 'openreview_snapshot') raw = json.dumps([self.legacy()]).encode() legacy_path.write_bytes(raw) with patch.multiple(iclr_legacy, LEGACY_BYTES=len(raw), LEGACY_SHA256=hashlib.sha256(raw).hexdigest(), EXPECTED=1): report = iclr_legacy.audit(program_path, snapshot_path, legacy_path, root / 'out') db = root / 'papers.sqlite' ingest_file(db, program_path) queue = root / 'out/iclr-2020-backfill-queue.json' result = backfill(db, queue_files(queue)) self.assertEqual(result['submissions_before'], result['submissions_after']) ingest_file(db, report['backfill_queue']['files'][0]) ingest_file(db, snapshot_path) store = Store(db) try: self.assertEqual(store.conn.execute('SELECT count(*) FROM papers').fetchone()[0], 1) self.assertEqual(store.conn.execute('SELECT count(*) FROM submissions').fetchone()[0], 1) self.assertEqual(store.conn.execute('SELECT count(*) FROM submission_sources').fetchone()[0], 2) self.assertEqual(store.conn.execute('SELECT url FROM source_records WHERE external_id=?', ('2020/program-42',)).fetchone()[0], 'https://iclr.cc/virtual_2020/poster_realForum.html') self.assertEqual(store.conn.execute('PRAGMA foreign_key_check').fetchall(), []) finally: store.conn.close() def test_unverified_legacy_bytes_publish_no_output(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) legacy = root / 'legacy.json' legacy.write_text('[]') with self.assertRaisesRegex(ValueError, 'checksum'): iclr_legacy.audit(root / 'program', root / 'snapshot', legacy, root / 'out') self.assertFalse((root / 'out').exists()) if __name__ == '__main__': unittest.main()