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()