161 lines
9.5 KiB
Python
161 lines
9.5 KiB
Python
"""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'(<meta[^>]*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+'<meta name="citation_doi" content="'+BOOK+'_1">', 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)
|