Zayden-Jung
4ec82c8f8f
Metadata integrity checks / metadata (push) Successful in 1m20s
Reviewed-on: #27
447 lines
27 KiB
Python
447 lines
27 KiB
Python
from contextlib import closing, redirect_stdout
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import patch
|
|
import uuid
|
|
import zipfile
|
|
|
|
from paper_atlas.db.backend import open_store, open_read_connection
|
|
from paper_atlas.db.store import Store
|
|
from paper_atlas.ingestion import arxiv
|
|
from paper_atlas.ingestion.backfill import backfill
|
|
from paper_atlas.ingestion.records import ingest_file
|
|
from paper_atlas.postgres.client import connect, ReadConnection
|
|
from paper_atlas.postgres.migration import migrate
|
|
from paper_atlas.postgres.schema import quote
|
|
|
|
|
|
class BackendSelectionTests(unittest.TestCase):
|
|
def test_ambiguous_targets_fail_before_opening_any_database(self):
|
|
for kwargs in ({}, {'db': 'example', 'pg_service': 'example'}, {'db': 'example', 'pg_schema': 'research'}):
|
|
with self.assertRaisesRegex(ValueError, 'Choose exactly one'):
|
|
open_store(**kwargs)
|
|
|
|
def test_readonly_audit_does_not_create_an_absent_sqlite_database(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = Path(tmp) / 'absent.sqlite'
|
|
with self.assertRaisesRegex(ValueError, 'Existing database'):
|
|
open_read_connection(path)
|
|
self.assertFalse(path.exists())
|
|
|
|
|
|
@unittest.skipUnless(os.environ.get('PAPER_ATLAS_TEST_PG_SERVICE'), 'Requires the approved remote PostgreSQL test service')
|
|
class PostgreSQLIngestionTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.service = os.environ['PAPER_ATLAS_TEST_PG_SERVICE']
|
|
self.schema = 'pa_write_' + uuid.uuid4().hex[:12]
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
source = self.root / 'seed.sqlite'
|
|
store = Store(source)
|
|
paper, record = store.record('arxiv', '2401.00001', 'https://arxiv.org/abs/2401.00001',
|
|
'Existing reference paper', 'Existing abstract', [])
|
|
store.decision(paper, record, 'ICLR', 2024, 'main', 'seed', 'https://example.org/seed', 'rejected', 'Reject')
|
|
store.conn.execute("INSERT INTO paper_merges(former_paper_id,surviving_paper_id,reason) VALUES(900,?,'fixture')", (paper,))
|
|
store.conn.execute("INSERT INTO submission_merges VALUES(5000,1,'fixture',CURRENT_TIMESTAMP)")
|
|
store.conn.execute("INSERT INTO decision_merges VALUES(6000,1,'fixture',CURRENT_TIMESTAMP)")
|
|
store.conn.execute("UPDATE sqlite_sequence SET seq=1000 WHERE name='papers'")
|
|
store.conn.commit(); store.conn.close()
|
|
self.addCleanup(self.cleanup)
|
|
with redirect_stdout(io.StringIO()):
|
|
migrate(source, self.service, self.root / 'migration.json', schema=self.schema, min_free_gib=0)
|
|
|
|
def cleanup(self):
|
|
with connect(self.service) as pg:
|
|
names = pg.execute('SELECT nspname FROM pg_namespace WHERE starts_with(nspname,%s)', (self.schema,)).fetchall()
|
|
for name, in names:
|
|
pg.execute('DROP SCHEMA ' + quote(name) + ' CASCADE')
|
|
|
|
def store(self):
|
|
return open_store(pg_service=self.service, pg_schema=self.schema)
|
|
|
|
def row(self, source='openreview_snapshot', external='forumNew1', status='rejected', year=2025, venue='ICLR'):
|
|
return {'source': source, 'external_id': external, 'url': 'https://example.org/' + external,
|
|
'venue': venue, 'year': year, 'track': 'main', 'status': status,
|
|
'title': 'A newly observed shared paper', 'abstract': 'Before\x00survivor and literal\x01\x02',
|
|
'authors': ['Jane Doe', 'John Smith'], 'identifiers': {'arxiv': '2501.99999'},
|
|
'observed_at': str(year) + '-01-20T00:00:00Z'}
|
|
|
|
def file(self, name, rows, count=None):
|
|
path = self.root / (name + '.jsonl')
|
|
path.write_text(''.join(json.dumps(r, ensure_ascii=False) + '\n' for r in rows))
|
|
first = rows[0]
|
|
manifest = {key: first[key] for key in ('source', 'venue', 'year', 'track')}
|
|
manifest.update(parser_version=2, list_url='https://example.org/' + name,
|
|
count=len(rows) if count is None else count, missing_abstract_count=0, status='complete')
|
|
path.with_suffix('.manifest.json').write_text(json.dumps(manifest))
|
|
return path
|
|
|
|
def ingest(self, path):
|
|
with redirect_stdout(io.StringIO()):
|
|
ingest_file(None, path, pg_service=self.service, pg_schema=self.schema)
|
|
|
|
def test_incremental_source_coverage_counts_prior_chunks_and_refreshes_on_postgres(self):
|
|
rows = [self.row(source='dblp',external=key,status='accepted',year=2001,venue='UAI')
|
|
for key in ('incremental1','incremental2','incremental3')]
|
|
listing = 'https://example.org/uai/2001'
|
|
def manifest(path, **options):
|
|
target=path.with_suffix('.manifest.json')
|
|
value=json.loads(target.read_text())
|
|
value.update(list_url=listing,status='partial',**options)
|
|
target.write_text(json.dumps(value))
|
|
first=self.file('initial',rows[:2]);manifest(first);self.ingest(first)
|
|
second=self.file('append',[rows[2]])
|
|
manifest(second,coverage_mode='incremental',coverage_previous_count=2,
|
|
coverage_previous_expected_count=2,coverage_expected_count=4)
|
|
self.ingest(second)
|
|
manifest(second,coverage_mode='incremental',coverage_previous_count=3,
|
|
coverage_previous_expected_count=4,coverage_expected_count=4)
|
|
self.ingest(second)
|
|
with closing(ReadConnection(self.service,self.schema)) as c:
|
|
coverage=c.execute('SELECT status,expected_count,imported_count FROM edition_sources WHERE list_url=?',(listing,)).fetchone()
|
|
self.assertEqual(tuple(coverage.values),('partial',4,3))
|
|
stored=c.execute('SELECT abstract FROM source_records WHERE external_id=?',('incremental3',)).fetchone()
|
|
self.assertEqual(stored[0],rows[2]['abstract'])
|
|
self.assertEqual(c.execute("SELECT count(*) FROM source_records r JOIN sources s ON s.id=r.source_id WHERE s.name='dblp'").fetchone()[0],3)
|
|
run=c.execute('SELECT records_written,checkpoint_json FROM ingest_runs ORDER BY id DESC LIMIT 1').fetchone()
|
|
self.assertEqual(run[0],1)
|
|
self.assertEqual(json.loads(run[1])['coverage']['imported_count'],3)
|
|
|
|
def test_incremental_postgres_bound_failure_rolls_back_rows_and_summary(self):
|
|
listing='https://example.org/uai/2001'
|
|
rows=[self.row(source='dblp',external=key,status='accepted',year=2001,venue='UAI')
|
|
for key in ('bound1','bound2','bound3')]
|
|
first=self.file('initial',rows[:2]);target=first.with_suffix('.manifest.json')
|
|
value=json.loads(target.read_text());value.update(list_url=listing,status='partial');target.write_text(json.dumps(value))
|
|
self.ingest(first)
|
|
with closing(ReadConnection(self.service,self.schema)) as c:
|
|
before={table:c.execute('SELECT count(*) FROM '+table).fetchone()[0]
|
|
for table in ('papers','source_records','submissions','decisions')}
|
|
path=self.file('overflow',[rows[2]]);target=path.with_suffix('.manifest.json')
|
|
value=json.loads(target.read_text());value.update(list_url=listing,status='partial',coverage_mode='incremental',
|
|
coverage_previous_count=2,coverage_previous_expected_count=2,coverage_expected_count=2)
|
|
target.write_text(json.dumps(value))
|
|
with self.assertRaisesRegex(ValueError,'source bounds'):self.ingest(path)
|
|
with closing(ReadConnection(self.service,self.schema)) as c:
|
|
self.assertEqual({table:c.execute('SELECT count(*) FROM '+table).fetchone()[0] for table in before},before)
|
|
coverage=c.execute('SELECT status,expected_count,imported_count FROM edition_sources WHERE list_url=?',(listing,)).fetchone()
|
|
self.assertEqual(tuple(coverage.values),('partial',2,2))
|
|
self.assertEqual(c.execute('SELECT status FROM ingest_runs ORDER BY id DESC LIMIT 1').fetchone()[0],'failed')
|
|
|
|
def test_incremental_postgres_does_not_overwrite_a_concurrently_changed_parent(self):
|
|
listing='https://example.org/uai/2001'
|
|
rows=[self.row(source='dblp',external=key,status='accepted',year=2001,venue='UAI')
|
|
for key in ('concurrent1','concurrent2','concurrent3')]
|
|
first=self.file('initial',rows[:2]);target=first.with_suffix('.manifest.json')
|
|
value=json.loads(target.read_text());value.update(list_url=listing,status='partial');target.write_text(json.dumps(value))
|
|
self.ingest(first)
|
|
path=self.file('racing',[rows[2]]);target=path.with_suffix('.manifest.json')
|
|
value=json.loads(target.read_text());value.update(list_url=listing,status='partial',coverage_mode='incremental',
|
|
coverage_previous_count=2,coverage_previous_expected_count=2,coverage_expected_count=4)
|
|
target.write_text(json.dumps(value))
|
|
original=Store.record
|
|
def change_parent(store,*args,**kwargs):
|
|
result=original(store,*args,**kwargs)
|
|
# A separate committed SQL writer need not honor the native
|
|
# advisory lock. Its update must survive our rejected file.
|
|
with connect(self.service) as other:
|
|
other.execute('SET search_path TO '+quote(self.schema)+',pg_catalog')
|
|
other.execute('UPDATE edition_sources SET expected_count=99 WHERE list_url=%s',(listing,))
|
|
return result
|
|
with patch.object(Store,'record',change_parent):
|
|
with self.assertRaisesRegex(ValueError,'parent changed during import'):self.ingest(path)
|
|
with closing(ReadConnection(self.service,self.schema)) as c:
|
|
coverage=c.execute('SELECT status,expected_count,imported_count FROM edition_sources WHERE list_url=?',(listing,)).fetchone()
|
|
self.assertEqual(tuple(coverage.values),('partial',99,2))
|
|
self.assertEqual(c.execute('SELECT count(*) FROM source_records WHERE external_id=?',('concurrent3',)).fetchone()[0],0)
|
|
self.assertEqual(c.execute('SELECT status FROM ingest_runs ORDER BY id DESC LIMIT 1').fetchone()[0],'failed')
|
|
|
|
def test_native_writes_keep_nul_historical_ids_and_later_acceptance(self):
|
|
store = self.store()
|
|
try:
|
|
paper, record = store.record('public', 'new-forum', 'https://example.org/new', 'New research identity',
|
|
'Left\x00right\x01\x02', [], {'arxiv': '2501.99999'})
|
|
self.assertGreater(paper, 1000)
|
|
store.decision(paper, record, 'ICLR', 2025, 'main', 'new-forum', 'https://example.org/reject', 'rejected', 'Reject')
|
|
_, second = store.record('official', 'new-pub', 'https://example.org/pub', 'Changed title', None, [], {'arxiv': '2501.99999'})
|
|
store.decision(paper, second, 'COLM', 2026, 'main', 'new-pub', 'https://example.org/accept', 'accepted', 'Accept')
|
|
store.conn.commit()
|
|
self.assertEqual(store.conn.execute('SELECT abstract FROM papers WHERE id=?', (paper,)).fetchone()[0], 'Left\x00right\x01\x02')
|
|
ids = store.conn.execute('SELECT s.id,d.id FROM submissions s JOIN decisions d ON d.submission_id=s.id WHERE s.paper_id=?', (paper,)).fetchall()
|
|
self.assertEqual(len(ids), 2)
|
|
self.assertTrue(all(sid > 5000 and did > 6000 for sid, did in ids))
|
|
finally:
|
|
store.conn.close()
|
|
reader = ReadConnection(self.service, self.schema)
|
|
try:
|
|
from paper_atlas.querying import known_acceptances
|
|
self.assertEqual(known_acceptances(reader, paper)[0]['venue'], 'COLM')
|
|
finally:
|
|
reader.close()
|
|
|
|
def test_identity_anchor_keeps_other_arxiv_version_and_ambiguous_file_rolls_back(self):
|
|
title='An identical long research title shared by conference and extended versions'
|
|
authors=['Jane Doe','John Smith']
|
|
store=self.store()
|
|
try:
|
|
first,_=store.record('arxiv','1408.1485','https://arxiv.org/abs/1408.1485',title,'Conference abstract',authors)
|
|
second,_=store.record('arxiv','cs/0307069','https://arxiv.org/abs/cs/0307069',title,'Extended abstract',authors)
|
|
resolved,_=store.record('dblp','known','https://example.org/known',title,None,authors,{'arxiv':'1408.1485'})
|
|
self.assertEqual(resolved,first);self.assertNotEqual(first,second)
|
|
store.conn.commit()
|
|
self.assertEqual(store.conn.execute('SELECT count(*) FROM papers').fetchone()[0],3)
|
|
self.assertEqual(store.conn.execute('SELECT count(*) FROM paper_merges').fetchone()[0],1) # migration fixture only
|
|
self.assertEqual(store.conn.execute('SELECT count(*) FROM match_candidates').fetchone()[0],1)
|
|
finally:store.conn.close()
|
|
rows=[self.row('official','earlier',status='accepted',venue='UAI',year=2001),
|
|
dict(self.row('official','ambiguous',status='accepted',venue='UAI',year=2001),
|
|
title=title,abstract=None,authors=authors,identifiers={})]
|
|
with self.assertRaisesRegex(ValueError,'Ambiguous paper identity'):self.ingest(self.file('ambiguous-identity',rows))
|
|
store=self.store()
|
|
try:
|
|
self.assertEqual(store.conn.execute('SELECT count(*) FROM papers').fetchone()[0],3)
|
|
self.assertEqual(store.conn.execute('SELECT count(*) FROM source_records').fetchone()[0],4)
|
|
self.assertEqual(store.conn.execute('SELECT count(*) FROM submissions').fetchone()[0],1)
|
|
self.assertEqual(store.conn.execute('SELECT count(*) FROM paper_merges').fetchone()[0],1)
|
|
self.assertEqual(store.conn.execute('SELECT status FROM ingest_runs ORDER BY id DESC LIMIT 1').fetchone()[0],'failed')
|
|
self.assertEqual(store.conn.execute("SELECT count(*) FROM source_records WHERE external_id IN (?,?)",('earlier','ambiguous')).fetchone()[0],0)
|
|
finally:store.conn.close()
|
|
|
|
def test_file_import_is_idempotent_and_refreshes_only_own_tags(self):
|
|
row = self.row(); row['tags'] = [{'namespace': 'openreview_snapshot:area', 'value': 'theory'}]
|
|
path = self.file('reject', [row]); self.ingest(path)
|
|
store = self.store()
|
|
try:
|
|
paper = store.conn.execute("SELECT paper_id FROM identifiers WHERE scheme='arxiv' AND value='2501.99999'").fetchone()[0]
|
|
store.conn.execute("INSERT INTO tags(namespace,value) VALUES('custom:topic','manual') ON CONFLICT DO NOTHING")
|
|
tag = store.conn.execute("SELECT id FROM tags WHERE namespace='custom:topic'").fetchone()[0]
|
|
store.conn.execute('INSERT INTO paper_tags VALUES(?,?,NULL,?)', (paper, tag, 'manual'))
|
|
store.conn.commit()
|
|
finally:
|
|
store.conn.close()
|
|
row['tags'][0]['value'] = 'learning'; self.ingest(self.file('reject', [row]))
|
|
self.ingest(self.file('accept', [self.row('official', 'published', 'accepted', 2026, 'COLM')]))
|
|
store = self.store()
|
|
try:
|
|
self.assertEqual(store.conn.execute('SELECT count(*) FROM papers').fetchone()[0], 2)
|
|
self.assertEqual(store.conn.execute('SELECT count(*) FROM submissions WHERE paper_id=?', (paper,)).fetchone()[0], 2)
|
|
self.assertEqual(store.conn.execute('SELECT count(*) FROM decisions d JOIN submissions s ON s.id=d.submission_id WHERE s.paper_id=?', (paper,)).fetchone()[0], 2)
|
|
tags = set(store.conn.execute('SELECT t.namespace,t.value FROM paper_tags p JOIN tags t ON t.id=p.tag_id WHERE p.paper_id=?', (paper,)))
|
|
self.assertEqual(tags, {('custom:topic', 'manual'), ('openreview_snapshot:area', 'learning')})
|
|
finally:
|
|
store.conn.close()
|
|
|
|
def test_bad_manifest_rolls_back_whole_file_and_keeps_failed_run(self):
|
|
with self.assertRaisesRegex(ValueError, 'Manifest count'):
|
|
self.ingest(self.file('bad', [self.row()], count=2))
|
|
store = self.store()
|
|
try:
|
|
self.assertEqual(store.conn.execute('SELECT count(*) FROM papers').fetchone()[0], 1)
|
|
self.assertEqual(store.conn.execute('SELECT count(*) FROM source_records').fetchone()[0], 1)
|
|
self.assertEqual(store.conn.execute('SELECT status,records_seen,records_written FROM ingest_runs').fetchone(), ('failed', 1, 0))
|
|
finally:
|
|
store.conn.close()
|
|
|
|
def test_changed_file_hash_rolls_back_data_before_commit(self):
|
|
path = self.file('changed', [self.row()])
|
|
manifest = json.loads(path.with_suffix('.manifest.json').read_text())
|
|
manifest['file_sha256'] = '0' * 64
|
|
path.with_suffix('.manifest.json').write_text(json.dumps(manifest))
|
|
with self.assertRaisesRegex(ValueError, 'hash differs'):
|
|
self.ingest(path)
|
|
store = self.store()
|
|
try:
|
|
self.assertEqual(store.conn.execute('SELECT count(*) FROM papers').fetchone()[0], 1)
|
|
self.assertEqual(store.conn.execute('SELECT count(*) FROM source_records').fetchone()[0], 1)
|
|
self.assertEqual(store.conn.execute('SELECT status,records_written FROM ingest_runs').fetchone(), ('failed', 0))
|
|
finally:
|
|
store.conn.close()
|
|
|
|
def test_writer_lock_survives_commit_and_rollback_but_readers_continue(self):
|
|
first = self.store()
|
|
try:
|
|
first.conn.commit()
|
|
with self.assertRaisesRegex(RuntimeError, 'Another writer'):
|
|
self.store()
|
|
first.conn.rollback()
|
|
with self.assertRaisesRegex(RuntimeError, 'Another writer'):
|
|
self.store()
|
|
reader = ReadConnection(self.service, self.schema)
|
|
try:
|
|
self.assertEqual(reader.execute('SELECT count(*) FROM papers').fetchone()[0], 1)
|
|
finally:
|
|
reader.close()
|
|
finally:
|
|
first.conn.close()
|
|
reopened = self.store(); reopened.conn.close()
|
|
|
|
def test_unpublished_schema_is_rejected_without_bootstrapping_tables(self):
|
|
staging = self.schema + '_unfinished'
|
|
with connect(self.service) as pg:
|
|
pg.execute('CREATE SCHEMA ' + quote(staging))
|
|
with self.assertRaisesRegex(RuntimeError, 'published dataset'):
|
|
open_store(pg_service=self.service, pg_schema=staging)
|
|
with connect(self.service) as pg:
|
|
self.assertEqual(pg.execute('SELECT count(*) FROM information_schema.tables WHERE table_schema=%s', (staging,)).fetchone()[0], 0)
|
|
|
|
def test_negative_preflight_uses_current_pg_attempt_and_blocks_stale_import(self):
|
|
import hashlib
|
|
from paper_atlas.collectors.public_openreview import SOURCE, REVISION
|
|
from paper_atlas.audits.public_openreview import audit
|
|
row = self.row(SOURCE, 'negativeForum')
|
|
row['identifiers']['openreview'] = 'negativeForum'
|
|
row['metadata'] = {'evidence_kind': 'curated_public_status_assertion',
|
|
'source_revision': REVISION, 'original_decision_signature_verified': False}
|
|
path = self.file('negative', [row])
|
|
manifest = json.loads(path.with_suffix('.manifest.json').read_text())
|
|
manifest.update(file=str(path), file_sha256=hashlib.sha256(path.read_bytes()).hexdigest())
|
|
path.with_suffix('.manifest.json').write_text(json.dumps(manifest))
|
|
inventory = self.root / 'negative-inventory.json'
|
|
inventory.write_text(json.dumps({'source': SOURCE, 'revision': REVISION,
|
|
'normalized_records': 1, 'files': [manifest]}))
|
|
# An acceptance in another conference does not block this rejection.
|
|
self.ingest(self.file('other-conference', [self.row('official', 'acceptedElsewhere', 'accepted', 2026, 'COLM')]))
|
|
def preflight(name):
|
|
with redirect_stdout(io.StringIO()):
|
|
return audit(None, inventory, self.root / (name + '.json'), pg_service=self.service, pg_schema=self.schema)
|
|
first = preflight('before')
|
|
self.assertEqual(first['status'], 'ready_for_serial_import')
|
|
self.assertEqual(first['engine'], 'postgresql')
|
|
self.assertEqual(first['schema'], self.schema)
|
|
self.ingest(path)
|
|
self.assertEqual(preflight('existing')['existing_same_edition_attempts'], 1)
|
|
store = self.store()
|
|
try:
|
|
paper, record = store.conn.execute('''SELECT r.paper_id,r.id FROM source_records r
|
|
JOIN sources s ON s.id=r.source_id WHERE s.name=? AND r.external_id=?''', (SOURCE, row['external_id'])).fetchone()
|
|
store.decision(paper, record, 'ICLR', 2025, 'main', row['external_id'],
|
|
'https://example.org/formal', 'accepted', 'Accept', observed_at='2027-01-01T00:00:00Z')
|
|
# Equal observation time, but an explicitly earlier decision date.
|
|
# NULL dates must fall back to observed_at, not the empty string.
|
|
store.decision(paper, record, 'ICLR', 2025, 'main', row['external_id'],
|
|
'https://example.org/older', 'rejected', 'Reject', observed_at='2027-01-01T00:00:00Z',
|
|
decision_date='2026-12-31T00:00:00Z')
|
|
store.conn.commit()
|
|
finally:
|
|
store.conn.close()
|
|
blocked = preflight('blocked')
|
|
self.assertEqual(blocked['status'], 'blocked')
|
|
self.assertEqual(blocked['conflicts'][0]['existing_status'], 'accepted')
|
|
self.assertEqual(blocked['files'], [])
|
|
# The earlier ready report cannot authorize overwriting current status.
|
|
with self.assertRaisesRegex(ValueError, 'conflicts with current'):
|
|
self.ingest(path)
|
|
reader = open_read_connection(pg_service=self.service, pg_schema=self.schema)
|
|
try:
|
|
self.assertEqual(reader.execute('SHOW transaction_read_only').fetchone()[0], 'on')
|
|
from paper_atlas.db.submissions import latest_status
|
|
sid = reader.execute("SELECT submission_id FROM submission_identifiers WHERE scheme='openreview' AND value=?", ('negativeForum',)).fetchone()[0]
|
|
self.assertEqual(latest_status(reader, sid), 'accepted')
|
|
finally:
|
|
reader.close()
|
|
|
|
def test_coverage_matches_sqlite_with_history_aliases_and_missing_source_abstracts(self):
|
|
from paper_atlas.audits.coverage import report
|
|
from tests.test_coverage_report import add_coverage_evidence
|
|
registry = self.root / 'venues.json'
|
|
registry.write_text(json.dumps({'scope_note': 'Candidate scope', 'venues': [
|
|
{'acronym': 'coling', 'field': 'language'}, {'acronym': 'AAAI', 'field': 'general_ai'}]}))
|
|
sample = Store(self.root / 'seed.sqlite')
|
|
try:
|
|
add_coverage_evidence(sample)
|
|
finally:
|
|
sample.conn.close()
|
|
store = self.store()
|
|
try:
|
|
add_coverage_evidence(store)
|
|
finally:
|
|
store.conn.close()
|
|
sqlite_report = report(self.root / 'seed.sqlite', registry)
|
|
observed = []
|
|
def reader(*args, **kwargs):
|
|
connection = open_read_connection(*args, **kwargs)
|
|
observed.append((connection.execute('SHOW transaction_read_only').fetchone()[0],
|
|
connection.execute('SHOW transaction_isolation').fetchone()[0]))
|
|
return connection
|
|
with patch('paper_atlas.audits.coverage.open_read_connection', side_effect=reader):
|
|
actual = report(None, registry, pg_service=self.service, pg_schema=self.schema)
|
|
self.assertEqual(observed, [('on', 'repeatable read')])
|
|
self.assertEqual(actual['engine'], 'postgresql')
|
|
self.assertEqual(actual['schema'], self.schema)
|
|
self.assertEqual(actual['pg_service'], self.service)
|
|
def content(result):
|
|
values = json.loads(json.dumps(result['venues']))
|
|
for venue in values:
|
|
for edition in venue['editions']:
|
|
for key in ('id', 'venue_id'):
|
|
edition.pop(key)
|
|
return values
|
|
self.assertEqual(content(actual), content(sqlite_report))
|
|
self.assertEqual([r['acronym'] for r in actual['imported_venues_outside_registry']], ['ICLR'])
|
|
edition = actual['venues'][0]['editions'][0]
|
|
self.assertEqual(edition['latest_observed_status_counts']['accepted'], 1)
|
|
self.assertEqual(edition['latest_observed_status_counts']['unknown'], 1)
|
|
self.assertEqual(edition['decision_observation_counts']['rejected'], 1)
|
|
self.assertEqual(edition['papers_without_abstract_text'], 1)
|
|
|
|
def test_backfill_merges_evidence_without_reimporting_or_losing_history(self):
|
|
files = []
|
|
for source, external, title in [('official', 'legacy', 'First bibliographic name'), ('openreview_snapshot', 'forum', 'Renamed submission')]:
|
|
row = self.row(source, external, 'accepted')
|
|
row.update(title=title, authors=[], identifiers={}, evidence_url='https://example.org/same-decision')
|
|
path = self.file(source, [row]); self.ingest(path)
|
|
row['identifiers'] = {'openreview': 'sharedForum'}
|
|
self.file(source, [row]); files.append(path)
|
|
with redirect_stdout(io.StringIO()):
|
|
result = backfill(None, files, pg_service=self.service, pg_schema=self.schema)
|
|
self.assertEqual((result['submissions_before'], result['submissions_after']), (3, 2))
|
|
store = self.store()
|
|
try:
|
|
self.assertEqual(store.conn.execute('SELECT count(*) FROM papers').fetchone()[0], 2)
|
|
self.assertEqual(store.conn.execute('SELECT count(*) FROM decision_sources').fetchone()[0], 3)
|
|
self.assertEqual(store.conn.execute('SELECT count(*) FROM decision_merges').fetchone()[0], 2)
|
|
finally:
|
|
store.conn.close()
|
|
|
|
def test_arxiv_chunk_resume_preserves_committed_ids_and_category_evidence(self):
|
|
archive = self.root / 'arxiv.zip'
|
|
rows = [{'id': f'2501.{i:05d}', 'title': f'Checkpoint paper {i}', 'abstract': 'Metadata abstract',
|
|
'authors': [], 'categories': 'cs.AI'} for i in range(1, 502)]
|
|
with zipfile.ZipFile(archive, 'w') as zipped:
|
|
zipped.writestr('arxiv-metadata-oai-snapshot.json', ''.join(json.dumps(r) + '\n' for r in rows))
|
|
calls = 0
|
|
original = arxiv.refresh_categories
|
|
def interrupt(*args):
|
|
nonlocal calls
|
|
calls += 1
|
|
if calls == 501:
|
|
raise RuntimeError('controlled interruption')
|
|
return original(*args)
|
|
with patch('paper_atlas.ingestion.arxiv.shutil.disk_usage', return_value=SimpleNamespace(free=8 * 1024**3)), \
|
|
patch('paper_atlas.ingestion.arxiv.refresh_categories', side_effect=interrupt), redirect_stdout(io.StringIO()):
|
|
with self.assertRaisesRegex(RuntimeError, 'controlled interruption'):
|
|
arxiv.ingest(None, archive, pg_service=self.service, pg_schema=self.schema)
|
|
store = self.store()
|
|
try:
|
|
self.assertEqual(store.conn.execute('SELECT count(*) FROM papers').fetchone()[0], 501)
|
|
preserved = store.conn.execute("SELECT paper_id FROM identifiers WHERE scheme='arxiv' AND value='2501.00001'").fetchone()[0]
|
|
finally:
|
|
store.conn.close()
|
|
with redirect_stdout(io.StringIO()):
|
|
arxiv.ingest(None, archive, pg_service=self.service, pg_schema=self.schema)
|
|
store = self.store()
|
|
try:
|
|
self.assertEqual(store.conn.execute('SELECT count(*) FROM papers').fetchone()[0], 502)
|
|
self.assertEqual(store.conn.execute("SELECT paper_id FROM identifiers WHERE scheme='arxiv' AND value='2501.00001'").fetchone()[0], preserved)
|
|
self.assertEqual(store.conn.execute('SELECT count(*) FROM paper_tags').fetchone()[0], 501)
|
|
self.assertEqual(store.conn.execute('SELECT status,records_written FROM ingest_runs').fetchone(), ('complete', 501))
|
|
finally:
|
|
store.conn.close()
|