187 lines
11 KiB
Python
187 lines
11 KiB
Python
"""Frozen publisher examples, conflicting evidence and preserved attempt identity."""
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from paper_atlas.collectors import rss
|
|
from paper_atlas.ingestion.records import ingest_file
|
|
|
|
FIXTURES = Path(__file__).parent/'fixtures/rss2025'
|
|
OBSERVED = '2026-11-09T10:00:00Z'
|
|
|
|
|
|
def fixture(name):
|
|
return (FIXTURES/name).read_text()
|
|
|
|
|
|
class RssProceedingsTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.index = rss.parse_index(fixture('index.html'))
|
|
self.accepted = rss.parse_program(fixture('accepted.html'), 'Accepted Papers', ('ID', 'Session', 'Title', 'Authors'))
|
|
self.demos = rss.parse_program(fixture('demos.html'), 'Demos', ('ID', 'Location', 'Title', 'Day', 'Time'))
|
|
|
|
def paper(self, pid=1, html=None, entry=None):
|
|
return rss.parse_paper(html or fixture(f'p{pid:03d}.html'), pid, entry or self.index[pid],
|
|
self.accepted[pid], self.demos.get(pid), {'paper': {'observed_at': OBSERVED}})
|
|
|
|
def test_frozen_official_census_and_all_fixture_hashes(self):
|
|
receipts = json.loads((FIXTURES/'receipts.json').read_bytes())['receipts']
|
|
self.assertEqual(len(receipts), 6)
|
|
for r in receipts:
|
|
data = (FIXTURES/r['file']).read_bytes()
|
|
self.assertEqual((len(data), hashlib.sha256(data).hexdigest()), (r['bytes'], r['sha256']))
|
|
self.assertEqual(len(self.index), 163)
|
|
self.assertEqual(set(self.index), set(self.accepted))
|
|
self.assertEqual(len(self.demos), 15)
|
|
rss.validate_program(self.index, self.accepted, self.demos, fixture('authorinfo.html'))
|
|
self.assertFalse(self.demos[93]['Title'].startswith('Demonstrating'))
|
|
|
|
def test_complete_authors_abstract_and_source_metadata_are_preserved(self):
|
|
row = self.paper()
|
|
self.assertEqual(row['authors'], ['Pascal Roth', 'Jonas Frey', 'Cesar Cadena', 'Marco Hutter'])
|
|
self.assertTrue(row['abstract'].startswith('Ensuring safe navigation'))
|
|
self.assertTrue(row['abstract'].endswith('https://github.com/leggedrobotics/fdm.'))
|
|
self.assertNotIn('Download:', row['abstract'])
|
|
self.assertEqual(row['identifiers'], {'doi': '10.15607/rss.2025.xxi.001'})
|
|
self.assertEqual(row['metadata']['publication_date'], '2025/06/21')
|
|
self.assertIn('AUTHOR', row['metadata']['original_bibtex'])
|
|
self.assertEqual(row['status'], 'accepted')
|
|
self.assertNotIn('decision_date', row)
|
|
self.assertEqual(row['metadata']['decision_date_precision'], 'not_supplied')
|
|
|
|
def test_demo_evidence_is_independent_of_title_and_session_topic(self):
|
|
row = self.paper(2)
|
|
self.assertEqual(row['track'], 'demo')
|
|
self.assertEqual(row['metadata']['demo_program_entry']['ID'], '2')
|
|
self.assertEqual(self.paper()['track'], 'proceedings_unspecified')
|
|
self.assertEqual({t['namespace'] for t in row['tags']}, {'rss_proceedings:session', 'rss_proceedings:program_role'})
|
|
self.assertFalse(any('research_type' in t['namespace'] for t in row['tags']))
|
|
html = fixture('p001.html').replace('Learned Perceptive Forward Dynamics Model', 'Demonstrating a Learned Model')
|
|
entry = {**self.index[1], 'title': self.index[1]['title'].replace('Learned Perceptive Forward Dynamics Model', 'Demonstrating a Learned Model')}
|
|
self.assertEqual(self.paper(html=html, entry=entry)['track'], 'proceedings_unspecified')
|
|
|
|
def test_missing_abstract_is_explicit_and_multiple_paragraphs_are_complete(self):
|
|
html = fixture('p001.html')
|
|
beginning = html.index('<p style="text-align: justify;">')
|
|
end = html.index('</p>', beginning)
|
|
empty = html[:beginning]+'<p style="text-align: justify;">'+html[end:]
|
|
row = self.paper(html=empty)
|
|
self.assertIsNone(row['abstract'])
|
|
self.assertEqual(row['metadata']['missing_abstract_reason'], 'empty_official_abstract_section')
|
|
multi = html.replace('</p> \n<p> \n<b>Download:', '</p><p>Second abstract paragraph.</p><p><b>Download:', 1)
|
|
self.assertTrue(self.paper(html=multi)['abstract'].endswith('Second abstract paragraph.'))
|
|
|
|
def test_metadata_drift_author_truncation_and_wrong_paper_doi_fail(self):
|
|
original = fixture('p001.html')
|
|
for altered in [original.replace('XXI.001}', 'XXI.002}'),
|
|
original.replace('content="2025/06/21"', 'content="2026/06/21"'),
|
|
original.replace('<meta name="citation_author" content="Marco Hutter" />', ''),
|
|
original.replace('<b>Abstract:</b>', '<b>Summary:</b>'),
|
|
original.replace('rss21/p001.pdf', 'rss21/p002.pdf'),
|
|
original.replace('citation_volume" content="21', 'citation_volume" content="22')]:
|
|
with self.subTest(altered=altered[:60]), self.assertRaises(ValueError):
|
|
self.paper(html=altered)
|
|
|
|
def test_duplicate_and_unparsed_index_entries_fail(self):
|
|
for html in [fixture('index.html').replace('p002.html', 'p001.html'),
|
|
fixture('index.html')+'<a href="p999.html">Stray paper</a>',
|
|
fixture('index.html').replace( rss.HEADING, 'A different volume')]:
|
|
with self.subTest(html=html[:40]), self.assertRaises(ValueError):
|
|
rss.parse_index(html)
|
|
|
|
def test_program_title_id_category_and_header_changes_require_review(self):
|
|
for change in ['title', 'id', 'category']:
|
|
index = {k: dict(v) for k, v in self.index.items()}
|
|
demos = {k: dict(v) for k, v in self.demos.items()}
|
|
if change == 'title':
|
|
demos[2]['Title'] = 'Another paper'
|
|
elif change == 'id':
|
|
del index[1]
|
|
else:
|
|
del demos[93]
|
|
with self.subTest(change=change), self.assertRaises(ValueError):
|
|
rss.validate_program(index, self.accepted, demos, fixture('authorinfo.html'))
|
|
with self.assertRaises(ValueError):
|
|
rss.parse_program(fixture('demos.html').replace('>Location<', '>Other<'), 'Demos', ('ID', 'Location', 'Title', 'Day', 'Time'))
|
|
|
|
def snapshot(self, root):
|
|
root.mkdir()
|
|
data = (FIXTURES/'p001.html').read_bytes()
|
|
(root/'paper.html').write_bytes(data)
|
|
receipt = {'file': 'paper.html', 'url': self.index[1]['url'], 'final_url': self.index[1]['url'],
|
|
'http_status': 200, 'content_type': 'text/html', 'bytes': len(data),
|
|
'sha256': hashlib.sha256(data).hexdigest(), 'at': OBSERVED}
|
|
(root/'receipts.json').write_text(json.dumps({'receipts': [receipt]}))
|
|
return receipt
|
|
|
|
def test_frozen_receipt_hash_status_redirect_and_path_guards(self):
|
|
for failure in ['hash', 'status', 'redirect', 'path', 'duplicate', 'symlink']:
|
|
with self.subTest(failure=failure), tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)/'source'
|
|
r = self.snapshot(root)
|
|
receipts = [r]
|
|
if failure == 'hash': r['sha256'] = '0'*64
|
|
elif failure == 'status': r['http_status'] = 404
|
|
elif failure == 'redirect': r['final_url'] = 'https://example.org/p001.html'
|
|
elif failure == 'path': r['file'] = '../outside.html'
|
|
elif failure == 'duplicate': receipts.append(dict(r))
|
|
else:
|
|
(root.parent/'outside.html').write_bytes((root/'paper.html').read_bytes())
|
|
(root/'paper.html').unlink()
|
|
(root/'paper.html').symlink_to(root.parent/'outside.html')
|
|
(root/'receipts.json').write_text(json.dumps({'receipts': receipts}))
|
|
with self.assertRaises(ValueError): rss.Pages(snapshots=root).read(self.index[1]['url'])
|
|
|
|
def test_unreviewed_edition_and_existing_output_refuse_work(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
with self.assertRaises(ValueError): rss.collect(2026, tmp, snapshots=FIXTURES)
|
|
(Path(tmp)/'rss-2025.json').write_text('{}')
|
|
with self.assertRaises(FileExistsError): rss.collect(2025, tmp, snapshots=FIXTURES)
|
|
|
|
def test_native_roundtrip_is_idempotent_with_same_doi_and_distinct_tracks(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
original = self.paper()
|
|
dblp = {**original, 'source': 'dblp', 'external_id': 'conf/rss/example', 'metadata': {}, 'tags': []}
|
|
variant = {**original, 'external_id': '2025/other-track', 'track': 'workshop'}
|
|
for n, record in enumerate([dblp, original, variant, original]):
|
|
path = root/f'{n}.jsonl'
|
|
path.write_text(json.dumps(record)+'\n')
|
|
path.with_suffix('.manifest.json').write_text(json.dumps({'source': record['source'], 'list_url': rss.ROOT, 'count': 1}))
|
|
ingest_file(root/'sample.sqlite', path)
|
|
with sqlite3.connect(root/'sample.sqlite') as c:
|
|
self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0], 1)
|
|
self.assertEqual(c.execute('SELECT count(*) FROM source_records').fetchone()[0], 3)
|
|
self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0], 2)
|
|
self.assertEqual(c.execute('SELECT count(*) FROM decisions').fetchone()[0], 2)
|
|
self.assertEqual(c.execute('SELECT count(*) FROM submission_sources').fetchone()[0], 3)
|
|
self.assertEqual(c.execute('PRAGMA foreign_key_check').fetchall(), [])
|
|
|
|
def test_two_real_papers_collection_commits_evidence_only_after_success(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
index = {pid: self.index[pid] for pid in [1, 2]}
|
|
accepted = {pid: self.accepted[pid] for pid in [1, 2]}
|
|
demos = {2: self.demos[2]}
|
|
with (patch.object(rss, 'EXPECTED', 2), patch.object(rss, 'DEMO_IDS', {2}),
|
|
patch.object(rss, 'parse_index', return_value=index),
|
|
patch.object(rss, 'parse_program', side_effect=[accepted, demos])):
|
|
result = rss.collect(2025, root/'out', snapshots=FIXTURES)
|
|
self.assertEqual((result['records'], len(result['files'])), (2, 2))
|
|
self.assertFalse(result['database_written'])
|
|
self.assertFalse(result['pdfs_downloaded'])
|
|
self.assertEqual(len(result['source_evidence']), 6)
|
|
for item in result['source_evidence']:
|
|
self.assertEqual(hashlib.sha256((root/'out'/item['file']).read_bytes()).hexdigest(), item['sha256'])
|
|
for item in result['files']:
|
|
self.assertEqual(item['missing_abstract_count'], 0)
|
|
self.assertEqual(hashlib.sha256(Path(item['file']).read_bytes()).hexdigest(), item['file_sha256'])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|