173 lines
9.8 KiB
Python
173 lines
9.8 KiB
Python
"""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')
|