import gzip import hashlib import json import os from pathlib import Path import tempfile import unittest from unittest.mock import patch from paper_atlas.audits.source_identity import audit, classify, load_inputs from paper_atlas.postgres.client import ReadConnection def record(**updates): row = dict(source='reviewed_program', external_id='2026/forum-a', title='A sufficiently specific scientific paper title', authors=['Jane Doe', 'John Smith'], venue='NeurIPS', year=2026, track='main', status='accepted', identifiers={}, source_record_hits=[], identifier_hits=[], title_candidate_ids=[]) row.update(updates) return row def write_input(root, rows): path = root / 'source.jsonl' raw = b''.join((json.dumps(row) + '\n').encode() for row in rows) path.write_bytes(raw) manifest = {k: rows[0][k] for k in ('source', 'venue', 'year', 'track')} manifest.update(count=len(rows), file_sha256=hashlib.sha256(raw).hexdigest(), file='/ignored/untrusted/location.jsonl', status='partial') path.with_suffix('.manifest.json').write_text(json.dumps(manifest)) return path class CandidateTests(unittest.TestCase): def test_duplicate_names_do_not_equal_distinct_author_multiset(self): row = record(authors=['Jane Doe', 'Jane Doe'], title_candidate_ids=[7]) result = classify(row, {7: {'title': row['title'], 'authors': ['Jane Doe']}}) self.assertEqual(result['kind'], 'partial_author_overlap_requires_review') self.assertIn('duplicate_normalized_author_names', result['flags']) self.assertFalse(result['comparisons'][0]['nonempty_author_multiset_equal']) def test_arxiv_author_representation_and_order_match_but_never_approve(self): row = record(title_candidate_ids=[7]) result = classify(row, {7: {'title': row['title'], 'authors': [['Smith', 'John', ''], ['Doe', 'Jane', '']]}}) self.assertEqual(result['kind'], 'one_exact_title_author_multiset_requires_review') self.assertEqual(result['decision'], 'not_approved_for_import') def test_binding_conflict_overrides_existing_source(self): row = record(source_record_hits=[{'paper_id': 7}], identifier_hits=[{'paper_id': 8}]) entities = {pid: {'title': row['title'], 'authors': row['authors']} for pid in (7, 8)} self.assertEqual(classify(row, entities)['kind'], 'conflicting_existing_bindings_hold') def test_wrong_doi_binding_retains_author_title_disagreement(self): row = record(identifier_hits=[{'paper_id': 7}]) result = classify(row, {7: {'title': 'Completely different work', 'authors': ['Somebody Else']}}) self.assertEqual(result['kind'], 'existing_identifier_binding_requires_review') self.assertFalse(result['comparisons'][0]['normalized_title_equal']) self.assertEqual(result['comparisons'][0]['author_name_overlap'], []) def test_no_hit_is_not_novelty_and_missing_authors_never_agree(self): self.assertEqual(classify(record(), {})['kind'], 'no_exact_match_not_proof_of_novelty') row = record(authors=[], title_candidate_ids=[7]) result = classify(row, {7: {'title': row['title'], 'authors': []}}) self.assertEqual(result['kind'], 'no_author_agreement_hold') def test_title_candidate_cap_is_explicit(self): row = record(title_candidate_ids=list(range(21))) entities = {pid: {'title': row['title'], 'authors': row['authors']} for pid in range(21)} self.assertTrue(classify(row, entities)['title_candidates_truncated']) self.assertEqual(classify(row, entities)['kind'], 'title_candidate_limit_reached_hold') class InputTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) def test_raw_line_hash_and_manifest_path_not_followed(self): path = write_input(self.root, [record()]) rows, inputs = load_inputs([path]) self.assertEqual(rows[0]['raw_line_sha256'], hashlib.sha256(path.read_bytes()).hexdigest()) self.assertEqual(inputs[0]['count'], 1) def test_hash_tamper_rejected_before_database_connection(self): path = write_input(self.root, [record()]) path.write_bytes(path.read_bytes().replace(b'Jane Doe', b'Jane Roe')) with patch('paper_atlas.audits.source_identity.ReadConnection') as connection: with self.assertRaisesRegex(ValueError, 'hash mismatch'): audit([path], self.root / 'out', pg_service='test', pg_schema='research') connection.assert_not_called() def test_manifest_scope_and_count_tamper_and_duplicate_sources(self): path = write_input(self.root, [record()]) manifest_path = path.with_suffix('.manifest.json') manifest = json.loads(manifest_path.read_text()) for key, value in [('year', 2025), ('count', 2)]: altered = dict(manifest, **{key: value}) manifest_path.write_text(json.dumps(altered)) with self.assertRaises(ValueError): load_inputs([path]) manifest_path.write_text(json.dumps(manifest)) with self.assertRaisesRegex(ValueError, 'Duplicate source'): load_inputs([path, path]) def test_existing_output_never_replayed_and_failure_not_completion(self): path = write_input(self.root, [record()]) out = self.root / 'out' with patch('paper_atlas.audits.source_identity.ReadConnection', side_effect=RuntimeError('unavailable')): with self.assertRaises(RuntimeError): audit([path], out, pg_service='test', pg_schema='research') self.assertTrue((out / 'failed.json').exists()) self.assertFalse((out / 'postgresql-preflight.json').exists()) with self.assertRaises(FileExistsError): audit([path], out, pg_service='test', pg_schema='research') @unittest.skipUnless(os.environ.get('PAPER_ATLAS_TEST_PG_SERVICE'), 'Requires the authorized remote PostgreSQL service') class ReadOnlyIntegrationTests(unittest.TestCase): def test_existing_binding_full_evidence_and_no_import_approval(self): service = os.environ['PAPER_ATLAS_TEST_PG_SERVICE'] connection = ReadConnection(service, 'research') try: self.assertEqual(connection.execute('SHOW transaction_read_only').fetchone()[0], 'on') hit = connection.execute("SELECT value,paper_id FROM identifiers WHERE scheme='arxiv' ORDER BY value LIMIT 1").fetchone() self.assertIsNotNone(hit) paper = connection.execute('SELECT title,authors_json FROM papers WHERE id=?', (hit['paper_id'],)).fetchone() finally: connection.close() with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) path = write_input(root, [record(title=paper['title'], authors=json.loads(paper['authors_json']), identifiers={'arxiv': hit['value']})]) report = audit([path], root / 'out', pg_service=service, pg_schema='research') self.assertTrue(report['transaction_read_only']) self.assertFalse(report['database_written']) self.assertEqual(report['approved_for_import'], 0) row, = report['entries'] self.assertIn(hit['paper_id'], [r['paper_id'] for r in row['identifier_hits']]) shard = report['entities'][str(hit['paper_id'])] raw = (root / 'out' / shard['file']).read_bytes() self.assertEqual(hashlib.sha256(raw).hexdigest(), shard['sha256']) packet = json.loads(gzip.decompress(raw)) self.assertEqual(packet['paper']['id'], hit['paper_id']) self.assertEqual(packet['paper']['title'], paper['title']) self.assertEqual(packet['paper']['authors_json'], paper['authors_json']) self.assertIn('attempts', packet) self.assertTrue((root / 'out/postgresql-preflight.json').is_file())