87 lines
6.5 KiB
Python
87 lines
6.5 KiB
Python
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()
|