149 lines
7.9 KiB
Python
149 lines
7.9 KiB
Python
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())
|