157 lines
9.4 KiB
Python
157 lines
9.4 KiB
Python
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()
|