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()