import copy import fcntl import hashlib import json from pathlib import Path import sqlite3 import tempfile import unittest from unittest.mock import patch from paper_atlas.collectors.pmlr_tracks import program_tracks, record_track from paper_atlas.db.store import Store from paper_atlas.db.submissions import identities_for_row from paper_atlas.ingestion.pmlr_tracks import reclassify, plan from paper_atlas.ingestion.records import ingest_file from tests.test_pmlr_tracks import event, program, record class PmlrMigrationTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory(); self.addCleanup(self.tmp.cleanup) self.root = Path(self.tmp.name); self.db = self.root / 'papers.sqlite' self.inventory = self.root / 'inventory.json' self.rows = [record(title='A main research paper'), record('position1', 'v267/paper2', title='A position paper')] self.proof = {'url': 'https://icml.cc/static/virtual/data/icml-2025-orals-posters.json', 'sha256': 'a' * 64} mapping, _ = program_tracks(program(event(), event('position1', 'Position_Paper_Track', 2)), 2025) store = Store(self.db); c = store.conn for row in self.rows: row.update(url='https://proceedings.mlr.press/' + row['external_id'] + '.html', list_url='https://proceedings.mlr.press/v267/', status='accepted') paper, source = store.record('pmlr', row['external_id'], row['url'], row['title'], row['abstract'], row['authors'], row['identifiers'], {'openreview': row['identifiers']['openreview']}) store.decision(paper, source, 'ICML', 2025, 'main', row['external_id'], row['url'], 'accepted', 'listed in official proceedings', identities=identities_for_row(row)) record_track(row, mapping, self.proof) c.execute("INSERT INTO tags(namespace,value) VALUES('custom:topic','keep me')") c.execute("INSERT INTO paper_tags VALUES(1,1,NULL,'custom:manual')") c.commit(); c.close() self.write_inventory() def write_inventory(self): files = [] for row in self.rows: path = self.root / (row['track'] + '.jsonl') data = (json.dumps(row) + '\n').encode(); path.write_bytes(data) m = {'source': 'pmlr', 'venue': 'ICML', 'year': 2025, 'volume': 267, 'parser_version': 3, 'track': row['track'], 'track_program': self.proof, 'count': 1, 'file': str(path), 'file_sha256': hashlib.sha256(data).hexdigest(), 'missing_abstract_count': 0, 'list_url': row['list_url'], 'status': 'collected'} path.with_suffix('.manifest.json').write_text(json.dumps(m)); files.append(m) self.inventory.write_text(json.dumps({'source': 'pmlr', 'venue': 'ICML', 'year': 2025, 'volume': 267, 'parser_version': 3, 'track_program': self.proof, 'count': 2, 'files': files})) def tracks(self): with sqlite3.connect(self.db) as c: return c.execute('SELECT s.id,e.track FROM submissions s JOIN editions e ON e.id=s.edition_id ORDER BY s.id').fetchall() def test_plan_is_read_only_and_reports_exact_original_event(self): result = reclassify(self.db, [self.inventory], self.root / 'plan.json') self.assertEqual((result['status'], result['records_verified'], result['events_to_move']), ('planned', 2, 1)) self.assertEqual(result['changes'][0]['submission_id'], 2) self.assertEqual(self.tracks(), [(1, 'main'), (2, 'main')]) def test_apply_preserves_ids_decisions_text_tags_and_is_idempotent(self): with sqlite3.connect(self.db) as c: before = {table: c.execute('SELECT * FROM ' + table).fetchall() for table in ('papers', 'decisions', 'decision_sources', 'submission_sources', 'paper_tags')} result = reclassify(self.db, [self.inventory], self.root / 'apply.json', True) self.assertEqual((result['status'], result['events_moved']), ('complete', 1)) self.assertEqual(self.tracks(), [(1, 'main'), (2, 'position')]) with sqlite3.connect(self.db) as c: for table, rows in before.items(): self.assertEqual(c.execute('SELECT * FROM ' + table).fetchall(), rows, table) self.assertEqual(c.execute('SELECT count(*) FROM submission_identifiers i JOIN submissions s ON s.id=i.submission_id WHERE i.edition_id<>s.edition_id').fetchone()[0], 0) self.assertEqual(c.execute('SELECT e.track,es.imported_count FROM edition_sources es JOIN editions e ON e.id=es.edition_id ORDER BY e.track').fetchall(), [('main', 1), ('position', 1)]) self.assertEqual(c.execute('PRAGMA foreign_key_check').fetchall(), []) again = reclassify(self.db, [self.inventory], self.root / 'again.json', True) self.assertEqual(again['events_moved'], 0) for m in json.loads(self.inventory.read_text())['files']: ingest_file(self.db, m['file']) self.assertEqual(self.tracks(), [(1, 'main'), (2, 'position')]) with sqlite3.connect(self.db) as c: self.assertEqual(c.execute('SELECT * FROM decisions').fetchall(), before['decisions']) def test_conflicting_target_event_stops_without_merging_history(self): store = Store(self.db) _, source = store.record('other', 'position1', 'https://example.org', 'Other evidence', None, [], {'openreview': 'position1'}) store.decision(2, source, 'ICML', 2025, 'position', 'position1', 'https://example.org', 'rejected', 'Reject', identities=[('openreview', 'position1')]) store.conn.commit(); store.conn.close() before = self.tracks() with self.assertRaisesRegex(ValueError, 'another event'): reclassify(self.db, [self.inventory], self.root / 'blocked.json', True) self.assertEqual(self.tracks(), before) def test_shared_event_requires_reconciliation_and_keeps_all_sources(self): store = Store(self.db) _, source = store.record('mirror', 'position1', 'https://example.org', 'Mirror title', None, [], {'openreview': 'position1'}) store.conn.execute('INSERT INTO submission_sources VALUES(?,?,?)', (2, source, 'position1')) store.conn.commit(); store.conn.close() with self.assertRaisesRegex(ValueError, 'multiple sources'): reclassify(self.db, [self.inventory], self.root / 'shared.json', True) self.assertEqual(self.tracks(), [(1, 'main'), (2, 'main')]) def test_text_mismatch_in_last_record_prevents_any_changes(self): with sqlite3.connect(self.db) as c: c.execute("UPDATE source_records SET abstract='Changed elsewhere' WHERE id=2") with self.assertRaisesRegex(ValueError, 'preserve existing publisher text'): reclassify(self.db, [self.inventory], self.root / 'text.json', True) self.assertEqual(self.tracks(), [(1, 'main'), (2, 'main')]) def test_mutation_failure_rolls_back_editions_metadata_and_identifiers(self): calls = 0 def fail_after_updates(c, scopes): nonlocal calls calls += 1 if calls == 2: raise ValueError('Injected post-migration failure') return plan(c, scopes) with sqlite3.connect(self.db) as c: before = c.execute('SELECT * FROM source_records').fetchall() with patch('paper_atlas.ingestion.pmlr_tracks.plan', side_effect=fail_after_updates): with self.assertRaisesRegex(ValueError, 'Injected'): reclassify(self.db, [self.inventory], self.root / 'rollback.json', True) self.assertEqual(self.tracks(), [(1, 'main'), (2, 'main')]) with sqlite3.connect(self.db) as c: self.assertEqual(c.execute('SELECT * FROM source_records').fetchall(), before) self.assertEqual(c.execute("SELECT count(*) FROM editions WHERE track='position'").fetchone()[0], 0) self.assertEqual(json.loads((self.root / 'rollback.json').read_text())['status'], 'error') def test_changed_evidence_and_active_writer_are_refused(self): out = self.root / 'blocked.json' with Path(str(self.db) + '.maintenance.lock').open('a') as lock: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) with self.assertRaises(BlockingIOError): reclassify(self.db, [self.inventory], out, True) self.assertFalse(out.exists()) path = self.root / 'position.jsonl'; path.write_text(path.read_text() + ' ') with self.assertRaisesRegex(ValueError, 'differs from its verified inventory'): reclassify(self.db, [self.inventory], out, True) self.assertEqual(self.tracks(), [(1, 'main'), (2, 'main')]) def test_input_report_and_duplicate_forums_are_refused(self): original = self.inventory.read_bytes() with self.assertRaisesRegex(ValueError, 'preserve evidence'): reclassify(self.db, [self.inventory], self.inventory, True) self.assertEqual(self.inventory.read_bytes(), original) self.rows[1]['identifiers']['openreview'] = self.rows[0]['identifiers']['openreview'] self.rows[1]['metadata']['track_evidence']['identity_value'] = self.rows[0]['identifiers']['openreview'] self.write_inventory() with self.assertRaisesRegex(ValueError, 'repeated forum'): reclassify(self.db, [self.inventory], self.root / 'duplicate.json', True) if __name__ == '__main__': unittest.main()