"""Historical accepted lists, repeated presentations and preserved work histories."""
import hashlib
import json
from pathlib import Path
import sqlite3
import tempfile
import unittest
from unittest.mock import patch
from paper_atlas.collectors import iclr_archive as archive
from paper_atlas.db.store import Store
from paper_atlas.ingestion.records import ingest_file
OBSERVED = '2026-11-09T00:00:00Z'
def entry(identifier, title='A shared research paper', display='Jane Doe and Bob Smith'):
return f'
'
def page2015():
return ('Main Conference - Oral Presentations
'+entry('1412.0001v2')+'
'
'Main Conference - Poster Presentations
'+entry('1412.0001v2')+'
'
'Workshop Papers
'+entry('1412.0002', 'A workshop paper')+'
')
class IclrArchiveTests(unittest.TestCase):
def frozen(self, root, year, html):
root.mkdir()
data = html.encode()
(root/'page.html').write_bytes(data)
receipt = {'url': archive.PAGES[year][0], 'file': 'page.html', 'status': 200,
'bytes': len(data), 'sha256': hashlib.sha256(data).hexdigest(), 'finished_at': OBSERVED}
(root/'receipts.json').write_text(json.dumps({'receipts': [receipt]}))
return root
def test_oral_poster_repeat_is_one_attempt_with_all_source_variants(self):
rows = archive.parse_page(page2015(), 2015, {'observed_at': OBSERVED})
main, workshop = rows
self.assertEqual((main['track'], workshop['track']), ('main', 'workshop'))
self.assertEqual(main['identifiers'], {'arxiv': '1412.0001'})
self.assertEqual(main['external_id'], '2015/main/1412.0001')
self.assertEqual(main['metadata']['presentations'], ['oral', 'poster'])
self.assertEqual(len(main['metadata']['source_variants']), 2)
self.assertEqual(main['metadata']['source_variants'][0]['version'], 'v2')
self.assertEqual(main['authors'], ['Jane Doe', 'Bob Smith'])
self.assertTrue(all(r['status'] == 'accepted' and r['abstract'] is None for r in rows))
def test_2014_author_span_outside_paragraph_and_trailing_separator(self):
html = ('Conference Proceedings
conference papers accepted
'
''
'Jane Doe; '
'Bob Smith; Carol Wu;
'
'')
row, unwrapped = archive.parse_page(html, 2014, {'observed_at': OBSERVED})
self.assertEqual(row['title'], 'A research title')
self.assertEqual(row['authors'], ['Jane Doe', 'Bob Smith', 'Carol Wu'])
self.assertEqual(row['metadata']['source_variants'][0]['authors_display'], 'Jane Doe; Bob Smith; Carol Wu;')
self.assertEqual(row['track'], 'main')
self.assertEqual(unwrapped['authors'], ['First Author', 'Second Author'])
def test_same_arxiv_in_different_tracks_keeps_separate_source_attempts(self):
html = page2015().replace('1412.0002', '1412.0001v3')
rows = archive.parse_page(html, 2015, {'observed_at': OBSERVED})
self.assertEqual(len(rows), 2)
self.assertEqual({r['external_id'] for r in rows}, {'2015/main/1412.0001', '2015/workshop/1412.0001'})
self.assertEqual(rows[1]['metadata']['source_variants'][0]['version'], 'v3')
def test_2016_code_links_and_breaks_do_not_become_authors(self):
html = ('Accepted Papers (Conference Track)
Oral Presentations
'
'- A title'
'[code]
Jane Doe, Bob Smith
'
'Poster Presentations
')
row, = archive.parse_page(html, 2016, {'observed_at': OBSERVED})
self.assertEqual(row['authors'], ['Jane Doe', 'Bob Smith'])
self.assertEqual(row['metadata']['presentations'], ['oral'])
def test_conflicting_repeat_unknown_sections_and_unparsed_links_fail(self):
variants = [page2015().replace('Poster Presentations', 'Other Presentations'),
page2015().replace('', 'stray', 1),
page2015().replace('Jane Doe and Bob Smith', 'Different Author', 1)]
for html in variants:
with self.subTest(html=html), self.assertRaises(ValueError):
archive.parse_page(html, 2015, {'observed_at': OBSERVED})
def test_arxiv_versions_are_preserved_and_unsafe_links_rejected(self):
self.assertEqual(archive.arxiv_link('https://arxiv.org/pdf/1511.00001v3.pdf'), ('1511.00001', 'v3'))
self.assertIsNone(archive.arxiv_link('https://arxiv.org.example.org/abs/1511.00001'))
for url in ['https://arxiv.org/abs/1511.00001?other=1', 'ftp://arxiv.org/abs/1511.00001',
'https://user@arxiv.org/abs/1511.00001', 'https://arxiv.org/abs/not-a-paper']:
with self.subTest(url=url), self.assertRaises(ValueError):
archive.arxiv_link(url)
def test_receipt_hash_status_path_and_symlink_cannot_escape_snapshot(self):
for failure in ['hash', 'status', 'path', 'symlink']:
with self.subTest(failure=failure), tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
source = self.frozen(root/'source', 2015, page2015())
file = source/'receipts.json'
data = json.loads(file.read_text())
receipt = data['receipts'][0]
if failure == 'hash': receipt['sha256'] = '0'*64
elif failure == 'status': receipt['status'] = 404
elif failure == 'path': receipt['file'] = '../outside.html'
else:
(root/'outside.html').write_text(page2015())
(source/'page.html').unlink()
(source/'page.html').symlink_to(root/'outside.html')
file.write_text(json.dumps(data))
with self.assertRaises(ValueError):
archive.read_page(2015, snapshots=source)
def test_changed_inventory_leaves_no_completed_output(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
source = self.frozen(root/'source', 2015, page2015())
with self.assertRaises(ValueError):
archive.collect(2015, root/'out', snapshots=source)
self.assertFalse((root/'out/iclr-archive-2015.json').exists())
def test_native_import_keeps_rejected_to_accepted_history_and_abstract(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
source = self.frozen(root/'source', 2015, page2015())
with patch.dict(archive.PAGES, {2015: (archive.PAGES[2015][0], {'main': 1, 'workshop': 1})}):
result = archive.collect(2015, root/'out', snapshots=source)
with self.assertRaises(FileExistsError):
archive.collect(2015, root/'out', snapshots=source)
db = root/'sample.sqlite'
store = Store(db)
pid, rid = store.record('arxiv', '1412.0001', 'https://arxiv.org/abs/1412.0001',
'A shared research paper', 'Keep this canonical abstract.',
['Jane Doe', 'Bob Smith'], {'arxiv': '1412.0001'}, {})
store.decision(pid, rid, 'Earlier Conference', 2014, 'main', 'original-attempt',
'https://example.org/rejected', 'rejected', 'explicit earlier decision', observed_at=OBSERVED)
store.conn.commit()
store.conn.close()
for manifest in result['files']:
ingest_file(db, manifest['file'])
ingest_file(db, manifest['file'])
with sqlite3.connect(db) as c:
self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0], 2)
self.assertEqual(c.execute('SELECT abstract FROM papers WHERE id=?', (pid,)).fetchone()[0], 'Keep this canonical abstract.')
history = c.execute('SELECT v.acronym,e.year,d.status FROM decisions d JOIN submissions s ON s.id=d.submission_id JOIN editions e ON e.id=s.edition_id JOIN venues v ON v.id=e.venue_id WHERE s.paper_id=? ORDER BY e.year', (pid,)).fetchall()
self.assertEqual(history, [('Earlier Conference', 2014, 'rejected'), ('ICLR', 2015, 'accepted')])
self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0], 3)
self.assertEqual(c.execute('PRAGMA foreign_key_check').fetchall(), [])