"""Real 49-chapter snapshots compared with a separately frozen metadata audit.""" import copy import gzip import hashlib import json from pathlib import Path import sqlite3 import tempfile import unittest from paper_atlas.collectors.springer_selected import ( BOOK, SOURCE, collect, parse_chapter, parse_toc, read_page, snapshot_index, ) from paper_atlas.db.store import Store from paper_atlas.ingestion.records import ingest_file FIXTURE = Path(__file__).parent/'fixtures/springer-aistats1993-fixture.json.gz' class SpringerSelectedTests(unittest.TestCase): @classmethod def setUpClass(cls): raw = FIXTURE.read_bytes() if hashlib.sha256(raw).hexdigest() != '94028fd37080f9d8027e5790e379cc151ee6eab17ed0d1dcc0f59e6e75f9d4d0': raise AssertionError('Fixed source fixture differs') cls.fixture = json.loads(gzip.decompress(raw)) def snapshots(self, root): for group in ['toc','chapters']: folder = root/group folder.mkdir() for name, text in self.fixture[group].items(): (folder/name).write_text(text) return {'toc_snapshots':root/'toc','chapter_snapshots':root/'chapters'} def test_all49_original_fields_against_independent_audit(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) report = collect(root/'out', **self.snapshots(root)) manifest, = report['files'] rows = [json.loads(s) for s in Path(manifest['file']).read_text().splitlines()] self.assertEqual(len(rows), 49) self.assertEqual(manifest['status'], 'partial') self.assertEqual(manifest['missing_abstract_count'], 0) self.assertTrue(manifest['excluded_materials']) self.assertTrue(all(x['title'] == 'Front Matter' for x in manifest['excluded_materials'])) for row, expected in zip(rows, self.fixture['expected']): for field in ['title','abstract']: self.assertEqual(row[field], expected[field]) self.assertEqual(row['identifiers'], {'doi':expected['doi']}) self.assertEqual(row['submission_identifiers'], [{'scheme':'publication_doi','value':expected['doi']}]) self.assertEqual(row['external_id'], expected['doi']) self.assertEqual(row['metadata']['authors_citation_raw'], expected['authors_citation_raw']) self.assertEqual(row['authors'], [{'family':s.split(',',1)[0].strip(), 'given':s.split(',',1)[1].strip()} for s in expected['authors_citation_raw']]) self.assertEqual(row['metadata']['first_page'], expected['first_page']) self.assertEqual(row['metadata']['last_page'], expected['last_page']) self.assertEqual((row['year'],row['metadata']['bibliographic_year']), (1993,1994)) self.assertEqual(row['track'], 'selected_volume') self.assertEqual(row['event_key'], 'aistats-1993-selected-volume-1994') self.assertNotIn('decision_date', row) self.assertIn('original 1993 review decision not supplied', row['raw_decision']) for e in [row['metadata']['source_evidence'], *row['metadata']['scope_evidence']]: self.assertEqual(hashlib.sha256((root/'out'/e['file']).read_bytes()).hexdigest(), e['sha256']) self.assertFalse(report['database_written']) with self.assertRaises(FileExistsError): collect(root/'out', toc_snapshots=root/'toc', chapter_snapshots=root/'chapters') def test_native_history_namespaces_and_idempotent_import(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) report = collect(root/'out', **self.snapshots(root)) manifest, = report['files'] first = json.loads(Path(manifest['file']).read_text().splitlines()[0]) db = root/'sample.sqlite' store = Store(db) pid, rid = store.record('prior_public','earlier','https://example.org/earlier',first['title'],first['abstract'],first['authors']) store.decision(pid,rid,'AISTATS',1993,'main','earlier','https://example.org/earlier','rejected','original public rejection') # Explicitly verified work identity; the short title alone must not # force a merge or attach this publication DOI to the old attempt. store.conn.execute('INSERT INTO identifiers VALUES(?,?,?)', ('doi',first['identifiers']['doi'],pid)) store.conn.commit() old = store.conn.execute('SELECT * FROM decisions').fetchall() store.conn.close() ingest_file(db, manifest['file']) ingest_file(db, manifest['file']) with sqlite3.connect(db) as c: self.assertEqual(c.execute('SELECT * FROM decisions WHERE status=\'rejected\'').fetchall(), [tuple(r) for r in old]) self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0], 49) self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0], 50) self.assertEqual(c.execute('SELECT count(*) FROM decisions').fetchone()[0], 50) self.assertEqual(c.execute('SELECT e.year,e.track,e.event_key FROM editions e ORDER BY e.id').fetchall(), [(1993,'main',''),(1993,'selected_volume','aistats-1993-selected-volume-1994')]) self.assertEqual(c.execute('SELECT count(*) FROM paper_tags pt JOIN tags t ON t.id=pt.tag_id WHERE t.namespace=?', (SOURCE+':publication_format',)).fetchone()[0], 49) self.assertEqual(c.execute('SELECT count(*) FROM submission_identifiers WHERE scheme=\'publication_doi\'').fetchone()[0], 49) self.assertIsNone(c.execute("SELECT 1 FROM submission_identifiers si JOIN submissions s ON s.id=si.submission_id JOIN editions e ON e.id=s.edition_id WHERE e.track='main'").fetchone()) def test_short_unanchored_title_keeps_identity_unconfirmed(self): with tempfile.TemporaryDirectory() as tmp: store = Store(Path(tmp)/'identity.sqlite') title = 'Statistical strategy: step 1' authors = [{'given':'D. J.','family':'Hand'}] old, _ = store.record('prior_public','earlier','https://example.org/old',title,'Same abstract',authors) new, _ = store.record(SOURCE,BOOK+'_1','https://link.springer.com/chapter/'+BOOK+'_1',title,'Same abstract',authors,{'doi':BOOK+'_1'}) self.assertNotEqual(old, new) self.assertEqual(store.conn.execute('SELECT count(*) FROM match_candidates WHERE method=\'exact_title_unconfirmed\'').fetchone()[0], 1) store.conn.close() def test_receipt_hash_path_redirect_and_content_type_guards(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) self.snapshots(root) index = snapshot_index(root/'chapters') url = 'https://link.springer.com/chapter/'+BOOK+'_1' self.assertTrue(read_page(index,url)[0]) for key, value in [('sha256','0'*64),('file','../escape.html'),('final_url','https://example.org/chapter'),('http_status',403),('headers',{'Content-Type':'application/pdf'})]: changed = copy.deepcopy(index) changed[url][1][key] = value with self.subTest(key=key), self.assertRaises((ValueError,KeyError)): read_page(changed,url) changed = copy.deepcopy(index) changed[url][1]['final_url'] = url+'?page=2' with self.assertRaises(ValueError): read_page(changed,url) def test_missing_or_duplicate_snapshot_stops_collection(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) args = self.snapshots(root) path = root/'chapters/receipts.json' d = json.loads(path.read_bytes()) original = copy.deepcopy(d) d['responses'].pop() path.write_text(json.dumps(d)) with self.assertRaises(ValueError): collect(root/'out', **args) self.assertFalse((root/'out').exists()) original['responses'].append(original['responses'][0]) path.write_text(json.dumps(original)) with self.assertRaises(ValueError): snapshot_index(root/'chapters') def test_book_doi_year_and_duplicate_citation_fields_reject_drift(self): html = self.fixture['chapters']['chapter-01.html'] for key, value in [('citation_doi',BOOK+'_2'),('citation_inbook_title','Another Book'),('citation_publication_date','1993')]: import re changed, count = re.subn(r'(]*name="'+key+r'"[^>]*content=")[^"]*', lambda m:m[1]+value, html) self.assertEqual(count, 1) with self.subTest(key=key), self.assertRaises(ValueError): parse_chapter(changed, BOOK+'_1') with self.assertRaises(ValueError): parse_chapter(html+'', BOOK+'_1') def test_missing_abstract_is_explicit_and_does_not_use_other_page_text(self): html = self.fixture['chapters']['chapter-01.html'].replace('id="Abs1-content"','id="not-an-abstract"') self.assertIsNone(parse_chapter(html, BOOK+'_1')['abstract']) def test_toc_ignores_frontmatter_but_rejects_duplicate_publications(self): html = self.fixture['toc']['springer-toc1.html'] self.assertEqual(len(parse_toc(html)), 17) with self.assertRaises(ValueError): parse_toc(html+html)