149 lines
8.8 KiB
Python
149 lines
8.8 KiB
Python
"""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'<li><div><a href="http://arxiv.org/abs/{identifier}">{title}</a>, {display}</div></li>'
|
|
|
|
|
|
def page2015():
|
|
return ('<h3>Main Conference - Oral Presentations</h3><ol>'+entry('1412.0001v2')+'</ol>'
|
|
'<h3>Main Conference - Poster Presentations</h3><ol>'+entry('1412.0001v2')+'</ol>'
|
|
'<h3>Workshop Papers</h3><ol>'+entry('1412.0002', 'A workshop paper')+'</ol>')
|
|
|
|
|
|
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 = ('<h3>Conference Proceedings</h3><p>conference papers accepted</p>'
|
|
'<div><b><a href="http://arxiv.org/abs/1312.0001">A <i>research</i> title</a></b></div>'
|
|
'<p><span style="font-style: italic">Jane Doe; </span>'
|
|
'<span style="font-style:italic">Bob Smith; Carol Wu;</span></p>'
|
|
'<div><a href="http://arxiv.org/abs/1312.0002">Another title</a>'
|
|
'<div><i><span>First Author; </span><span>Second Author</span></i></div></div>')
|
|
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 = ('<h1>Accepted Papers (Conference Track)</h1><h3>Oral Presentations</h3><ol>'
|
|
'<li><a href="http://arxiv.org/abs/1511.00001">A title</a>'
|
|
'<a href="https://example.org/code">[code]</a><br>Jane Doe, Bob Smith</li></ol>'
|
|
'<h3>Poster Presentations</h3>')
|
|
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('</ol>', '</ol><a href="http://arxiv.org/abs/1412.0003">stray</a>', 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(), [])
|