import json import subprocess import sys import tempfile import unittest from pathlib import Path from paper_atlas.db.store import Store class QueryTests(unittest.TestCase): def run_query(self,db,*args): return json.loads(subprocess.run([sys.executable,'-B','-m','paper_atlas','query','--db',str(db),*args],check=True,capture_output=True,text=True).stdout) def test_rejection_then_acceptance_at_another_venue_preserves_both_attempts(self): with tempfile.TemporaryDirectory() as tmp: db=Path(tmp)/'papers.sqlite';store=Store(db);self.addCleanup(store.conn.close) paper,first=store.record('openreview_snapshot','forum1','https://example.org/1','Original title','Abstract',['Jane Doe'],{'arxiv':'2401.00001','openreview':'forum1'}) store.decision(paper,first,'ICLR',2024,'main','forum1','https://example.org/reject','rejected','Reject',identities=[('openreview','forum1')],observed_at='2024-02-01T00:00:00Z') same,second=store.record('colm','forum2','https://example.org/2','Revised title','Revised abstract',['Jane Doe'],{'arxiv':'2401.00001','openreview':'forum2'}) store.decision(same,second,'COLM',2025,'main','forum2','https://example.org/accept','accepted','Accept',identities=[('openreview','forum2')],observed_at='2025-08-01T00:00:00Z') store.conn.commit();self.assertEqual(same,paper) rows=self.run_query(db,'--venue','ICLR','--status','rejected','--outcome','accepted') self.assertEqual([row['id'] for row in rows],[paper]) self.assertEqual(rows[0]['publication_outcome'],'accepted') self.assertEqual([a['venue'] for a in rows[0]['known_acceptances']],['COLM']) detail=self.run_query(db,'--paper',str(paper))[0] self.assertEqual([(s['acronym'],s['latest_status']) for s in detail['submissions']],[('ICLR','rejected'),('COLM','accepted')]) self.assertEqual(self.run_query(db,'--venue','ICLR','--status','rejected','--outcome','no_known_acceptance'),[]) self.assertEqual(self.run_query(db,'--stats')['papers_with_known_acceptance'],1) def test_later_rejection_elsewhere_does_not_erase_acceptance_and_unknown_is_not_never(self): with tempfile.TemporaryDirectory() as tmp: db=Path(tmp)/'papers.sqlite';store=Store(db);self.addCleanup(store.conn.close) paper,one=store.record('source','one','https://example.org/1','One paper','Abstract',[]) store.decision(paper,one,'COLM',2024,'main','one','https://example.org/accept','accepted','Accept',observed_at='2024-01-01T00:00:00Z') store.decision(paper,one,'ICLR',2025,'main','two','https://example.org/reject','rejected','Reject',observed_at='2025-01-01T00:00:00Z') unknown,_=store.record('arxiv','other','https://example.org/other','No observed submissions','Abstract',[]) store.conn.commit() self.assertEqual([r['id'] for r in self.run_query(db,'--outcome','accepted')],[paper]) self.assertEqual([r['id'] for r in self.run_query(db,'--outcome','no_known_acceptance')],[unknown]) store.decision(paper,one,'COLM',2024,'main','one','https://example.org/withdraw','withdrawn','Withdrawn',observed_at='2026-01-01T00:00:00Z') store.conn.commit();self.assertEqual(self.run_query(db,'--outcome','accepted'),[]) history=self.run_query(db,'--paper',str(paper))[0] self.assertEqual([d['status'] for d in history['submissions'][0]['decisions']],['accepted','withdrawn']) def test_joint_conference_alias_does_not_duplicate_submission(self): with tempfile.TemporaryDirectory() as tmp: db=Path(tmp)/'papers.sqlite';store=Store(db);self.addCleanup(store.conn.close) paper,record=store.record('acl_anthology','paper1','https://example.org/1','Joint event paper','Abstract',[]) store.decision(paper,record,'COLING+LREC',2024,'main','paper1','https://example.org/1','accepted','official XML',['COLING','LREC']) store.conn.commit() for venue in ('COLING','LREC'): command=[sys.executable,'-B','-m','paper_atlas','query','--db',str(db),'--venue',venue,'--track','main'] rows=json.loads(subprocess.run(command,check=True,capture_output=True,text=True).stdout) self.assertEqual([r['id'] for r in rows],[paper]) self.assertEqual(store.conn.execute('SELECT count(*) FROM submissions').fetchone()[0],1) def test_venue_filter_is_case_insensitive_and_uses_latest_decision(self): with tempfile.TemporaryDirectory() as tmp: db=Path(tmp)/'papers.sqlite' store=Store(db) self.addCleanup(store.conn.close) paper,record=store.record('peerread','paper1','https://example.org/paper1','A real submission','Abstract',['Jane Doe']) store.decision(paper,record,'ICLR',2017,'main','paper1','https://example.org/paper1','rejected','explicit rejection') store.record('arxiv','2601.00001','https://example.org/other','Another paper','Abstract',['John Smith']) store.conn.commit() command=[sys.executable,'-B','-m','paper_atlas','query','--db',str(db),'--venue','iclr','--year','2017','--status','rejected'] result=json.loads(subprocess.run(command,check=True,capture_output=True,text=True).stdout) self.assertEqual([r['id'] for r in result],[paper]) details_command=[sys.executable,'-B','-m','paper_atlas','query','--db',str(db),'--paper',str(paper)] details=json.loads(subprocess.run(details_command,check=True,capture_output=True,text=True).stdout)[0] self.assertEqual(details['authors'],['Jane Doe']) self.assertEqual(details['submissions'][0]['decisions'][0]['status'],'rejected') self.assertEqual(details['sources'][0]['external_id'],'paper1') stats_command=[sys.executable,'-B','-m','paper_atlas','query','--db',str(db),'--stats'] stats=json.loads(subprocess.run(stats_command,check=True,capture_output=True,text=True).stdout) self.assertEqual(stats['papers'],2) self.assertEqual({r['name']:r['imported_records'] for r in stats['records_by_source']},{'arxiv':1,'peerread':1}) store.decision(paper,record,'ICLR',2017,'main','paper1','https://example.org/paper1','accepted','changed decision') store.conn.commit() self.assertEqual(json.loads(subprocess.run(command,check=True,capture_output=True,text=True).stdout),[]) if __name__=='__main__': unittest.main()