186 lines
11 KiB
Python
186 lines
11 KiB
Python
import copy
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
|
|
from paper_atlas.curation.identifiers import SCOPE, digest, json_bytes, prepare
|
|
from paper_atlas.db.store import Store
|
|
from paper_atlas.ingestion.records import ingest_file
|
|
|
|
|
|
class IdentifierCurationTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.root = Path(self.tmp.name)
|
|
self.output = self.root / 'prepared'
|
|
self.row = {'source':'dblp','external_id':'conf/uai/example','url':'https://example.org/paper',
|
|
'title':'A sufficiently long example research paper title','abstract':None,
|
|
'authors':['Jane Doe','John Smith'],'venue':'UAI','year':2001,'track':'main',
|
|
'status':'accepted','tags':[{'namespace':'dblp:topic','value':'version\u2029study\x00'}],
|
|
'metadata':{'pages':'203-210','raw_title':'Original\u2029title'}}
|
|
self.evidence = self.root / 'evidence.json'
|
|
self.evidence.write_bytes(json_bytes({'url':'https://example.org/archive','comment':'Conference version'}))
|
|
self.path = self.root / 'records.jsonl'
|
|
self.raw = json_bytes(self.row)
|
|
self.path.write_bytes(self.raw)
|
|
self.manifest = {'source':'dblp','venue':'UAI','year':2001,'track':'main','count':1,
|
|
'list_url':'https://example.org/list','file_sha256':digest(self.raw),
|
|
'status':'partial','missing_abstract_count':1,'parser_version':2}
|
|
self.path.with_suffix('.manifest.json').write_bytes(json_bytes(self.manifest))
|
|
self.plan = {'format_version':1,'inputs':[self.input(self.path)],'assertions':[
|
|
{'source':'dblp','external_id':self.row['external_id'],'record_sha256':digest(self.raw),
|
|
'identifiers':{'arxiv':'1408.1485'},'evidence':[{
|
|
'file':str(self.evidence),'sha256':digest(self.evidence.read_bytes()),
|
|
'url':'https://example.org/archive','checked_at':'2026-11-08T08:00:00Z',
|
|
'note':'Reviewed title, authors, publication and version evidence'}]}]}
|
|
|
|
def input(self, path):
|
|
return {'file':str(path),'sha256':digest(path.read_bytes()),
|
|
'manifest_sha256':digest(path.with_suffix('.manifest.json').read_bytes())}
|
|
|
|
def run_plan(self, plan=None):
|
|
path = self.root / 'plan.json'
|
|
path.write_bytes(json_bytes(self.plan if plan is None else plan))
|
|
return prepare(path, self.output)
|
|
|
|
def test_only_identifiers_and_provenance_change_and_raw_inputs_survive(self):
|
|
result = self.run_plan()
|
|
self.assertEqual(self.path.read_bytes(), self.raw)
|
|
prepared = json.loads((self.output / self.path.name).read_bytes())
|
|
provenance = prepared['metadata'].pop('identifier_curation')
|
|
self.assertEqual(prepared.pop('identifiers'), {'arxiv':'1408.1485'})
|
|
self.assertEqual(prepared, self.row)
|
|
self.assertEqual(provenance[0]['record_sha256'], digest(self.raw))
|
|
self.assertEqual((self.output / provenance[0]['evidence'][0]['file']).read_bytes(), self.evidence.read_bytes())
|
|
manifest = json.loads((self.output / 'records.manifest.json').read_bytes())
|
|
for key, value in self.manifest.items():
|
|
if key != 'file_sha256': self.assertEqual(manifest[key], value)
|
|
self.assertEqual(manifest['file_sha256'], digest((self.output / self.path.name).read_bytes()))
|
|
self.assertEqual(result['assertions_applied'], 1)
|
|
self.assertEqual(result['identifiers_added'], 1)
|
|
self.assertEqual(result['verification_scope'], SCOPE)
|
|
self.assertFalse(result['semantic_identity_verified'])
|
|
self.assertFalse(result['database_written'])
|
|
self.assertEqual(list(self.output.glob('*.jsonl')), [self.output / self.path.name])
|
|
with self.assertRaisesRegex(ValueError, 'already exist'): self.run_plan()
|
|
|
|
def test_late_invalid_file_publishes_nothing(self):
|
|
second = self.root / 'second.jsonl'
|
|
second.write_bytes(json_bytes({**self.row,'external_id':'second'}))
|
|
manifest = {**self.manifest,'file_sha256':digest(second.read_bytes()),'count':2}
|
|
second.with_suffix('.manifest.json').write_bytes(json_bytes(manifest))
|
|
self.plan['inputs'].append(self.input(second))
|
|
with self.assertRaisesRegex(ValueError, 'count mismatch'): self.run_plan()
|
|
self.assertFalse(self.output.exists())
|
|
self.assertFalse(list(self.root.glob('.identifier-curation-*')))
|
|
self.assertEqual(self.path.read_bytes(), self.raw)
|
|
|
|
def test_unasserted_rows_keep_exact_physical_bytes_and_doi_is_additive(self):
|
|
untouched = (json.dumps({**self.row,'external_id':'unselected'},ensure_ascii=False,indent=None) + ' ').encode()
|
|
self.path.write_bytes(self.raw + untouched)
|
|
self.path.with_suffix('.manifest.json').write_bytes(json_bytes({**self.manifest,
|
|
'file_sha256':digest(self.path.read_bytes()),'count':2,'missing_abstract_count':2}))
|
|
self.plan['inputs'] = [self.input(self.path)]
|
|
self.plan['assertions'][0]['identifiers']['doi'] = '10.1234/example'
|
|
result = self.run_plan()
|
|
with (self.output / self.path.name).open('rb') as stream:
|
|
first = json.loads(next(stream))
|
|
self.assertEqual(next(stream), untouched)
|
|
self.assertEqual(first['identifiers'], {'arxiv':'1408.1485','doi':'10.1234/example'})
|
|
self.assertEqual(result['identifiers_added'], 2)
|
|
|
|
def test_missing_duplicate_and_mismatched_assertions_fail_closed(self):
|
|
for mode in ('missing','duplicate','line_hash','duplicate_input'):
|
|
with self.subTest(mode=mode):
|
|
plan = copy.deepcopy(self.plan)
|
|
assertion = plan['assertions'][0]
|
|
if mode == 'missing': assertion['external_id'] = 'absent'
|
|
elif mode == 'duplicate': plan['assertions'].append(copy.deepcopy(assertion))
|
|
elif mode == 'line_hash': assertion['record_sha256'] = '0' * 64
|
|
else:
|
|
other = self.root / 'duplicate.jsonl'
|
|
other.write_bytes(self.raw)
|
|
other.with_suffix('.manifest.json').write_bytes(json_bytes(self.manifest))
|
|
plan['inputs'].append(self.input(other))
|
|
with self.assertRaises(ValueError): self.run_plan(plan)
|
|
self.assertFalse(self.output.exists())
|
|
|
|
def test_identifier_conflict_does_not_overwrite_and_same_value_is_auditable(self):
|
|
self.row['identifiers'] = {'arxiv':'cs/0307069'}
|
|
self.path.write_bytes(json_bytes(self.row))
|
|
self.manifest['file_sha256'] = digest(self.path.read_bytes())
|
|
self.path.with_suffix('.manifest.json').write_bytes(json_bytes(self.manifest))
|
|
self.plan['inputs'] = [self.input(self.path)]
|
|
self.plan['assertions'][0]['record_sha256'] = digest(self.path.read_bytes())
|
|
with self.assertRaisesRegex(ValueError, 'identifier conflict'): self.run_plan()
|
|
self.assertFalse(self.output.exists())
|
|
self.plan['assertions'][0]['identifiers']['arxiv'] = 'cs/0307069'
|
|
self.assertEqual(self.run_plan()['identifiers_added'], 0)
|
|
|
|
def test_unsupported_changes_identifiers_and_unbound_evidence_are_rejected(self):
|
|
for mode in ('title','bad_month','version','unknown_scheme','doi_case','evidence_hash','credentials','timestamp'):
|
|
with self.subTest(mode=mode):
|
|
plan = copy.deepcopy(self.plan)
|
|
assertion = plan['assertions'][0]
|
|
evidence = assertion['evidence'][0]
|
|
if mode == 'title': assertion['title'] = 'Rewrite forbidden'
|
|
elif mode == 'bad_month': assertion['identifiers'] = {'arxiv':'1413.12345'}
|
|
elif mode == 'version': assertion['identifiers'] = {'arxiv':'1408.1485v2'}
|
|
elif mode == 'unknown_scheme': assertion['identifiers'] = {'url':'https://example.org'}
|
|
elif mode == 'doi_case': assertion['identifiers'] = {'doi':'10.1234/UPPER'}
|
|
elif mode == 'evidence_hash': evidence['sha256'] = '0' * 64
|
|
elif mode == 'credentials': evidence['url'] = 'https://user:secret@example.org'
|
|
else: evidence['checked_at'] = '2026-11-08T08:00:00'
|
|
with self.assertRaises(ValueError): self.run_plan(plan)
|
|
self.assertFalse(self.output.exists())
|
|
|
|
def test_input_and_manifest_hashes_and_duplicate_json_keys_are_checked(self):
|
|
for mode in ('input','manifest','duplicate_json','invalid_evidence_json'):
|
|
with self.subTest(mode=mode):
|
|
plan = copy.deepcopy(self.plan)
|
|
if mode == 'input':
|
|
plan['inputs'][0]['sha256'] = '0' * 64
|
|
elif mode == 'manifest':
|
|
plan['inputs'][0]['manifest_sha256'] = '0' * 64
|
|
elif mode == 'duplicate_json':
|
|
raw = self.raw[:-2] + b',"source":"dblp"}\n'
|
|
self.path.write_bytes(raw)
|
|
self.path.with_suffix('.manifest.json').write_bytes(json_bytes({**self.manifest,'file_sha256':digest(raw)}))
|
|
plan['inputs'] = [self.input(self.path)]
|
|
else:
|
|
self.evidence.write_bytes(b'not json')
|
|
plan['assertions'][0]['evidence'][0]['sha256'] = digest(self.evidence.read_bytes())
|
|
with self.assertRaises(ValueError): self.run_plan(plan)
|
|
self.assertFalse(self.output.exists())
|
|
self.path.write_bytes(self.raw)
|
|
self.path.with_suffix('.manifest.json').write_bytes(json_bytes(self.manifest))
|
|
|
|
def test_curated_anchor_import_does_not_merge_extended_version_or_its_history(self):
|
|
db = self.root / 'papers.sqlite'
|
|
store = Store(db)
|
|
try:
|
|
original, _ = store.record('arxiv','1408.1485','https://example.org/one',self.row['title'],None,self.row['authors'],{'arxiv':'1408.1485'})
|
|
extended, record = store.record('arxiv','cs/0307069','https://example.org/two',self.row['title'],None,self.row['authors'],{'arxiv':'cs/0307069'})
|
|
store.decision(extended,record,'ICLR',2025,'main','another-attempt','https://example.org/reject','rejected','Reject')
|
|
store.conn.commit()
|
|
finally:
|
|
store.conn.close()
|
|
self.run_plan()
|
|
ingest_file(db,self.output / self.path.name)
|
|
store = Store(db)
|
|
try:
|
|
c = store.conn
|
|
self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0], 2)
|
|
self.assertEqual(c.execute('SELECT paper_id FROM source_records WHERE external_id=?',(self.row['external_id'],)).fetchone()[0], original)
|
|
self.assertEqual(c.execute('SELECT s.paper_id FROM submissions s JOIN editions e ON e.id=s.edition_id JOIN venues v ON v.id=e.venue_id WHERE v.acronym=?',('ICLR',)).fetchone()[0], extended)
|
|
self.assertEqual(c.execute('SELECT count(*) FROM paper_merges').fetchone()[0], 0)
|
|
self.assertEqual(c.execute('PRAGMA foreign_key_check').fetchall(), [])
|
|
finally:
|
|
store.conn.close()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|