"""Interactions between event identity and previously independent write safeguards.""" from contextlib import contextmanager, redirect_stdout import hashlib import io import json import os from pathlib import Path import tempfile import unittest import uuid from paper_atlas import querying from paper_atlas.audits.coverage import report as coverage_report from paper_atlas.collectors.dblp import XML_SHA from paper_atlas.curation.annotations import initialize, assert_annotation, retract_annotation, observations from paper_atlas.db.backend import open_store from paper_atlas.ingestion.event_identity import plan, upgrade from paper_atlas.ingestion.records import ingest_file from paper_atlas.postgres.client import connect from paper_atlas.postgres.migration import migrate from paper_atlas.postgres.schema import quote from tests.test_event_identity import legacy, add class RuntimeCases: @contextmanager def store(self): s = open_store(**self.target) try: yield s finally: s.conn.close() def query(self, *args): target = (['--db', str(self.target['db'])] if 'db' in self.target else ['--pg-service', self.target['pg_service'], '--pg-schema', self.target['pg_schema']]) with redirect_stdout(io.StringIO()) as out: querying.main([*target, *map(str,args)]) return json.loads(out.getvalue()) def assertion(self, c, paper, request, **changes): args = dict(paper=paper, kind='research_type', namespace='custom:method', value='empirical', request_id=request, evidence_key='review:'+request, actor='reviewer', reason='Reviewed metadata') args.update(changes) return assert_annotation(c, **args) def test_event_upgrade_preserves_annotation_history_and_combined_queries(self): with self.store() as s: initialize(s.conn) event = self.assertion(s.conn,1,'first') retract_annotation(s.conn,event,request_id='retract',actor='reviewer',reason='Corrected') self.assertion(s.conn,1,'second') original = observations(s.conn,1,history=True) s.conn.commit() proposal = plan(**self.target) self.assertEqual(proposal['before']['counts']['annotation_events'],3) result = upgrade(proposal, **self.target) self.assertEqual(result['after']['counts'],proposal['before']['counts']) with self.store() as s: self.assertEqual(observations(s.conn,1,history=True),original) paper,_,_ = add(s,'spring','www-1995-spring','rejected') self.assertEqual(paper,1) s.conn.commit() row = self.query('--venue','WWW','--event-key','www-1995-spring','--status','rejected', '--research-type','custom:method=empirical','--paper',1,'--annotation-history')[0] self.assertEqual([a['event_key'] for a in row['known_acceptances']],['www-1995-fall']) self.assertEqual({a['latest_status'] for a in row['submissions']},{'accepted','rejected'}) self.assertEqual(len(row['annotation_history']),3) self.assertEqual(self.query('--event-key','www-1995-fall','--status','rejected'),[]) def collected(self, name, event, external, *, incremental=False, unscoped=False): row = dict(source='fixture',external_id=external,url='https://example.org/'+external, title='Independent contribution '+external,authors=[],abstract='Known abstract', identifiers={'doi':'10.1234/'+external},venue='WWW',year=1995,track='main', event_key=event,status='accepted') p = self.root/(name+'.jsonl');p.write_text(json.dumps(row)+'\n') manifest = {k:row[k] for k in ('source','venue','year','track','event_key')} manifest.update(list_url='https://example.org/'+event,count=1,status='partial', file_sha256=hashlib.sha256(p.read_bytes()).hexdigest()) if incremental: manifest.update(coverage_mode='incremental',coverage_previous_count=1, coverage_previous_expected_count=1,coverage_expected_count=2) if unscoped: manifest.pop('event_key') p.with_suffix('.manifest.json').write_text(json.dumps(manifest)) return p def ingest(self, path): with redirect_stdout(io.StringIO()): ingest_file(self.target.get('db'),path,**{k:v for k,v in self.target.items() if k != 'db'}) def test_coverage_labels_both_same_year_events_and_keeps_legacy_readable(self): registry=self.root/'registry.json' registry.write_text(json.dumps({'scope_note':'Fixture only','venues':[{'acronym':'WWW','field':'web'}]})) def snapshot(): return coverage_report(self.target.get('db'),registry,**{k:v for k,v in self.target.items() if k!='db'})['venues'][0] self.assertEqual([e['event_key']for e in snapshot()['editions']],['']) upgrade(plan(**self.target),**self.target) with self.store() as s: add(s,'spring','www-1995-spring','rejected');s.conn.commit() result=snapshot();events={e['event_key']:e for e in result['editions']} self.assertEqual(set(events),{'www-1995-fall','www-1995-spring'}) self.assertEqual((result['imported_submissions'],result['distinct_papers']),(2,1)) self.assertEqual(events['www-1995-fall']['latest_observed_status_counts']['accepted'],1) self.assertEqual(events['www-1995-spring']['latest_observed_status_counts']['rejected'],1) self.assertEqual(events['www-1995-fall']['source_record_counts'][0]['source'],'dblp') self.assertEqual(events['www-1995-spring']['source_record_counts'][0]['source'],'fixture') def test_incremental_parent_is_scoped_to_exact_event(self): upgrade(plan(**self.target), **self.target) self.ingest(self.collected('spring','www-1995-spring','spring-one')) self.ingest(self.collected('fall','www-1995-fall','fall-one')) self.ingest(self.collected('append','www-1995-fall','fall-two',incremental=True)) with self.store() as s: rows = s.conn.execute('''SELECT e.event_key,es.expected_count,es.imported_count FROM edition_sources es JOIN editions e ON e.id=es.edition_id JOIN sources src ON src.id=es.source_id WHERE src.name='fixture' ORDER BY e.event_key''').fetchall() self.assertEqual([tuple(r) for r in rows],[('www-1995-fall',2,2),('www-1995-spring',1,1)]) with self.assertRaisesRegex(ValueError,'event_key'): self.ingest(self.collected('unscoped','www-1995-spring','unwanted',incremental=True,unscoped=True)) with self.assertRaisesRegex(ValueError,'parent changed'): self.ingest(self.collected('stale','www-1995-fall','stale',incremental=True)) with self.store() as s: self.assertEqual(s.conn.execute("SELECT count(*) FROM source_records r JOIN sources src ON src.id=r.source_id WHERE src.name='fixture'").fetchone()[0],3) def test_joint_refresh_keeps_identity_annotations_and_event_query(self): upgrade(plan(**self.target), **self.target) metadata = dict(snapshot_sha256=XML_SHA,proceedings_key='conf/ijcai/2022', bibliographic_year=2022,bibliography={'booktitle':['IJCAI']}) with self.store() as s: initialize(s.conn) paper,record = s.record('dblp','conf/ijcai/Joint22','https://example.org/joint', 'Joint paper',None,[],{'doi':'10.1234/joint'},metadata) eid = s.decision(paper,record,'IJCAI',2022,'main','conf/ijcai/Joint22','https://example.org/joint','accepted','Accept') self.assertion(s.conn,paper,'joint') s.conn.commit() with self.store() as s: again,r = s.record('dblp','conf/ijcai/Joint22','https://example.org/joint', 'Corrected joint title',None,[],{'doi':'10.1234/joint'},metadata) self.assertEqual(again,paper) self.assertEqual(s.decision(again,r,'IJCAI',2022,'main','conf/ijcai/Joint22','https://example.org/joint','accepted','Accept'),eid) self.assertEqual(s.conn.execute("SELECT count(*) FROM editions e JOIN venues v ON v.id=e.venue_id WHERE v.acronym='IJCAI' AND e.year=2022").fetchone()[0],0) s.conn.commit() for venue in ['IJCAI','ECAI']: rows = self.query('--venue',venue,'--year',2022,'--event-key','', '--research-type','custom:method=empirical','--paper',paper) self.assertEqual(len(rows),1) self.assertEqual(len(rows[0]['submissions']),1) self.assertEqual(rows[0]['submissions'][0]['acronym'],'ECAI+IJCAI') class SQLiteRuntimeTests(RuntimeCases, unittest.TestCase): def setUp(self): t=tempfile.TemporaryDirectory();self.addCleanup(t.cleanup);self.root=Path(t.name) db=self.root/'source.sqlite';legacy(db);self.target={'db':db} @unittest.skipUnless(os.environ.get('PAPER_ATLAS_TEST_PG_SERVICE'),'Requires approved remote PostgreSQL service') class PostgreSQLRuntimeTests(RuntimeCases, unittest.TestCase): def setUp(self): t=tempfile.TemporaryDirectory();self.addCleanup(t.cleanup);self.root=Path(t.name) db=self.root/'source.sqlite';legacy(db) self.service=os.environ['PAPER_ATLAS_TEST_PG_SERVICE'];self.schema='pa_runtime_'+uuid.uuid4().hex[:12] self.addCleanup(self.cleanup) with redirect_stdout(io.StringIO()): migrate(db,self.service,self.root/'migration.json',schema=self.schema,min_free_gib=0) self.target={'pg_service':self.service,'pg_schema':self.schema} 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,)): c.execute('DROP SCHEMA '+quote(name)+' CASCADE')