117 lines
6.1 KiB
Python
117 lines
6.1 KiB
Python
from contextlib import redirect_stdout
|
|
import hashlib
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
|
|
from paper_atlas.ingestion.records import ingest_file
|
|
|
|
|
|
class IncrementalCoverageTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.root = Path(self.tmp.name)
|
|
self.db = self.root / 'papers.sqlite'
|
|
|
|
def file(self, name, external_ids, **options):
|
|
rows = [{'source':'example','external_id':key,'url':'https://example.org/'+key,
|
|
'venue':'Example','year':2026,'track':'main','status':'accepted',
|
|
'title':'Independent research contribution '+key,'abstract':'Known abstract',
|
|
'authors':[], 'identifiers':{'doi':'10.1234/'+key}} for key in external_ids]
|
|
path = self.root / (name+'.jsonl')
|
|
path.write_bytes((''.join(json.dumps(row)+'\n' for row in rows)).encode())
|
|
manifest = {key:rows[0][key] for key in ('source','venue','year','track')}
|
|
manifest.update(list_url='https://example.org/2026', count=len(rows),status='partial',
|
|
file_sha256=hashlib.sha256(path.read_bytes()).hexdigest())
|
|
manifest.update(options)
|
|
path.with_suffix('.manifest.json').write_text(json.dumps(manifest))
|
|
return path
|
|
|
|
def ingest(self, path):
|
|
with redirect_stdout(io.StringIO()):
|
|
ingest_file(self.db,path)
|
|
|
|
def incremental(self, previous=2, prior_expected=2, expected=4):
|
|
return {'coverage_mode':'incremental','coverage_previous_count':previous,
|
|
'coverage_previous_expected_count':prior_expected,'coverage_expected_count':expected}
|
|
|
|
def coverage(self):
|
|
with sqlite3.connect(self.db) as c:
|
|
return c.execute('SELECT status,expected_count,imported_count FROM edition_sources').fetchall()
|
|
|
|
def test_append_and_refresh_count_actual_source_records_without_erasing_earlier_batch(self):
|
|
self.ingest(self.file('first',['a','b']))
|
|
self.ingest(self.file('append',['c'],**self.incremental()))
|
|
self.assertEqual(self.coverage(),[('partial',4,3)])
|
|
self.ingest(self.file('refresh',['c'],**self.incremental(3,4,4)))
|
|
self.assertEqual(self.coverage(),[('partial',4,3)])
|
|
with sqlite3.connect(self.db) as c:
|
|
self.assertEqual(c.execute('SELECT count(*) FROM source_records').fetchone()[0],3)
|
|
self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0],3)
|
|
runs = c.execute('SELECT records_written,checkpoint_json FROM ingest_runs ORDER BY id').fetchall()
|
|
self.assertEqual([r[0] for r in runs],[2,1,1])
|
|
self.assertEqual(json.loads(runs[-1][1])['coverage'],
|
|
{'mode':'incremental','previous_count':3,'expected_count':4,'imported_count':3})
|
|
self.assertEqual(c.execute('PRAGMA foreign_key_check').fetchall(),[])
|
|
|
|
def test_stale_parent_stops_before_source_writes(self):
|
|
self.ingest(self.file('first',['a','b']))
|
|
with self.assertRaisesRegex(ValueError,'parent changed'):
|
|
self.ingest(self.file('stale',['c'],**self.incremental(1,2,4)))
|
|
self.assertEqual(self.coverage(),[('partial',2,2)])
|
|
with sqlite3.connect(self.db) as c:
|
|
self.assertEqual(c.execute("SELECT count(*) FROM source_records WHERE external_id='c'").fetchone()[0],0)
|
|
self.assertEqual(c.execute('SELECT status FROM ingest_runs ORDER BY id DESC LIMIT 1').fetchone()[0],'failed')
|
|
|
|
def test_late_source_bound_failure_rolls_back_new_records_and_coverage(self):
|
|
self.ingest(self.file('first',['a','b']))
|
|
with self.assertRaisesRegex(ValueError,'source bounds'):
|
|
self.ingest(self.file('overflow',['c'],**self.incremental(2,2,2)))
|
|
self.assertEqual(self.coverage(),[('partial',2,2)])
|
|
with sqlite3.connect(self.db) as c:
|
|
for table in ('papers','source_records','submissions','decisions'):
|
|
self.assertEqual(c.execute('SELECT count(*) FROM '+table).fetchone()[0],2)
|
|
|
|
def test_multiple_listings_cannot_be_silently_aggregated(self):
|
|
self.ingest(self.file('one',['a']))
|
|
self.ingest(self.file('two',['b'],list_url='https://example.org/another-list'))
|
|
before=self.coverage()
|
|
with self.assertRaisesRegex(ValueError,'unambiguous source listing'):
|
|
self.ingest(self.file('ambiguous',['c'],**self.incremental(2,1,4)))
|
|
self.assertEqual(self.coverage(),before)
|
|
|
|
def test_untracked_source_rows_require_reconciliation(self):
|
|
self.ingest(self.file('first',['a','b']))
|
|
with sqlite3.connect(self.db) as c:
|
|
c.execute('UPDATE edition_sources SET imported_count=1')
|
|
with self.assertRaisesRegex(ValueError,'actual source records'):
|
|
self.ingest(self.file('stale-summary',['c'],**self.incremental(1,2,4)))
|
|
self.assertEqual(self.coverage(),[('partial',2,1)])
|
|
|
|
def test_first_partial_chunk_can_declare_an_independently_known_source_total(self):
|
|
self.ingest(self.file('first',['a'],**self.incremental(0,None,4)))
|
|
self.assertEqual(self.coverage(),[('partial',4,1)])
|
|
|
|
def test_complete_listing_is_not_demoted_by_a_partial_chunk(self):
|
|
self.ingest(self.file('complete',['a','b'],status='complete'))
|
|
with self.assertRaisesRegex(ValueError,'not partial'):
|
|
self.ingest(self.file('append',['c'],**self.incremental()))
|
|
self.assertEqual(self.coverage(),[('complete',2,2)])
|
|
|
|
def test_invalid_mode_counts_and_missing_scope_fail_before_database_creation(self):
|
|
for options in ({'coverage_mode':'unknown'}, self.incremental(True,2,4),
|
|
self.incremental(2,5,4),self.incremental(-1,2,4),
|
|
{'coverage_mode':'incremental'},
|
|
{**self.incremental(),'status':'complete'}):
|
|
with self.subTest(options=options):
|
|
with self.assertRaises(ValueError):self.ingest(self.file('invalid',['a'],**options))
|
|
self.assertFalse(self.db.exists())
|
|
|
|
|
|
if __name__=='__main__':
|
|
unittest.main()
|