import hashlib import json from contextlib import closing from pathlib import Path import sqlite3 import tempfile from types import SimpleNamespace import unittest from unittest.mock import patch from paper_atlas.collectors.dblp import XML_SHA from paper_atlas.collectors.dblp_records import collect,identifiers,normalize_row,material_reason from paper_atlas.collectors.dblp_scope import scope from paper_atlas.db.store import Store from paper_atlas.ingestion.records import ingest_file class DblpRecordTests(unittest.TestCase): def setUp(self): # Tiny fixtures may live on a small tmpfs; production reserve stays 8 GiB. budget=patch('paper_atlas.collectors.dblp_records.shutil.disk_usage',return_value=SimpleNamespace(free=16*1024**3)) budget.start();self.addCleanup(budget.stop) def row(self,family='aaai',book='AAAI',year=2024,key='one'): return {'record_type':'inproceedings','key':f'conf/{family}/{key}','publtype':None, 'fields':{'title':['A research publication'],'author':['Jane Doe 0001','John Smith'], 'booktitle':[book],'year':[str(year)],'crossref':[f'conf/{family}/{year}'],'ee':['https://doi.org/10.1234/'+key]}} def volume(self,family,year,title=None): return {'record_type':'proceedings','key':f'conf/{family}/{year}','fields':{'title':[title or f'{family.upper()} {year}'],'year':[str(year)]}} def test_real_directory_aliases_do_not_confuse_other_conferences(self): row=self.row('clear2','CLeaR');volume=self.volume('clear2',2024,'Causal Learning and Reasoning, 2024') self.assertEqual(scope(row,volume)[0]['venue'],'CLeaR') self.assertIsNone(scope(self.row('clear','CLEAR'),self.volume('clear',2024))[0]) self.assertIsNone(scope(self.row('aamas','Adaptive Agents and Multi-Agent Systems'),self.volume('aamas',2024))[0]) self.assertEqual(scope(self.row('atal','AAMAS'),self.volume('atal',2024))[0]['venue'],'AAMAS') self.assertEqual(scope(self.row('aips','ICAPS'),self.volume('aips',2024))[0]['venue'],'ICAPS') def test_conference_year_is_distinct_from_publication_year(self): row=self.row('nips','NIPS',1988);row['fields']['crossref']=['conf/nips/1987'] volume=self.volume('nips',1987,'NIPS Conference, Denver, 1987; published 1988') result,_=scope(row,volume);self.assertEqual(result['year'],1987);self.assertEqual(result['bibliographic_year'],1988) volume['fields']['title']=['NIPS collected papers'] self.assertEqual(scope(row,volume)[1],'unverified_year_difference') def test_joint_events_and_tracks_are_explicit(self): result,_=scope(self.row('ijcai','IJCAI/ECAI',2022),self.volume('ijcai',2022)) self.assertEqual(result['venue'],'ECAI+IJCAI');self.assertEqual(result['venue_aliases'],['ECAI','IJCAI']) result,_=scope(self.row('coling','LREC/COLING'),self.volume('coling',2024)) self.assertEqual(result['venue'],'COLING+LREC') for book,track in [('ACL (Findings)','findings'),('ACL (Student Research Workshop)','workshop'),('ACL (System Demonstrations)','demo')]: self.assertEqual(scope(self.row('acl',book),self.volume('acl',2024))[0]['track'],track) for suffix,track in [('Special interest tracks and posters','special_interest'),('Alternate Track Papers & Posters','alternate')]: self.assertEqual(scope(self.row('www','WWW ('+suffix+')'),self.volume('www',2024))[0]['track'],track) volume=self.volume('pkdd',2026,'ECML PKDD 2026, Applied Data Science Track, Demo Track and Industrial Track') self.assertEqual(scope(self.row('pkdd','ECML PKDD (10)',2026),volume)[0]['track'],'proceedings_unspecified') def test_short_year_keys_require_a_unique_explicit_year_not_a_guessed_century(self): row=self.row('sigir','SIGIR',1980);volume=self.volume('sigir',1980,'Information Retrieval Research, Cambridge, June 1980') volume['key']='conf/sigir/80';volume['fields']['year']=['1981'];row['fields']['crossref']=[volume['key']] edition,_=scope(row,volume);self.assertEqual(edition['year'],1980) self.assertIn('two-digit',edition['edition_year_basis']) volume['fields']['title']=['SIGIR 1980, published 1981'] self.assertEqual(scope(row,volume)[1],'unverified_edition_year') volume['fields']['title']=['SIGIR 1989'] self.assertEqual(scope(row,volume)[1],'unverified_edition_year') def test_crossref_mismatch_and_future_editions_are_held_for_review(self): self.assertEqual(scope(self.row(),self.volume('kdd',2024))[1],'crossref_venue_conflict') self.assertEqual(scope(self.row(year=2027),self.volume('aaai',2027))[1],'future_edition') def test_only_own_identifier_links_are_used_and_ambiguity_is_preserved(self): own,ambiguous=identifiers({'ee':['https://doi.org/10.1234/published','https://doi.org/10.48550/arXiv.2401.00001','https://openreview.net/forum?id=realForum'], 'cite':['https://doi.org/10.9999/reference']}) self.assertEqual(own,{'doi':'10.1234/published','arxiv':'2401.00001','openreview':'realForum'});self.assertFalse(ambiguous) own,ambiguous=identifiers({'ee':['https://doi.org/10.1234/a','https://doi.org/10.1234/b','https://openreview.net/forum?id=2026-Oral--99-placeholder']}) self.assertEqual(own,{});self.assertEqual(len(ambiguous['doi']),2) def test_frontmatter_and_anonymous_event_headers_are_not_paper_acceptances(self): for title in ('Title, Preface.','Title, Conference Committee, Table of Contents, Author Index.','Front Matter.','Backmatter.','Foreword.'): self.assertEqual(material_reason(title,['An Editor'],'conf/sigir/C78',1978,'main'),'frontmatter') for title in ('Sponsors.','IJCAI-09 Awards and Distinguished Papers.','2013 Conference Organization.','Area Chairs and Program Committee.','Welcome to ICDM 2005.','Workshop Summaries.'): self.assertIsNotNone(material_reason(title,[],'conf/ijcai/X09',2009,'main')) self.assertEqual(material_reason("Wiki Workshop Chairs' Welcome.",['An Organizer'],'conf/www/X19t',2019,'companion'),'frontmatter') self.assertEqual(material_reason('Attention - from Neuroscience to the Web and Wellbeing.',[],'conf/www/X19b',2019,'companion'),'unverified_proceedings_material') row=self.row();row['fields']['title']=['Sponsors.'];row['fields']['author']=[] self.assertEqual(normalize_row(row,{'conf/aaai/2024':self.volume('aaai',2024)},{'AAAI'})[1],'proceedings_heading') def test_missing_authors_alone_do_not_remove_real_research_records(self): self.assertIsNone(material_reason('The Onomastica interlanguage pronunciation lexicon.',[],'conf/interspeech/X95',1995,'main')) self.assertIsNone(material_reason('Learning Models of Relational Stochastic Processes.',[],'conf/ecml/SanghaiDW05',2005,'main')) self.assertIsNone(material_reason('Sponsors.', ['A Researcher'],'conf/aaai/research',2024,'main')) def fixture(self,root,rows): catalogue=root/'dblp-catalogue.sqlite';volume=self.volume('aaai',2024) with closing(sqlite3.connect(catalogue)) as c, c: c.execute('CREATE TABLE proceedings(key TEXT PRIMARY KEY,metadata_json TEXT)') c.execute('INSERT INTO proceedings VALUES(?,?)',(volume['key'],json.dumps(volume))) candidates=root/'candidates.jsonl';candidates.write_text(''.join(json.dumps(row)+'\n' for row in rows)) manifest={'snapshot_sha256':XML_SHA,'candidate_sha256':hashlib.sha256(candidates.read_bytes()).hexdigest(), 'candidate_records':len(rows),'record_types':{'proceedings':1,'inproceedings':len(rows)}} catalogue.with_name('dblp-ai-candidates.manifest.json').write_text(json.dumps(manifest)) return catalogue,candidates def test_normalized_files_import_with_unknown_abstracts_and_no_invented_retracted_acceptance(self): with tempfile.TemporaryDirectory() as tmp: root=Path(tmp);withdrawn=self.row(key='two');withdrawn['publtype']='withdrawn' catalogue,candidates=self.fixture(root,[self.row(),withdrawn,self.row(book='AAAI Workshops',key='workshop')]) result=collect(catalogue,root/'out',['AAAI'],candidates=candidates) self.assertEqual(result['normalized_records'],2);self.assertEqual(result['excluded_by_reason']['separate_workshop'],1) manifest=result['files'][0];self.assertEqual(manifest['missing_abstract_count'],2) rows=[json.loads(line) for line in Path(manifest['file']).read_text().splitlines()] self.assertEqual(rows[0]['authors'],['Jane Doe','John Smith']);self.assertIsNone(rows[0]['abstract']) self.assertEqual(rows[1]['status'],'unknown');self.assertEqual(rows[1]['decision_events'],[]) db=root/'papers.sqlite';ingest_file(db,manifest['file']) with closing(sqlite3.connect(db)) as c:self.assertEqual(c.execute('SELECT count(*) FROM decisions').fetchone()[0],1) def test_checksum_failure_and_duplicate_keys_do_not_publish_files(self): with tempfile.TemporaryDirectory() as tmp: root=Path(tmp);catalogue,candidates=self.fixture(root,[self.row(),self.row()]) with self.assertRaisesRegex(ValueError,'Repeated DBLP'):collect(catalogue,root/'out',['AAAI'],candidates=candidates) self.assertFalse(list((root/'out').glob('*.jsonl'))) candidates.write_text('tampered') with self.assertRaisesRegex(ValueError,'checksum mismatch'):collect(catalogue,root/'out',['AAAI'],candidates=candidates) def test_case_variants_use_one_conference_edition_and_submission(self): with tempfile.TemporaryDirectory() as tmp: store=Store(Path(tmp)/'papers.sqlite');self.addCleanup(store.conn.close) paper,one=store.record('pmlr','one','https://example.org/one','A paper','Abstract',[],{'doi':'10.1234/one'}) store.decision(paper,one,'CORL',2024,'main','one','https://example.org/one','accepted','Accept',identities=[('publication_doi','10.1234/one')]) same,two=store.record('dblp','two','https://example.org/two','A paper',None,[],{'doi':'10.1234/one'}) store.decision(same,two,'CoRL',2024,'main','two','https://example.org/two','accepted','Published',identities=[('publication_doi','10.1234/one')]) self.assertEqual(store.conn.execute('SELECT count(*) FROM venues').fetchone()[0],1) self.assertEqual(store.conn.execute('SELECT count(*) FROM submissions').fetchone()[0],1) def test_production_disk_reserve_stops_before_publishing(self): with tempfile.TemporaryDirectory() as tmp: root=Path(tmp);catalogue,candidates=self.fixture(root,[self.row()]) with patch('paper_atlas.collectors.dblp_records.shutil.disk_usage',return_value=SimpleNamespace(free=1024)): with self.assertRaisesRegex(RuntimeError,'8 GiB'):collect(catalogue,root/'out',['AAAI'],candidates=candidates) self.assertFalse(list((root/'out').glob('*.jsonl'))) if __name__=='__main__':unittest.main()