45 lines
2.4 KiB
Python
45 lines
2.4 KiB
Python
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
import json
|
|
from pathlib import Path
|
|
import sqlite3
|
|
from contextlib import closing
|
|
import tempfile
|
|
import unittest
|
|
import zipfile
|
|
|
|
from paper_atlas.audits.arxiv import audit,MEMBER
|
|
|
|
|
|
class ArxivSourceAuditTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp=tempfile.TemporaryDirectory();self.addCleanup(self.temp.cleanup);self.root=Path(self.temp.name)
|
|
self.archive=self.root/'metadata.zip';self.out=self.root/'audit.json'
|
|
disk=patch('paper_atlas.audits.arxiv.shutil.disk_usage',return_value=SimpleNamespace(free=16*1024**3));disk.start();self.addCleanup(disk.stop)
|
|
|
|
def write(self,body):
|
|
with zipfile.ZipFile(self.archive,'w',compression=zipfile.ZIP_DEFLATED) as z:z.writestr(MEMBER,body)
|
|
|
|
def test_repeated_ids_and_payload_changes_are_distinct_from_unique_papers(self):
|
|
one={'id':'2601.00001','title':'A paper','abstract':'First text'};updated={**one,'abstract':'Revised text'}
|
|
two={'id':'2601.00002','title':'Another paper','abstract':''}
|
|
self.write('\n'.join(json.dumps(row) for row in (one,updated,updated,two)))
|
|
result=audit(self.archive,self.out)
|
|
self.assertEqual((result['raw_rows'],result['unique_external_ids'],result['extra_rows_for_repeated_ids']),(4,2,2))
|
|
self.assertTrue(result['zip_eof_crc_verified']);self.assertEqual(result['missing_abstract_rows'],1)
|
|
duplicate=result['duplicate_ids'][0];self.assertEqual(duplicate['occurrences'],3)
|
|
self.assertTrue(duplicate['has_payload_variants']);self.assertEqual([v['occurrences'] for v in duplicate['variants']],[1,2])
|
|
with closing(sqlite3.connect(result['id_index'])) as c:self.assertEqual(c.execute('SELECT count(*) FROM raw_ids').fetchone()[0],2)
|
|
with self.assertRaisesRegex(ValueError,'output exists'):audit(self.archive,self.out)
|
|
|
|
def test_invalid_or_unverified_input_never_publishes_audit_outputs(self):
|
|
self.write('{"id":"2601.00001","title":"Valid"}\nnot JSON')
|
|
with self.assertRaises(json.JSONDecodeError):audit(self.archive,self.out)
|
|
self.assertFalse(self.out.exists());self.assertFalse(self.out.with_suffix('.ids.sqlite').exists())
|
|
self.write('{"id":"2601.00001","title":"Valid"}')
|
|
with self.assertRaisesRegex(ValueError,'checksum mismatch'):audit(self.archive,self.out,'0'*64)
|
|
self.assertFalse(self.out.exists())
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|