from contextlib import closing, redirect_stdout import hashlib import io import json import os from pathlib import Path import tempfile import unittest from unittest.mock import patch import uuid from paper_atlas.collectors.dblp import XML_SHA from paper_atlas.collectors.dblp_scope import scope from paper_atlas.collectors.dblp_records import normalize_row from paper_atlas.db.store import Store from paper_atlas.db.backend import open_store from paper_atlas.ingestion.joint_editions import (TARGET, MARKER, inspect, apply_plan, run) from paper_atlas.postgres.client import connect from paper_atlas.postgres.migration import migrate from paper_atlas.postgres.schema import quote from paper_atlas import querying def seed(store): c = store.conn c.execute("INSERT INTO venues(acronym,name,field,venue_type) VALUES(?,?,'reasoning','conference')", (TARGET,TARGET)) target = c.execute('SELECT id FROM venues WHERE acronym=?',(TARGET,)).fetchone()[0] for alias in ['ECAI','IJCAI']: c.execute('INSERT INTO venue_aliases VALUES(?,?)',(target,alias)) for year in [2018,2019,2022]: metadata = {'snapshot_sha256':XML_SHA,'proceedings_key':'conf/ijcai/'+str(year), 'bibliographic_year':year,'bibliography':{'booktitle':['IJCAI']}} p,r=store.record('dblp','conf/ijcai/fixture'+str(year),'https://dblp.org/rec/fixture', 'Original publication '+str(year),'Abstract\x00with\x01data',[], {'doi':'10.24963/ijcai.'+str(year)+'/1'}, {}) store.record_tags(p,r,'dblp',[{'namespace':'dblp:booktitle','value':'IJCAI'}]) store.decision(p,r,'IJCAI',year,'main','fixture'+str(year),'https://example.org/decision','accepted','listed') stored=json.loads(c.execute('SELECT metadata_json FROM source_records WHERE id=?',(r,)).fetchone()[0]);stored.update(metadata) c.execute('UPDATE source_records SET metadata_json=? WHERE id=?',(json.dumps(stored),r)) c.commit() def plan(c): return {**inspect(c),'database':{'engine':'sqlite','path':'fixture'}} class JointEditionTests(unittest.TestCase): def setUp(self): self.tmp=tempfile.TemporaryDirectory(); self.addCleanup(self.tmp.cleanup) self.db=Path(self.tmp.name)/'sample.sqlite' self.store=Store(self.db); self.addCleanup(self.store.conn.close) seed(self.store); self.c=self.store.conn def test_apply_preserves_every_protected_row_and_edition_id(self): before=plan(self.c); ids=[r['id']for r in before['editions']] result=apply_plan(self.c,before);self.c.commit() self.assertEqual(result['updated_editions'],2) self.assertEqual(result['content_fingerprints'],before['content_fingerprints']) self.assertEqual([r['id']for r in result['editions']],ids) untouched=self.c.execute('SELECT v.acronym FROM editions e JOIN venues v ON v.id=e.venue_id WHERE e.year=2019').fetchone()[0] self.assertEqual(untouched,'IJCAI') self.assertEqual(self.c.execute('PRAGMA foreign_key_check').fetchall(),[]) self.assertEqual(self.c.execute('SELECT count(*) FROM schema_migrations WHERE name=?',(MARKER,)).fetchone()[0],1) with redirect_stdout(io.StringIO())as out: querying.main(['--db',str(self.db),'--venue','ECAI']) self.assertEqual(len(json.loads(out.getvalue())),2) self.assertEqual(apply_plan(self.c,before)['updated_editions'],0) def test_refresh_of_old_normalized_source_does_not_recreate_ordinary_edition(self): before=plan(self.c);apply_plan(self.c,before) for year in [2018,2022]: row=self.c.execute('''SELECT r.paper_id,r.id,d.observed_at FROM source_records r JOIN submissions s ON s.source_record_id=r.id JOIN decisions d ON d.submission_id=s.id WHERE r.external_id=?''',('conf/ijcai/fixture'+str(year),)).fetchone() self.store.decision(row[0],row[1],'IJCAI',year,'main','fixture'+str(year), 'https://example.org/decision','accepted','listed',observed_at=row[2]) self.assertEqual(inspect(self.c)['content_fingerprints'],before['content_fingerprints']) self.assertEqual(self.c.execute('SELECT count(*) FROM editions').fetchone()[0],3) def test_any_changed_source_text_rejects_stale_plan_before_membership_updates(self): before=plan(self.c) self.c.execute("UPDATE source_records SET title='New evidence' WHERE external_id='conf/ijcai/fixture2018'") with self.assertRaisesRegex(ValueError,'Stale repair plan'): apply_plan(self.c,before) self.assertEqual(inspect(self.c)['status'],'ready') def test_conflicting_target_edition_is_never_merged_or_deleted(self): target=self.c.execute('SELECT id FROM venues WHERE acronym=?',(TARGET,)).fetchone()[0] self.c.execute("INSERT INTO editions(venue_id,year,track) VALUES(?,2018,'main')",(target,)) with self.assertRaisesRegex(ValueError,'colliding'): inspect(self.c) self.assertEqual(self.c.execute('SELECT count(*) FROM editions WHERE year=2018').fetchone()[0],2) def test_wrong_source_volume_and_snapshot_fail_closed(self): original=self.c.execute("SELECT metadata_json FROM source_records WHERE external_id='conf/ijcai/fixture2018'").fetchone()[0] for key,value in [('proceedings_key','conf/ijcai/2018w'),('snapshot_sha256','wrong'),('bibliographic_year',2017),('bibliography',{'booktitle':['Other']})]: m=json.loads(original);m[key]=value self.c.execute("UPDATE source_records SET metadata_json=? WHERE external_id='conf/ijcai/fixture2018'",(json.dumps(m),)) with self.subTest(key=key),self.assertRaisesRegex(ValueError,'Source evidence'): inspect(self.c) def test_missing_source_support_and_wrong_aliases_are_not_guessed(self): self.c.execute("DELETE FROM venue_aliases WHERE alias='ECAI'") with self.assertRaisesRegex(ValueError,'aliases'): inspect(self.c) self.c.rollback() self.c.execute('DELETE FROM submission_sources WHERE source_record_id=(SELECT id FROM source_records WHERE external_id=?)',('conf/ijcai/fixture2022',)) with self.assertRaisesRegex(ValueError,'Every submission'): inspect(self.c) def test_cli_plan_readonly_hash_binding_identity_and_repeat_apply(self): out=Path(self.tmp.name)/'plan.json' with redirect_stdout(io.StringIO()):run(self.db,out) self.assertEqual(inspect(self.c)['status'],'ready') raw=out.read_bytes();sha=hashlib.sha256(raw).hexdigest() with patch('paper_atlas.ingestion.joint_editions.open_store')as writer: with self.assertRaisesRegex(ValueError,'hash differs'): run(self.db,Path(self.tmp.name)/'bad.json',plan=out,plan_sha256='0'*64) writer.assert_not_called() with redirect_stdout(io.StringIO()): result=run(self.db,Path(self.tmp.name)/'applied.json',plan=out,plan_sha256=sha) again=run(self.db,Path(self.tmp.name)/'retry.json',plan=out,plan_sha256=sha) self.assertEqual(result['updated_editions'],2);self.assertEqual(again['updated_editions'],0) self.assertEqual(result['plan_sha256'],sha);self.assertEqual(again['plan_sha256'],sha) altered=json.loads(raw);altered['database']['path']='another.sqlite' bad=Path(self.tmp.name)/'wrong-db.json';bad.write_text(json.dumps(altered)) with self.assertRaisesRegex(ValueError,'another database'): run(self.db,Path(self.tmp.name)/'wrong-result.json',plan=bad,plan_sha256=hashlib.sha256(bad.read_bytes()).hexdigest()) def test_invalid_empty_plan_fails_before_opening_writer(self): f=Path(self.tmp.name)/'empty.json';f.write_text('{}') with patch('paper_atlas.ingestion.joint_editions.open_store')as writer: with self.assertRaises(ValueError): run(self.db,Path(self.tmp.name)/'empty-result.json',plan=f,plan_sha256=hashlib.sha256(f.read_bytes()).hexdigest()) writer.assert_not_called() def test_collector_only_changes_the_two_exact_main_volume_keys(self): for year in [2018,2019,2022]: volume={'key':'conf/ijcai/'+str(year),'fields':{'title':['IJCAI '+str(year)],'year':[str(year)]}} row={'record_type':'inproceedings','key':'conf/ijcai/One'+str(year),'fields':{'booktitle':['IJCAI'], 'year':[str(year)],'crossref':[volume['key']],'title':['Original research title'], 'author':['Jane Doe'],'ee':['https://doi.org/10.24963/ijcai.'+str(year)+'/1']}} result,error=normalize_row(row,{volume['key']:volume},{'IJCAI','ECAI'}) self.assertIsNone(error) self.assertEqual(result['venue'],TARGET if year in(2018,2022)else'IJCAI') self.assertEqual(result['metadata']['bibliography']['booktitle'],['IJCAI']) self.assertEqual(result['title'],'Original research title') if year in(2018,2022): self.assertIn('joint_edition_evidence',result['metadata']) row['fields']['booktitle']=['IJCAI (Workshop)'] scoped,_=scope(row,volume) self.assertEqual(scoped['venue'],'IJCAI') self.assertEqual(scoped['track'],'workshop') @unittest.skipUnless(os.environ.get('PAPER_ATLAS_TEST_PG_SERVICE'),'Requires approved remote PostgreSQL service') class PostgreSQLJointTests(unittest.TestCase): def setUp(self): self.service=os.environ['PAPER_ATLAS_TEST_PG_SERVICE'];self.target='pa_joint_'+uuid.uuid4().hex[:12] self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup) self.db=Path(self.tmp.name)/'sample.sqlite';s=Store(self.db);seed(s);s.conn.close() self.addCleanup(self.cleanup) with redirect_stdout(io.StringIO()):migrate(self.db,self.service,Path(self.tmp.name)/'migration.json',schema=self.target,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.target,)).fetchall(): c.execute('DROP SCHEMA '+quote(name)+' CASCADE') def test_native_repair_is_idempotent_keeps_codec_and_provenance(self): p=Path(self.tmp.name)/'plan.json' with redirect_stdout(io.StringIO()): before=run(None,p,pg_service=self.service,pg_schema=self.target) after=run(None,Path(self.tmp.name)/'applied.json',pg_service=self.service,pg_schema=self.target,plan=p,plan_sha256=hashlib.sha256(p.read_bytes()).hexdigest()) self.assertEqual(after['content_fingerprints'],before['content_fingerprints']) with closing(open_store(pg_service=self.service,pg_schema=self.target).conn)as c: self.assertEqual(apply_plan(c,before)['updated_editions'],0) self.assertEqual(c.execute('SELECT abstract FROM papers ORDER BY id LIMIT 1').fetchone()[0],'Abstract\x00with\x01data') def test_native_stale_source_plan_preserves_both_original_memberships(self): with closing(open_store(pg_service=self.service,pg_schema=self.target).conn)as c: before=plan(c) c.execute("UPDATE source_records SET title='Changed' WHERE external_id=?",('conf/ijcai/fixture2022',));c.commit() with self.assertRaisesRegex(ValueError,'Stale repair plan'): apply_plan(c,before) c.rollback() self.assertEqual(inspect(c)['status'],'ready')