import copy import json from pathlib import Path import tempfile import unittest from unittest.mock import patch from paper_atlas.collectors.pmlr import collect_volume from paper_atlas.collectors.pmlr_tracks import program_tracks, record_track, position_list_policy from paper_atlas.ingestion.records import ingest_file from paper_atlas.ingestion.backfill import backfill def event(forum='realForum1', track='Conference', event_id=1, year=2025, **values): return {'id': event_id, 'sourceid': event_id, 'name': 'A published paper', 'sourceurl': f'https://openreview.net/group?id=ICML.cc/{year}/' + track, 'paper_url': 'https://openreview.net/forum?id=' + forum, 'decision': 'Accept (Poster)', **values} def program(*events, **changes): return json.dumps({'count': len(events), 'results': list(events), **changes}).encode() def record(forum='realForum1', external='v267/paper1', year=2025, title='A published paper'): return {'source': 'pmlr', 'external_id': external, 'venue': 'ICML', 'year': year, 'title': title, 'abstract': 'Original abstract', 'authors': ['Jane Doe', 'John Smith'], 'identifiers': {'openreview': forum}, 'metadata': {}, 'track': 'main'} class PmlrTrackTests(unittest.TestCase): def provenance(self): return {'url': 'https://icml.cc/static/virtual/data/icml-2025-orals-posters.json', 'sha256': 'a' * 64} def test_real_groups_not_title_prefixes_determine_modern_tracks(self): mapping, inventory = program_tracks(program( event(name='Position: still a main-track paper'), event('position1', 'Position_Paper_Track', 2, name='No prefix needed'), event('2025-Oral-fake', event_id=3, sourceid=-1), event('workshop1', 'Workshop', 4)), 2025) main = record(title='Position: still a main-track paper') position = record('position1', 'v267/paper2', title='No prefix needed') record_track(main, mapping, self.provenance()); record_track(position, mapping, self.provenance()) self.assertEqual((main['track'], position['track']), ('main', 'position')) self.assertEqual(inventory['paper_identity_count'], 2) self.assertEqual(inventory['excluded_events'], {'synthetic_oral_event': 1, 'other_source': 1}) def test_conflicting_forums_pagination_and_missing_real_ids_fail(self): for data in [program(event(), event(track='Position_Paper_Track', event_id=2)), program(event(), next='page2'), program(event(), count=2), program(event(paper_url='')), program(event(decision='Reject')), program(event(), event())]: with self.subTest(data=data), self.assertRaises(ValueError): program_tracks(data, 2025) def test_missing_publication_identity_has_no_main_track_fallback(self): mapping, _ = program_tracks(program(event()), 2025) with self.assertRaisesRegex(ValueError, 'absent'): record_track(record('unknownForum'), mapping, self.provenance()) def test_2024_requires_organizer_endorsed_list_and_publisher_self_links(self): html = b'position papers' policy = position_list_policy(html) with self.assertRaises(ValueError): position_list_policy(html.replace(b'2024', b'2023')) e = event(year=2024, name='Position: An explicit position', paper_url='', paper_pdf_url='https://proceedings.mlr.press/v235/paper1.html') with self.assertRaisesRegex(ValueError, 'policy'): program_tracks(program(e), 2024) mapping, _ = program_tracks(program(e), 2024, policy) row = record(external='v235/paper1', year=2024) record_track(row, mapping, {**self.provenance(), 'position_list_policy': policy}) self.assertEqual(row['track'], 'position') self.assertEqual(row['metadata']['track_evidence']['identity_scheme'], 'pmlr') def test_reviewed_2024_link_alias_still_needs_title_and_two_authors(self): e = event(year=2024, paper_url='', paper_pdf_url='https://proceedings.mlr.press/v235/theo24a.html', authors=[{'fullname': 'Jane Doe'}, {'fullname': 'John Smith'}]) mapping, _ = program_tracks(program(e), 2024, {'verified': True}) row = record(external='v235/guyard24a', year=2024) record_track(row, mapping, self.provenance()) self.assertEqual(row['metadata']['track_evidence']['corroborating_author_count'], 2) self.assertNotIn('_authors', row['metadata']['track_evidence']) for changed in [dict(title='Different title'), dict(authors=['Someone Else'])]: bad = record(external='v235/guyard24a', year=2024); bad.update(changed) with self.assertRaisesRegex(ValueError, 'corroboration'): record_track(bad, mapping, self.provenance()) def test_collector_splits_tracks_preserves_ids_and_imports_one_event_each(self): config = b'shortname: ICML\nyear: 2025\nbooktitle: ICML\n' template = '---\nid: {id}\nlayout: inproceedings\ntitle: {title}\nabstract: Original abstract\nopenreview: {forum}\n---\n' posts = [('a.md', template.format(id='paper1', title='Main paper', forum='realForum1')), ('b.md', template.format(id='paper2', title='Position paper', forum='position1'))] mapping, inventory = program_tracks(program(event(), event('position1', 'Position_Paper_Track', 2)), 2025) with tempfile.TemporaryDirectory() as tmp, patch('paper_atlas.collectors.pmlr.fetch', return_value=config), \ patch('paper_atlas.collectors.pmlr.metadata_texts', return_value=iter(posts)), \ patch('paper_atlas.collectors.pmlr.collect_track_evidence', return_value=(mapping, self.provenance())): result = collect_volume(267, tmp, tmp) self.assertNotIn('file', result) self.assertEqual(result['track'], 'mixed') self.assertEqual({m['track']: m['count'] for m in result['files']}, {'main': 1, 'position': 1}) self.assertEqual(result['count'], 2) db = Path(tmp) / 'papers.sqlite' for manifest in result['files']: ingest_file(db, manifest['file']) import sqlite3 with sqlite3.connect(db) as c: self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0], 2) self.assertEqual(c.execute('SELECT track,count(*) FROM submissions s JOIN editions e ON s.edition_id=e.id GROUP BY track').fetchall(), [('main', 1), ('position', 1)]) self.assertEqual(c.execute('PRAGMA foreign_key_check').fetchall(), []) def test_unknown_track_does_not_replace_existing_volume_output(self): config = b'shortname: ICML\nyear: 2025\nbooktitle: ICML\n' post = '---\nid: paper1\nlayout: inproceedings\ntitle: Paper\nopenreview: missing123\n---\n' with tempfile.TemporaryDirectory() as tmp, patch('paper_atlas.collectors.pmlr.fetch', return_value=config), \ patch('paper_atlas.collectors.pmlr.metadata_texts', return_value=iter([('a.md', post)])), \ patch('paper_atlas.collectors.pmlr.collect_track_evidence', return_value=({}, self.provenance())): old = Path(tmp) / 'pmlr-v267.jsonl'; old.write_text('previous collection') with self.assertRaisesRegex(ValueError, 'absent'): collect_volume(267, tmp, tmp) self.assertEqual(old.read_text(), 'previous collection') self.assertFalse((Path(tmp) / 'pmlr-v267.inventory.json').exists()) def test_old_mixed_manifest_is_rejected_before_opening_database(self): with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / 'pmlr-v267.jsonl'; path.write_text('') path.with_suffix('.manifest.json').write_text(json.dumps({'source': 'pmlr', 'venue': 'ICML', 'year': 2025, 'parser_version': 2})) db = Path(tmp) / 'absent.sqlite' for action in [lambda: ingest_file(db, path), lambda: backfill(db, [path])]: with self.assertRaisesRegex(ValueError, 'Recollect mixed'): action() self.assertFalse(db.exists()) def test_changed_track_or_own_identity_rolls_back_import(self): from paper_atlas.collectors.pmlr_tracks import require_track_record row = record(); mapping, _ = program_tracks(program(event()), 2025) record_track(row, mapping, self.provenance()) manifest = {'source': 'pmlr', 'venue': 'ICML', 'year': 2025, 'parser_version': 3, 'track_program': self.provenance()} require_track_record(row, manifest) for change in [dict(track='position'), dict(identifiers={'openreview': 'otherForum'})]: bad = copy.deepcopy(row); bad.update(change) with self.assertRaises(ValueError): require_track_record(bad, manifest) if __name__ == '__main__': unittest.main()