99 lines
5.9 KiB
Python
99 lines
5.9 KiB
Python
import gzip
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from paper_atlas.collectors import rss2026 as rss
|
|
|
|
FIXTURE = Path(__file__).parent/'fixtures/rss2026-source.json.gz'
|
|
|
|
class Rss2026Tests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.fixture=json.loads(gzip.decompress(FIXTURE.read_bytes()))
|
|
|
|
def setUp(self):
|
|
self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup)
|
|
self.root=Path(self.tmp.name);self.snap=self.root/'snapshots';self.snap.mkdir()
|
|
for name,text in self.fixture['pages'].items():(self.snap/name).write_bytes(text.encode())
|
|
(self.snap/'receipts.json').write_text(json.dumps(self.fixture['receipts']))
|
|
self.pages=self.fixture['pages'];self.index,self.program=self.lists()
|
|
|
|
def lists(self,**changes):
|
|
pages={**self.pages,**changes}
|
|
return rss.parse_lists(*(pages[n] for n in rss.PAGES))
|
|
|
|
def test_all210_records_match_independent_primary_inventory_and_evidence(self):
|
|
report=rss.collect(self.root/'out',snapshots=self.snap)
|
|
self.assertEqual(report['records'],210);self.assertFalse(report['database_written'])
|
|
self.assertFalse(report['pdfs_downloaded']);self.assertEqual(len(report['source_evidence']),214)
|
|
manifest,=report['files'];self.assertEqual(manifest['status'],'partial')
|
|
rows=[json.loads(l) for l in Path(manifest['file']).read_text().splitlines()]
|
|
for row,gold in zip(rows,self.fixture['independent_gold']):
|
|
self.assertEqual(row['external_id'],f"2026/p{gold['paper_id']:03d}")
|
|
for key in ('title','authors','abstract'):self.assertEqual(row[key],gold[key])
|
|
self.assertEqual(row['identifiers'],{'doi':gold['doi_raw'].lower()})
|
|
self.assertEqual(row['metadata']['publisher_date'],gold['publication_date'])
|
|
self.assertNotIn('submission_identifiers',row);self.assertEqual(row['track'],'main')
|
|
e=row['metadata']['source_evidence'];self.assertEqual(hashlib.sha256((self.root/'out'/e['file']).read_bytes()).hexdigest(),e['sha256'])
|
|
self.assertEqual(manifest['file_sha256'],hashlib.sha256(Path(manifest['file']).read_bytes()).hexdigest())
|
|
self.assertEqual(manifest['missing_abstract_count'],0)
|
|
|
|
def test_full_hidden_authors_are_separate_from_six_short_previews(self):
|
|
shortened=[p for p in self.program.values() if p['authors_complete']!=p['authors_preview']]
|
|
self.assertEqual(len(shortened),6)
|
|
p=rss.parse_paper(self.pages['p087.html'],87,self.index[87],self.program[87])
|
|
self.assertEqual(len(p['authors']),55)
|
|
self.assertFalse(p['program_variants'])
|
|
|
|
def test_six_reviewed_name_variants_preserve_original_strings(self):
|
|
for pid in rss.REVIEWED_AUTHOR_VARIANTS:
|
|
p=rss.parse_paper(self.pages[f'p{pid:03d}.html'],pid,self.index[pid],self.program[pid])
|
|
self.assertIn('reviewed_program_author_name_variant_not_person_alias',p['program_variants'])
|
|
self.assertNotEqual(', '.join(p['authors']),self.program[pid]['authors_complete'])
|
|
|
|
def test_reviewed_title_variant_is_not_an_automatic_erratum(self):
|
|
p=rss.parse_paper(self.pages['p089.html'],89,self.index[89],self.program[89])
|
|
self.assertTrue(p['title'].startswith('RLinf-VLA'))
|
|
self.assertTrue(self.program[89]['title'].startswith('RLux-VLA'))
|
|
self.assertIn('reviewed_program_title_variant_not_erratum',p['program_variants'])
|
|
|
|
def test_unreviewed_author_or_title_variant_stops(self):
|
|
for changed in [{'authors_complete':'An Unrelated Person'},{'title':'A different paper'}]:
|
|
with self.assertRaisesRegex(ValueError,'Unreviewed'):
|
|
rss.parse_paper(self.pages['p001.html'],1,self.index[1],{**self.program[1],**changed})
|
|
|
|
def test_demo_policy_cannot_be_reused_from2025(self):
|
|
with self.assertRaisesRegex(ValueError,'policy'):
|
|
self.lists(**{'cfp.html':self.pages['cfp.html'].replace('will not have a separate category','will have a separate category')})
|
|
with self.assertRaises(ValueError):
|
|
self.lists(**{'sponsor-demos.html':self.pages['sponsor-demos.html'].replace('Sponsor Demos','Demo Papers')})
|
|
|
|
def test_wrong_doi_year_pdf_or_missing_abstract_is_rejected(self):
|
|
html=self.pages['p001.html']
|
|
for old,new in [('10.15607/RSS.2026.XXII.001','10.15607/RSS.2025.XXI.001'),
|
|
('2026/07/13','2025/07/13'),('rss22/p001.pdf','rss22/p002.pdf'),('Abstract:','Summary:')]:
|
|
self.assertIn(old,html)
|
|
with self.assertRaises(ValueError):rss.parse_paper(html.replace(old,new),1,self.index[1],self.program[1])
|
|
|
|
def test_index_census_and_preview_tampering_stop(self):
|
|
with self.assertRaises(ValueError):self.lists(**{'index.html':self.pages['index.html'].replace('p001.html','p999.html')})
|
|
bad=self.pages['accepted.html'].replace('Huayi Zhou, Kui Jia','Somebody Else',1)
|
|
with self.assertRaisesRegex(ValueError,'prefix'):self.lists(**{'accepted.html':bad})
|
|
|
|
def test_snapshot_tamper_has_no_completion_marker_and_never_fetches(self):
|
|
p=self.snap/'p001.html';p.write_bytes(p.read_bytes()+b'changed')
|
|
with patch.object(rss,'fetch',side_effect=AssertionError('network must not be used')):
|
|
with self.assertRaisesRegex(ValueError,'receipt'):rss.collect(self.root/'out',snapshots=self.snap)
|
|
self.assertFalse((self.root/'out/collection-report.json').exists())
|
|
self.assertTrue((self.root/'out/failed.json').exists())
|
|
|
|
def test_existing_output_and_unsafe_receipt_are_rejected(self):
|
|
out=self.root/'existing';out.mkdir()
|
|
with self.assertRaises(ValueError):rss.collect(out,snapshots=self.snap)
|
|
receipts=[*self.fixture['receipts'],{**self.fixture['receipts'][0],'file':'../outside'}]
|
|
(self.snap/'receipts.json').write_text(json.dumps(receipts))
|
|
with self.assertRaisesRegex(ValueError,'Unsafe'):rss.collect(self.root/'unsafe',snapshots=self.snap)
|