import json import sqlite3 import tempfile import unittest from pathlib import Path from paper_atlas.db.store import Store from paper_atlas.ingestion.records import ingest_file class StoreTests(unittest.TestCase): def test_partial_source_refresh_preserves_known_abstract_and_can_later_restore_it(self): paper,record=self.store.record('cvf','paper','https://example.org/paper','Known title','Known abstract',['Jane Doe']) self.store.record('cvf','paper','https://example.org/paper','Known title',None,[],metadata={'abstract_fetch_error':{'status':404}}) self.store.record('cvf','paper','https://example.org/paper','Known title',None,[],metadata={'abstract_fetch_error':{'status':404}}) c=self.store.conn self.assertEqual(c.execute('SELECT abstract,authors_json FROM papers').fetchone(),('Known abstract','["Jane Doe"]')) source=c.execute('SELECT abstract,metadata_json FROM source_records WHERE id=?',(record,)).fetchone() self.assertIsNone(source[0]);self.assertEqual(json.loads(source[1])['last_known_abstract']['abstract'],'Known abstract') self.store.record('cvf','paper','https://example.org/paper','Updated title','Restored revised abstract',['Jane Doe']) self.assertEqual(c.execute('SELECT title,abstract FROM papers').fetchone(),('Updated title','Restored revised abstract')) def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.db = Path(self.tmp.name)/'papers.sqlite' self.store = Store(self.db) self.addCleanup(self.store.conn.close) def record(self,source,identifier,ids=None,authors=None): return self.store.record(source,identifier,'https://example.org/'+identifier,'A sufficiently long example research paper title','Example abstract',authors or ['Jane Doe','John Smith'],ids) def test_bridge_identifiers_preserves_submission_and_tags(self): first,record = self.record('arxiv','2601.00001',{'arxiv':'2601.00001'},['First Author']) second,other_record = self.record('pmlr','v1/other',{'doi':'10.0000/example'},['Other Author']) self.store.accepted(second,other_record,'ICML',2025,'main','v1/other','https://example.org/other') tag = self.store.conn.execute("INSERT INTO tags(namespace,value) VALUES('custom:topic','test')").lastrowid self.store.conn.execute('INSERT INTO paper_tags VALUES(?,?,?,?)',(second,tag,other_record,'manual')) resolved,_ = self.record('openreview','forum1',{'arxiv':'2601.00001','doi':'10.0000/example'},['Anonymous']) self.assertEqual(resolved,first) c=self.store.conn self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0],1) self.assertEqual(c.execute('SELECT paper_id FROM submissions').fetchone()[0],first) self.assertEqual(c.execute('SELECT paper_id FROM paper_tags').fetchone()[0],first) self.assertEqual(c.execute('PRAGMA foreign_key_check').fetchall(),[]) self.assertEqual(c.execute('SELECT count(*) FROM paper_merges').fetchone()[0],1) def test_cross_platform_exact_title_needs_author_corroboration(self): paper,_=self.record('arxiv','2601.00001') same,_=self.record('pmlr','v1/paper') other,_=self.record('acl','2025.acl.1',authors=['Unrelated Person']) self.assertEqual(same,paper) self.assertNotEqual(other,paper) self.assertEqual(self.store.conn.execute('SELECT count(*) FROM match_candidates').fetchone()[0],1) def test_identifier_anchor_does_not_absorb_a_same_title_arxiv_version(self): conference,first=self.record('arxiv','1408.1485') extended,second=self.record('arxiv','cs/0307069') self.store.decision(extended,second,'ICLR',2025,'main','separate','https://example.org/reject','rejected','Reject') resolved,_=self.record('dblp','conf/uai/example',{'arxiv':'1408.1485'}) self.assertEqual(resolved,conference);self.assertNotEqual(conference,extended) c=self.store.conn self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0],2) self.assertEqual(c.execute('SELECT paper_id FROM submissions').fetchone()[0],extended) self.assertEqual(c.execute('SELECT count(*) FROM paper_merges').fetchone()[0],0) self.assertEqual(c.execute('SELECT count(*) FROM match_candidates').fetchone()[0],1) # A refresh of the anchored original source is equally unambiguous. self.assertEqual(self.record('arxiv','1408.1485')[0],conference) self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0],2) def test_ambiguous_title_aborts_whole_file_without_merging_existing_entities(self): self.record('arxiv','1408.1485');self.record('arxiv','cs/0307069') self.store.conn.commit() rows=[{'source':'dblp','external_id':'conf/uai/new','url':'https://example.org/new', 'title':'An unrelated earlier record in this file','abstract':None,'authors':[], 'venue':'UAI','year':2001,'track':'main','status':'accepted'}, {'source':'dblp','external_id':'conf/uai/ambiguous','url':'https://example.org/ambiguous', 'title':'A sufficiently long example research paper title','abstract':None, 'authors':['Jane Doe','John Smith'],'identifiers':{'doi':'10.0000/not-yet-known'}, 'venue':'UAI','year':2001,'track':'main','status':'accepted'}] path=Path(self.tmp.name)/'ambiguous.jsonl' path.write_text(''.join(json.dumps(row)+'\n' for row in rows)) path.with_suffix('.manifest.json').write_text(json.dumps({'source':'dblp','list_url':'https://example.org/list','count':2})) with self.assertRaisesRegex(ValueError,'Ambiguous paper identity'):ingest_file(self.db,path) c=self.store.conn for table,count in [('papers',2),('source_records',2),('submissions',0),('decisions',0),('paper_merges',0)]: self.assertEqual(c.execute('SELECT count(*) FROM '+table).fetchone()[0],count,table) self.assertEqual(c.execute('SELECT status FROM ingest_runs').fetchone()[0],'failed') def test_explicit_bridge_merges_only_its_identifier_matches(self): first,_=self.record('arxiv','2601.00001',authors=['First Author']) second,_=self.record('arxiv','2601.00002',{'doi':'10.0000/bridge'},authors=['Second Author']) unrelated,_=self.record('arxiv','2601.00003') resolved,_=self.record('official','bridge',{'arxiv':'2601.00001','doi':'10.0000/bridge'}) self.assertEqual(resolved,min(first,second)) self.assertEqual(self.store.conn.execute('SELECT count(*) FROM papers').fetchone()[0],2) self.assertIsNotNone(self.store.conn.execute('SELECT 1 FROM papers WHERE id=?',(unrelated,)).fetchone()) self.assertEqual(self.store.conn.execute('SELECT count(*) FROM paper_merges').fetchone()[0],1) def test_manifest_count_mismatch_rolls_back_records(self): path=Path(self.tmp.name)/'source.jsonl' path.write_text(json.dumps({'source':'pmlr','external_id':'v1/paper','url':'https://example.org/paper','title':'Paper','abstract':'Abstract','authors':[],'venue':'ICML','year':2023,'track':'main','status':'accepted'})+'\n') path.with_suffix('.manifest.json').write_text(json.dumps({'source':'pmlr','list_url':'https://example.org/list','count':2})) with self.assertRaisesRegex(ValueError,'differs'): ingest_file(self.db,path) self.assertEqual(self.store.conn.execute('SELECT count(*) FROM source_records').fetchone()[0],0) self.assertEqual(self.store.conn.execute('SELECT status FROM ingest_runs').fetchone()[0],'failed') def test_source_update_refreshes_canonical_text_and_fts(self): paper,_=self.record('arxiv','2601.00001') updated,_=self.store.record('arxiv','2601.00001','https://arxiv.org/abs/2601.00001','Updated research title','New searchable content',['Jane Doe']) self.assertEqual(updated,paper) self.assertEqual(self.store.conn.execute('SELECT title FROM papers').fetchone()[0],'Updated research title') self.assertEqual(self.store.conn.execute('SELECT url FROM source_records').fetchone()[0],'https://arxiv.org/abs/2601.00001') self.assertEqual(self.store.conn.execute("SELECT count(*) FROM papers_fts WHERE papers_fts MATCH 'searchable'").fetchone()[0],1) def test_tag_refresh_keeps_other_record_evidence_and_custom_tags(self): paper,first=self.record('iclr_program','2025/a',{'openreview':'sharedForum'}) same,second=self.record('iclr_program','2025/b',{'openreview':'sharedForum'}) self.store.record_tags(paper,first,'iclr_program',[{'namespace':'iclr_program:topic','value':'Shared'}]) self.store.record_tags(same,second,'iclr_program',[{'namespace':'iclr_program:topic','value':'Shared'}]) tag=self.store.conn.execute("INSERT INTO tags(namespace,value) VALUES('custom:topic','Manual')").lastrowid self.store.conn.execute('INSERT INTO paper_tags VALUES(?,?,?,?)',(paper,tag,None,'manual')) self.store.record_tags(paper,first,'iclr_program',[]) self.assertEqual(self.store.conn.execute('SELECT source_record_id FROM paper_tags ORDER BY source_record_id').fetchall(),[(None,),(second,)]) with self.assertRaisesRegex(ValueError,'namespace'): self.store.record_tags(paper,second,'iclr_program',[{'namespace':'custom:topic','value':'No overwrite'}]) self.assertEqual(self.store.conn.execute('SELECT count(*) FROM paper_tags').fetchone()[0],2) def test_missing_abstract_does_not_become_complete_coverage(self): path=Path(self.tmp.name)/'partial.jsonl' path.write_text(json.dumps({'source':'colm','external_id':'paper1','url':'https://example.org/paper1','title':'Paper','abstract':None,'authors':[],'venue':'COLM','year':2024,'track':'main','status':'accepted'})+'\n') path.with_suffix('.manifest.json').write_text(json.dumps({'source':'colm','parser_version':2,'list_url':'https://example.org/list','count':1,'status':'partial','missing_abstract_count':1})) ingest_file(self.db,path) self.assertEqual(self.store.conn.execute('SELECT status FROM edition_sources').fetchone()[0],'partial') def test_legacy_proceedings_manifest_is_rejected_before_import(self): path=Path(self.tmp.name)/'old.jsonl' path.write_text('') path.with_suffix('.manifest.json').write_text(json.dumps({'source':'colm','count':1})) with self.assertRaisesRegex(ValueError,'parser_version'): ingest_file(self.db,path) self.assertEqual(self.store.conn.execute('SELECT count(*) FROM source_records').fetchone()[0],0) def test_missing_abstract_is_checked_even_when_manifest_claims_complete(self): path=Path(self.tmp.name)/'incomplete.jsonl' row={'source':'pmlr','external_id':'v1/paper','url':'https://example.org/paper','title':'Paper','abstract':None,'authors':[],'venue':'ICML','year':2023,'track':'main','status':'accepted'} path.write_text(json.dumps(row)+'\n') path.with_suffix('.manifest.json').write_text(json.dumps({'source':'pmlr','list_url':'https://example.org/list','count':1,'status':'collected','missing_abstract_count':0})) ingest_file(self.db,path) self.assertEqual(self.store.conn.execute('SELECT status FROM edition_sources').fetchone()[0],'partial') def test_manifest_edition_mismatch_rolls_back_instead_of_marking_coverage(self): path=Path(self.tmp.name)/'wrong-edition.jsonl' row={'source':'pmlr','external_id':'v1/paper','url':'https://example.org/paper','title':'Paper','abstract':'Abstract','authors':[],'venue':'ICML','year':2023,'track':'main','status':'accepted'} path.write_text(json.dumps(row)+'\n') path.with_suffix('.manifest.json').write_text(json.dumps({'source':'pmlr','list_url':'https://example.org/list','count':1,'year':2024})) with self.assertRaisesRegex(ValueError,'edition differs'): ingest_file(self.db,path) self.assertEqual(self.store.conn.execute('SELECT count(*) FROM source_records').fetchone()[0],0) self.assertEqual(self.store.conn.execute('SELECT count(*) FROM edition_sources').fetchone()[0],0) if __name__=='__main__': unittest.main()