60 lines
3.8 KiB
Python
60 lines
3.8 KiB
Python
"""RSS publisher/DBLP observations share one event without losing old decisions."""
|
|
from pathlib import Path
|
|
import json
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
from paper_atlas.db.submissions import identities_for_row
|
|
from tests.test_aamas_publication_identity import row, ingest
|
|
|
|
DOI='10.15607/rss.2026.xxii.041'
|
|
|
|
def rss(source,external,**kwargs):
|
|
return row(source,external,venue='RSS',year=2026,track='main',
|
|
identifiers=kwargs.pop('identifiers',{'doi':DOI}),**kwargs)
|
|
|
|
class RssPublicationIdentityTests(unittest.TestCase):
|
|
def test_only_verified_accepted_source_adds_publication_identity(self):
|
|
r=rss('rss_proceedings','2026/p041',identifiers={'doi':'https://doi.org/'+DOI.upper()})
|
|
self.assertEqual(identities_for_row(r),[('publication_doi',DOI)])
|
|
for status in ('rejected','desk_rejected','withdrawn','under_review','unknown'):
|
|
self.assertEqual(identities_for_row({**r,'status':status}),[])
|
|
self.assertEqual(identities_for_row({**r,'source':'unverified_rss_copy'}),[])
|
|
self.assertEqual(identities_for_row({**r,'identifiers':{'doi':'not-a-doi'}}),[])
|
|
|
|
def test_publisher_and_bibliography_both_orders_preserve_two_observations(self):
|
|
a=rss('rss_proceedings','2026/p041');b=rss('dblp','conf/rss/Example26')
|
|
for pair in ((a,b),(b,a)):
|
|
with self.subTest(first=pair[0]['source']),tempfile.TemporaryDirectory() as tmp:
|
|
root=Path(tmp);db=root/'papers.sqlite'
|
|
for r in pair:ingest(db,root,r)
|
|
ingest(db,root,a)
|
|
with sqlite3.connect(db) as c:
|
|
self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0],1)
|
|
self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0],1)
|
|
self.assertEqual(c.execute('SELECT count(*) FROM source_records').fetchone()[0],2)
|
|
self.assertEqual(c.execute('SELECT count(*) FROM submission_sources').fetchone()[0],2)
|
|
self.assertEqual(c.execute('SELECT count(*) FROM decisions').fetchone()[0],2)
|
|
self.assertEqual(c.execute('SELECT count(*) FROM decision_sources').fetchone()[0],2)
|
|
self.assertEqual(c.execute('SELECT scheme,value FROM submission_identifiers').fetchall(),[('publication_doi',DOI)])
|
|
metadata=[json.loads(r[0]) for r in c.execute('SELECT metadata_json FROM source_records')]
|
|
self.assertEqual({x['original_source'] for x in metadata},{'rss_proceedings','dblp'})
|
|
self.assertEqual(c.execute('PRAGMA foreign_key_check').fetchall(),[])
|
|
|
|
def test_prior_rejection_remains_independent_and_raw_evidence_unchanged(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root=Path(tmp);db=root/'papers.sqlite'
|
|
old=row('openreview_snapshot','old-forum',venue='ICLR',year=2025,track='main',
|
|
status='rejected',identifiers={'arxiv':'2501.00001','openreview':'old-forum'})
|
|
ingest(db,root,old)
|
|
with sqlite3.connect(db) as c:
|
|
before=c.execute('SELECT * FROM decisions').fetchall()
|
|
new=rss('rss_proceedings','2026/p041',identifiers={'arxiv':'2501.00001','doi':DOI})
|
|
ingest(db,root,new)
|
|
with sqlite3.connect(db) as c:
|
|
self.assertEqual(c.execute('SELECT * FROM decisions WHERE status=\'rejected\'').fetchall(),before)
|
|
self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0],2)
|
|
self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0],1)
|
|
self.assertEqual(c.execute('SELECT count(*) FROM decisions WHERE status=\'accepted\'').fetchone()[0],1)
|
|
self.assertEqual(c.execute('PRAGMA foreign_key_check').fetchall(),[])
|