160 lines
11 KiB
Python
160 lines
11 KiB
Python
"""Stable-attempt correction, rollback and stale whole-volume refresh on both engines."""
|
|
from contextlib import closing, redirect_stdout
|
|
import copy
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
import uuid
|
|
|
|
from paper_atlas.db.backend import open_store
|
|
from paper_atlas.db.edition_scopes import KEY
|
|
from paper_atlas.db.store import Store
|
|
from paper_atlas.ingestion.edition_tracks import proposal,apply_plan,verify,validate,run
|
|
from paper_atlas.ingestion.event_identity import database_identity,values
|
|
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
|
|
|
|
|
|
def seed(db):
|
|
s=Store(db); rows=[]
|
|
for i,track in enumerate(['main','survey','doctoral_consortium','proceedings_unspecified'],1):
|
|
external='entry'+str(i);doi='10.1234/entry'+str(i)
|
|
p,r=s.record('fixture',external,'https://example.org/'+external,'Publication '+external,
|
|
'Abstract with NUL\x00 and SOH\x01',[],{'doi':doi})
|
|
eid=s.decision(p,r,'JOINT',2022,'main',external,'https://example.org/'+external,'accepted','published',identities=[('doi',doi)],observed_at='2022-01-01T00:00:00Z')
|
|
sid=s.conn.execute('SELECT id FROM submissions WHERE source_record_id=?',(r,)).fetchone()[0]
|
|
s.record_tags(p,r,'fixture',[{'namespace':'fixture:subject','value':'Original'}])
|
|
rows.append(dict(submission_id=sid,source_record_id=r,external_id=external,source='fixture',doi=doi,
|
|
from_edition_id=eid,year=2022,venue='JOINT',venue_names=['JOINT'],track=track,
|
|
section=track,evidence_url='https://example.org/proceedings/2022/',evidence_sha256='a'*64))
|
|
source=s.conn.execute("SELECT id FROM sources WHERE name='fixture'").fetchone()[0]
|
|
s.conn.execute("INSERT INTO edition_sources VALUES(?,?,?,'complete',4,4,'checkpoint','2022-01-01T00:00:00Z',NULL)",(eid,source,'https://example.org/volume'))
|
|
s.conn.commit();s.conn.close();return {'format':1,'records':rows}
|
|
|
|
|
|
class TrackCases:
|
|
def store(self):return open_store(**self.target)
|
|
|
|
def plan(self,c):return proposal(c,self.inventory,database_identity(c,self.target.get('db')))
|
|
|
|
def test_move_preserves_all_history_tags_text_and_stable_identities(self):
|
|
with closing(self.store().conn)as c:
|
|
plan=self.plan(c);before=plan['before']['protected'];r=apply_plan(c,plan);c.commit()
|
|
self.assertEqual((r['events_moved'],r['records_verified']),(3,4))
|
|
self.assertEqual(r['protected'],before)
|
|
self.assertEqual(apply_plan(c,plan)['status'],'already_applied')
|
|
self.assertEqual(values(c.execute('SELECT status,expected_count,imported_count FROM edition_sources ORDER BY edition_id')),[['partial',1,1]]*4)
|
|
self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0],4)
|
|
self.assertEqual(c.execute('SELECT count(*) FROM decisions').fetchone()[0],4)
|
|
self.assertEqual(c.execute('SELECT count(*) FROM paper_merges').fetchone()[0],0)
|
|
|
|
def test_source_record_refresh_keeps_scope_and_original_attempt(self):
|
|
with closing(self.store().conn)as c:
|
|
plan=self.plan(c);apply_plan(c,plan);c.commit()
|
|
with closing(self.store().conn)as c:
|
|
# Use Store methods with the same writer connection as production.
|
|
s=object.__new__(Store);s.conn=c;s.event_identity=True
|
|
r=self.inventory['records'][1];metadata=json.loads(c.execute('SELECT metadata_json FROM source_records WHERE id=?',(r['source_record_id'],)).fetchone()[0]);saved=metadata[KEY]
|
|
p,record=s.record('fixture',r['external_id'],'https://example.org/new','Refreshed title',None,[],{'doi':r['doi']},{'new_source_field':True})
|
|
eid=s.decision(p,record,'JOINT',2022,'main',r['external_id'],'https://example.org/entry2','accepted','published',identities=[('doi',r['doi'])],observed_at='2022-01-01T00:00:00Z')
|
|
self.assertEqual(c.execute('SELECT track FROM editions WHERE id=?',(eid,)).fetchone()[0],'survey')
|
|
self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0],4)
|
|
self.assertEqual(json.loads(c.execute('SELECT metadata_json FROM source_records WHERE id=?',(record,)).fetchone()[0])[KEY],saved)
|
|
with self.assertRaisesRegex(ValueError,'cannot supply'):
|
|
s.record('fixture',r['external_id'],'https://example.org/new','Forged',None,[],metadata={KEY:[]})
|
|
c.rollback()
|
|
|
|
def test_stale_snapshot_and_partial_inventory_are_rejected(self):
|
|
with closing(self.store().conn)as c:
|
|
plan=self.plan(c)
|
|
c.execute("UPDATE papers SET title='Changed' WHERE id=(SELECT paper_id FROM submissions WHERE id=?)",(self.inventory['records'][0]['submission_id'],))
|
|
with self.assertRaisesRegex(ValueError,'Stale'):apply_plan(c,plan)
|
|
c.rollback()
|
|
bad=copy.deepcopy(self.inventory);bad['records'].pop()
|
|
with self.assertRaisesRegex(ValueError,'every attempt'):proposal(c,bad,database_identity(c,self.target.get('db')))
|
|
bad=copy.deepcopy(self.inventory);bad['records'][0]['doi']='10.1234/wrong'
|
|
with self.assertRaisesRegex(ValueError,'DOI'):proposal(c,bad,database_identity(c,self.target.get('db')))
|
|
|
|
def test_failure_after_updates_rolls_back_all_scope_and_metadata_changes(self):
|
|
with closing(self.store().conn)as c:
|
|
plan=self.plan(c)
|
|
# A late coverage failure must roll back earlier moved attempts.
|
|
source=c.execute("SELECT id FROM sources WHERE name='fixture'").fetchone()[0]
|
|
c.execute("INSERT INTO sources(name,homepage) VALUES('other','https://example.org')")
|
|
other=c.execute("SELECT id FROM sources WHERE name='other'").fetchone()[0]
|
|
c.execute('UPDATE edition_sources SET source_id=? WHERE source_id=?',(other,source));c.commit()
|
|
plan=self.plan(c)
|
|
try:
|
|
with self.assertRaisesRegex(ValueError,'outside'):apply_plan(c,plan)
|
|
finally:c.rollback()
|
|
self.assertEqual(self.plan(c)['before'],plan['before'])
|
|
self.assertEqual(c.execute('SELECT count(*) FROM editions').fetchone()[0],1)
|
|
|
|
def test_import_old_whole_volume_fails_without_corrupting_coverage(self):
|
|
with closing(self.store().conn)as c:
|
|
plan=self.plan(c);apply_plan(c,plan);c.commit()
|
|
r=self.inventory['records'][1];path=self.root/'old.jsonl'
|
|
row=dict(source='fixture',external_id=r['external_id'],url='https://example.org/entry2',title='Old title',authors=[],identifiers={'doi':r['doi']},venue='JOINT',year=2022,track='main',status='accepted')
|
|
path.write_text(json.dumps(row)+'\n');path.with_suffix('.manifest.json').write_text(json.dumps(dict(source='fixture',venue='JOINT',year=2022,track='main',count=1,list_url='https://example.org/volume',status='complete')))
|
|
with self.assertRaisesRegex(ValueError,'Recollect'):ingest_file(self.target.get('db'),path,**{k:v for k,v in self.target.items()if k!='db'})
|
|
with closing(self.store().conn)as c:verify(c,plan)
|
|
|
|
def test_nonempty_target_and_alias_drift_fail_before_moving(self):
|
|
with closing(self.store().conn)as c:
|
|
c.execute("INSERT INTO venue_aliases VALUES((SELECT id FROM venues WHERE acronym='JOINT'),'ADDED')")
|
|
with self.assertRaisesRegex(ValueError,'aliases'):self.plan(c)
|
|
c.rollback()
|
|
s=object.__new__(Store);s.conn=c;s.event_identity=True
|
|
p,r=s.record('fixture','collision','https://example.org/collision','Distinct contribution',None,[],{'doi':'10.1234/collision'})
|
|
s.decision(p,r,'JOINT',2022,'survey','collision','https://example.org/collision','accepted','published')
|
|
with self.assertRaisesRegex(ValueError,'Nonempty target'):self.plan(c)
|
|
c.rollback()
|
|
self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0],4)
|
|
|
|
def test_repeated_plan_detects_coverage_or_scope_evidence_drift(self):
|
|
with closing(self.store().conn)as c:
|
|
plan=self.plan(c);apply_plan(c,plan);c.commit()
|
|
c.execute("UPDATE edition_sources SET imported_count=99")
|
|
with self.assertRaisesRegex(ValueError,'coverage'):apply_plan(c,plan)
|
|
c.rollback()
|
|
record=self.inventory['records'][0]['source_record_id']
|
|
m=json.loads(c.execute('SELECT metadata_json FROM source_records WHERE id=?',(record,)).fetchone()[0]);m[KEY]=[]
|
|
c.execute('UPDATE source_records SET metadata_json=? WHERE id=?',(json.dumps(m),record))
|
|
with self.assertRaisesRegex(ValueError,'unchanged track'):apply_plan(c,plan)
|
|
c.rollback()
|
|
|
|
def test_cli_hash_binding_preflight_and_duplicate_inventory(self):
|
|
inventory=self.root/'inventory.json';inventory.write_text(json.dumps(self.inventory))
|
|
planfile=self.root/'plan.json';p=run(self.target.get('db'),planfile,inventory=inventory,**{k:v for k,v in self.target.items()if k!='db'})
|
|
with self.assertRaisesRegex(ValueError,'hash'):run(self.target.get('db'),self.root/'bad.json',plan=planfile,plan_sha256='0'*64,**{k:v for k,v in self.target.items()if k!='db'})
|
|
report=run(self.target.get('db'),self.root/'applied.json',plan=planfile,plan_sha256=hashlib.sha256(planfile.read_bytes()).hexdigest(),**{k:v for k,v in self.target.items()if k!='db'})
|
|
self.assertEqual(report['events_moved'],3)
|
|
bad=copy.deepcopy(self.inventory);bad['records'].append(bad['records'][0])
|
|
with self.assertRaisesRegex(ValueError,'Repeated'):validate(bad)
|
|
|
|
|
|
class SQLiteTrackTests(TrackCases,unittest.TestCase):
|
|
def setUp(self):
|
|
t=tempfile.TemporaryDirectory();self.addCleanup(t.cleanup);self.root=Path(t.name);db=self.root/'source.sqlite'
|
|
self.inventory=seed(db);self.target={'db':db}
|
|
|
|
|
|
@unittest.skipUnless(os.environ.get('PAPER_ATLAS_TEST_PG_SERVICE'),'Requires approved remote PostgreSQL service')
|
|
class PostgreSQLTrackTests(TrackCases,unittest.TestCase):
|
|
def setUp(self):
|
|
t=tempfile.TemporaryDirectory();self.addCleanup(t.cleanup);self.root=Path(t.name);db=self.root/'source.sqlite';self.inventory=seed(db)
|
|
self.service=os.environ['PAPER_ATLAS_TEST_PG_SERVICE'];self.schema='pa_tracks_'+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')
|