203 lines
13 KiB
Python
203 lines
13 KiB
Python
from contextlib import closing, redirect_stdout, contextmanager
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
import uuid
|
|
from importlib.resources import files
|
|
|
|
from paper_atlas.db.store import Store
|
|
from paper_atlas.db.backend import open_store
|
|
from paper_atlas.db.events import has_event_key
|
|
from paper_atlas.collectors.dblp import XML_SHA
|
|
from paper_atlas.ingestion.event_identity import plan, upgrade, exact_counts
|
|
from paper_atlas.ingestion.records import ingest_file
|
|
from paper_atlas.ingestion.backfill import backfill
|
|
from paper_atlas.postgres.client import connect
|
|
from paper_atlas.postgres.schema import quote
|
|
from paper_atlas.postgres.migration import migrate
|
|
from paper_atlas import querying
|
|
|
|
|
|
@contextmanager
|
|
def opened_store(**kwargs):
|
|
store=open_store(**kwargs)
|
|
try:
|
|
yield store
|
|
finally:
|
|
store.conn.close()
|
|
|
|
|
|
def legacy(db):
|
|
sql = files('paper_atlas.db').joinpath('schema.sql').read_text().replace(" event_key TEXT NOT NULL DEFAULT '',\n", '').replace('UNIQUE(venue_id,year,track,event_key)', 'UNIQUE(venue_id,year,track)')
|
|
c = sqlite3.connect(db); c.executescript(sql); c.close()
|
|
s = Store(db)
|
|
p,r = s.record('dblp','conf/www/Fall95','https://example.org/fall','Same paper in two events',
|
|
'Abstract\x00with\x01text',['A Person'],{'arxiv':'2501.00001'})
|
|
s.record_tags(p,r,'dblp',[{'namespace':'dblp:booktitle','value':'WWW (Fall)'}])
|
|
s.decision(p,r,'Legacy',1995,'main','conf/www/Fall95','https://example.org/accept','accepted','Accept')
|
|
s.conn.execute("UPDATE venues SET acronym='WWW',name='WWW'")
|
|
meta={'snapshot_sha256':XML_SHA,'proceedings_key':'conf/www/1995f','bibliographic_year':1995,
|
|
'bibliography':{'booktitle':['WWW (Fall)'],'url':['db/conf/www/www1995f.html#Fall95']}}
|
|
s.conn.execute('UPDATE source_records SET metadata_json=?',(json.dumps(meta),));s.conn.commit();s.conn.close()
|
|
return p,r
|
|
|
|
|
|
def add(store, external, event=None, status='accepted', venue='WWW', year=1995):
|
|
p,r = store.record('fixture',external,'https://example.org/'+external,'Same paper in two events',
|
|
'Abstract\x00with\x01text',['A Person'],{'arxiv':'2501.00001'})
|
|
edition = store.decision(p,r,venue,year,'main',external,'https://example.org/'+status,status,status,event_key=event)
|
|
return p,r,edition
|
|
|
|
|
|
def file(root, name, rows, event=None):
|
|
p=root/(name+'.jsonl');p.write_text(''.join(json.dumps(r)+'\n'for r in rows))
|
|
manifest={'source':'fixture','venue':'WWW','year':1995,'track':'main','count':len(rows),'list_url':'https://example.org/'+name}
|
|
if event is not None:manifest['event_key']=event
|
|
p.with_suffix('.manifest.json').write_text(json.dumps(manifest));return p
|
|
|
|
|
|
class EventIdentityTests(unittest.TestCase):
|
|
def setUp(self):
|
|
t=tempfile.TemporaryDirectory();self.addCleanup(t.cleanup);self.root=Path(t.name);self.db=self.root/'sample.sqlite'
|
|
|
|
def test_same_paper_distinct_events_keep_both_attempts_and_decisions(self):
|
|
s=Store(self.db);self.addCleanup(s.conn.close)
|
|
paper,_,spring=add(s,'spring','www-1995-spring','rejected');_,_,fall=add(s,'fall','www-1995-fall')
|
|
s.conn.commit();self.assertNotEqual(spring,fall)
|
|
with redirect_stdout(io.StringIO())as out:querying.main(['--db',str(self.db),'--paper',str(paper)])
|
|
row=json.loads(out.getvalue())[0];self.assertEqual({a['event_key']for a in row['submissions']},{'www-1995-spring','www-1995-fall'})
|
|
self.assertEqual([a['event_key']for a in row['known_acceptances']],['www-1995-fall'])
|
|
self.assertEqual({a['latest_status']for a in row['submissions']},{'rejected','accepted'})
|
|
with redirect_stdout(io.StringIO())as out:querying.main(['--db',str(self.db),'--venue','WWW','--event-key','www-1995-spring','--status','rejected'])
|
|
self.assertEqual(len(json.loads(out.getvalue())),1)
|
|
|
|
def test_explicit_migration_preserves_ids_history_and_source_bytes(self):
|
|
paper,record=legacy(self.db);proposal=plan(self.db);self.assertFalse(proposal['before']['event_column'])
|
|
result=upgrade(proposal,self.db);self.assertEqual(result['updated_editions'],1)
|
|
self.assertEqual(result['after']['protected'],proposal['before']['protected'])
|
|
self.assertEqual(result['after']['editions_sha256'],proposal['before']['editions_sha256'])
|
|
self.assertEqual(upgrade(proposal,self.db)['status'],'already_applied')
|
|
s=Store(self.db);self.addCleanup(s.conn.close)
|
|
self.assertEqual(s.conn.execute('SELECT id,event_key FROM editions').fetchall(),[(1,'www-1995-fall')])
|
|
# An old file without event_key still has its original source anchor.
|
|
self.assertEqual(s.decision(paper,record,'WWW',1995,'main','conf/www/Fall95','https://example.org/accept','accepted','Accept'),1)
|
|
self.assertEqual(s.conn.execute('SELECT count(*) FROM submissions').fetchone()[0],1)
|
|
self.assertEqual(s.conn.execute('PRAGMA foreign_key_check').fetchall(),[])
|
|
|
|
def test_unverified_legacy_volume_is_not_automatically_labelled(self):
|
|
legacy(self.db)
|
|
with sqlite3.connect(self.db)as c:c.execute("UPDATE source_records SET metadata_json='{}'")
|
|
with self.assertRaisesRegex(ValueError,'complete pinned Fall'):plan(self.db)
|
|
with sqlite3.connect(self.db)as c:self.assertFalse(has_event_key(c))
|
|
|
|
def test_stale_plan_rolls_back_before_schema_change(self):
|
|
legacy(self.db);proposal=plan(self.db)
|
|
with sqlite3.connect(self.db)as c:c.execute("UPDATE editions SET track='unspecified'")
|
|
with self.assertRaisesRegex(ValueError,'stale'):upgrade(proposal,self.db)
|
|
with sqlite3.connect(self.db)as c:self.assertFalse(has_event_key(c));self.assertEqual(c.execute('SELECT track FROM editions').fetchone()[0],'unspecified')
|
|
|
|
def test_unscoped_source_and_wrong_refresh_event_fail(self):
|
|
s=Store(self.db);self.addCleanup(s.conn.close);_,r,e=add(s,'first','www-1995-fall');s.conn.commit()
|
|
p,r2=s.record('fixture','unscoped','https://example.org/u','Other work',None,[])
|
|
with self.assertRaisesRegex(ValueError,'event_key'):s.decision(p,r2,'WWW',1995,'main','unscoped','https://example.org/u','accepted','Accept')
|
|
with self.assertRaisesRegex(ValueError,'Cannot move'):s.decision(p,r,'WWW',1995,'main','first','https://example.org/u','accepted','Accept',event_key='www-1995-spring')
|
|
self.assertEqual(s.conn.execute('SELECT count(*) FROM editions').fetchone()[0],1)
|
|
|
|
def test_pinned_fall_cannot_claim_spring(self):
|
|
p,r=legacy(self.db);upgrade(plan(self.db),self.db);s=Store(self.db);self.addCleanup(s.conn.close)
|
|
with self.assertRaisesRegex(ValueError,'pinned WWW Fall'):s.decision(p,r,'WWW',1995,'main','conf/www/Fall95','https://example.org/u','accepted','Accept',event_key='www-1995-spring')
|
|
|
|
def test_manifest_event_mismatch_rolls_back_the_whole_file(self):
|
|
s=Store(self.db);s.conn.close()
|
|
base={'source':'fixture','venue':'WWW','year':1995,'track':'main','status':'accepted','title':'Input work','url':'https://example.org/p','authors':[]}
|
|
path=file(self.root,'bad',[dict(base,external_id='a',event_key='www-1995-spring'),dict(base,external_id='b',event_key='www-1995-fall')],'www-1995-spring')
|
|
with self.assertRaisesRegex(ValueError,'edition differs'):ingest_file(self.db,path)
|
|
with sqlite3.connect(self.db)as c:
|
|
self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0],0)
|
|
self.assertEqual(c.execute('SELECT count(*) FROM editions').fetchone()[0],0)
|
|
|
|
def test_backfill_selects_the_explicit_event_and_preserves_attempts(self):
|
|
s=Store(self.db);add(s,'spring','www-1995-spring','rejected');add(s,'fall','www-1995-fall');s.conn.commit();s.conn.close()
|
|
row={'source':'fixture','external_id':'fall','venue':'WWW','year':1995,'track':'main','event_key':'www-1995-fall','status':'accepted','identifiers':{'openreview':'forumFall'}}
|
|
path=file(self.root,'backfill',[row],'www-1995-fall')
|
|
with redirect_stdout(io.StringIO()):result=backfill(self.db,[path])
|
|
self.assertEqual(result['submissions_after'],2)
|
|
with sqlite3.connect(self.db)as c:
|
|
self.assertEqual(c.execute("SELECT e.event_key FROM submission_identifiers si JOIN submissions s ON s.id=si.submission_id JOIN editions e ON e.id=s.edition_id WHERE si.scheme='openreview' AND si.value='forumFall'").fetchone()[0],'www-1995-fall')
|
|
|
|
def test_exact_paged_counts_cover_composite_key_boundaries(self):
|
|
s=Store(self.db);self.addCleanup(s.conn.close)
|
|
for i in range(7):add(s,'entry'+str(i),'www-1995-fall')
|
|
s.conn.commit();events=[];counts=exact_counts(s.conn,page_size=2,progress=events.append)
|
|
self.assertEqual(counts['source_records'],7)
|
|
self.assertEqual(counts['identifiers'],8)
|
|
self.assertEqual(counts['submission_sources'],7)
|
|
self.assertEqual([e['rows_counted']for e in events if e['table']=='identifiers'and e['complete']],[8])
|
|
|
|
def test_legacy_single_event_writes_and_reads_remain_available(self):
|
|
legacy(self.db);s=Store(self.db);self.addCleanup(s.conn.close)
|
|
add(s,'ordinary',venue='COLM',year=2025);s.conn.commit()
|
|
with redirect_stdout(io.StringIO())as out:querying.main(['--db',str(self.db),'--venue','COLM'])
|
|
self.assertEqual(json.loads(out.getvalue())[0]['known_acceptances'][-1]['event_key'],'')
|
|
with self.assertRaisesRegex(ValueError,'migration required'):
|
|
querying.main(['--db',str(self.db),'--event-key','www-1995-fall'])
|
|
|
|
def test_unexpected_extra_edition_column_requires_review(self):
|
|
legacy(self.db)
|
|
with sqlite3.connect(self.db)as c:c.execute('ALTER TABLE editions ADD COLUMN custom_data TEXT')
|
|
with self.assertRaisesRegex(ValueError,'Unexpected edition schema'):plan(self.db)
|
|
|
|
def test_file_spanning_two_events_without_manifest_key_rolls_back(self):
|
|
Store(self.db).conn.close()
|
|
base={'source':'fixture','venue':'WWW','year':1995,'track':'main','status':'accepted','title':'Input work','url':'https://example.org/p','authors':[]}
|
|
path=file(self.root,'two',[dict(base,external_id='a',event_key='www-1995-spring'),dict(base,external_id='b',event_key='www-1995-fall')])
|
|
with self.assertRaisesRegex(ValueError,'exactly one'):ingest_file(self.db,path)
|
|
with sqlite3.connect(self.db)as c:self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0],0)
|
|
|
|
def test_cutover_sqlite_backup_remains_protected(self):
|
|
legacy(self.db);proposal=plan(self.db);Path(str(self.db.resolve())+'.postgresql-cutover.json').write_text('{}')
|
|
with self.assertRaisesRegex(RuntimeError,'cut over'):upgrade(proposal,self.db)
|
|
|
|
|
|
@unittest.skipUnless(os.environ.get('PAPER_ATLAS_TEST_PG_SERVICE'),'Requires approved remote PostgreSQL test service')
|
|
class PostgreSQLEventIdentityTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.service=os.environ['PAPER_ATLAS_TEST_PG_SERVICE'];self.schema='pa_event_'+uuid.uuid4().hex[:12]
|
|
t=tempfile.TemporaryDirectory();self.addCleanup(t.cleanup);self.root=Path(t.name);self.db=self.root/'seed.sqlite';legacy(self.db)
|
|
self.addCleanup(self.cleanup)
|
|
with redirect_stdout(io.StringIO()):migrate(self.db,self.service,self.root/'migrate.json',schema=self.schema,min_free_gib=0)
|
|
|
|
def cleanup(self):
|
|
with connect(self.service)as c:
|
|
for (name,)in c.execute('SELECT nspname FROM pg_namespace WHERE starts_with(nspname,%s)',(self.schema,)).fetchall():c.execute('DROP SCHEMA '+quote(name)+' CASCADE')
|
|
|
|
def test_native_upgrade_preserves_ids_and_codec_then_accepts_separate_event(self):
|
|
proposal=plan(pg_service=self.service,pg_schema=self.schema)
|
|
result=upgrade(proposal,pg_service=self.service,pg_schema=self.schema)
|
|
self.assertEqual(result['after']['protected'],proposal['before']['protected'])
|
|
self.assertEqual(upgrade(proposal,pg_service=self.service,pg_schema=self.schema)['status'],'already_applied')
|
|
with opened_store(pg_service=self.service,pg_schema=self.schema)as s:
|
|
paper,_,spring=add(s,'spring','www-1995-spring','rejected');s.conn.commit()
|
|
self.assertNotEqual(spring,1)
|
|
self.assertEqual(s.conn.execute('SELECT abstract FROM papers WHERE id=?',(paper,)).fetchone()[0],'Abstract\x00with\x01text')
|
|
with redirect_stdout(io.StringIO())as out:querying.main(['--pg-service',self.service,'--pg-schema',self.schema,'--paper',str(paper)])
|
|
row=json.loads(out.getvalue())[0];self.assertEqual(len(row['submissions']),2);self.assertEqual(len(row['known_acceptances']),1)
|
|
|
|
def test_native_paged_counts_preserve_encoded_composite_key_order(self):
|
|
with opened_store(pg_service=self.service,pg_schema=self.schema)as s:
|
|
for value in ['a\x00b','a\x01b','a\x02b','z']:
|
|
s.conn.execute('INSERT INTO identifiers(scheme,value,paper_id) VALUES(?,?,?)',('custom',value,1))
|
|
s.conn.commit();counts=exact_counts(s.conn,page_size=2)
|
|
self.assertEqual(counts['identifiers'],6)
|
|
self.assertEqual(counts['submissions'],1)
|
|
|
|
def test_stale_native_plan_is_rejected_and_lock_released(self):
|
|
proposal=plan(pg_service=self.service,pg_schema=self.schema)
|
|
with opened_store(pg_service=self.service,pg_schema=self.schema)as s:s.conn.execute("UPDATE editions SET track='unspecified'");s.conn.commit()
|
|
with self.assertRaisesRegex(ValueError,'stale'):upgrade(proposal,pg_service=self.service,pg_schema=self.schema)
|
|
with opened_store(pg_service=self.service,pg_schema=self.schema)as s:self.assertFalse(has_event_key(s.conn))
|