142 lines
8.2 KiB
Python
142 lines
8.2 KiB
Python
import gzip
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from paper_atlas.collectors.dblp import catalogue
|
|
from paper_atlas.collectors.dblp_recovery import recover
|
|
from paper_atlas.collectors.dblp_scope import scope
|
|
from paper_atlas.ingestion.records import ingest_file
|
|
|
|
|
|
class DblpRecoveryTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.root = Path(self.tmp.name)
|
|
self.dtd = self.root / 'dblp.dtd'
|
|
self.dtd.write_bytes(b'<!ENTITY eacute "é">')
|
|
self.archive = self.root / 'dblp.xml.gz'
|
|
# These include a keynote and a workshop, neither a main-paper inference.
|
|
body = '''<proceedings key="conf/ifaamas/2025"><title>AAMAS 2025</title><year>2025</year></proceedings>
|
|
<inproceedings key="conf/ifaamas/keynote"><author>René</author><title>Keynote: Agents</title><year>2025</year><booktitle>AAMAS</booktitle><crossref>conf/ifaamas/2025</crossref></inproceedings>
|
|
<inproceedings key="conf/ifaamas/paper"><author>Alice Doe</author><title>Research <i>agents</i></title><year>2025</year><booktitle>AAMAS</booktitle><crossref>conf/ifaamas/2025</crossref><ee>https://dl.acm.org/doi/10.5555/3709347.123</ee></inproceedings>
|
|
<proceedings key="conf/ifaamas/2025w"><title>AAMAS Workshop</title><year>2025</year></proceedings>
|
|
<inproceedings key="conf/ifaamas/workshop"><title>Workshop paper</title><year>2025</year><booktitle>AAMAS Workshop</booktitle><crossref>conf/ifaamas/2025w</crossref></inproceedings>
|
|
<inproceedings key="conf/unrelated/other"><title>Other family</title><year>2025</year></inproceedings>'''
|
|
with gzip.open(self.archive, 'wb') as stream:
|
|
stream.write(('<?xml version="1.0"?><!DOCTYPE dblp SYSTEM "dblp.dtd"><dblp>'
|
|
+ body + '</dblp>').encode())
|
|
self.sha = hashlib.sha256(self.archive.read_bytes()).hexdigest()
|
|
self.md5 = hashlib.md5(self.dtd.read_bytes()).hexdigest()
|
|
self.manifest = catalogue(self.archive, self.dtd, self.root / 'catalogue',
|
|
self.sha, self.md5)
|
|
self.catalogue = Path(self.manifest['catalogue'])
|
|
self.manifest_path = self.catalogue.with_name('dblp-ai-candidates.manifest.json')
|
|
|
|
def run_recovery(self, **kwargs):
|
|
options = dict(expected_sha=self.sha, expected_dtd_md5=self.md5)
|
|
options.update(kwargs)
|
|
return recover(self.archive, self.dtd, self.catalogue, self.root / 'recovered',
|
|
['ifaamas'], **options)
|
|
|
|
def test_fresh_discovery_and_recovery_from_old_candidate_omission(self):
|
|
self.assertEqual(self.manifest['candidate_family_counts']['ifaamas'], 3)
|
|
# Simulate an older filter without altering its complete catalogue.
|
|
manifest = dict(self.manifest)
|
|
manifest['candidate_family_counts'] = {}
|
|
self.manifest_path.write_text(json.dumps(manifest))
|
|
candidate = Path(manifest['file'])
|
|
candidate.write_text('old incomplete candidate file\n')
|
|
before = {p: p.read_bytes() for p in [self.archive, self.dtd, self.catalogue,
|
|
self.manifest_path, candidate]}
|
|
with patch('urllib.request.urlopen', side_effect=AssertionError('No network')):
|
|
report = self.run_recovery()
|
|
rows = [json.loads(line) for line in Path(report['file']).read_text().splitlines()]
|
|
self.assertEqual(report['recovered_publications'], 3)
|
|
self.assertEqual(report['recovered_volumes'], 2)
|
|
self.assertEqual(report['recovered_records'], 5)
|
|
self.assertEqual(report['records_scanned'], 6)
|
|
self.assertEqual(report['old_candidate_family_counts'], {'ifaamas': 0})
|
|
self.assertFalse(report['database_written'])
|
|
self.assertEqual(next(r for r in rows if r['key'].endswith('/keynote'))['fields']['author'], ['René'])
|
|
self.assertTrue(any(r['key'].endswith('/workshop') for r in rows))
|
|
self.assertTrue(all(not ({'source', 'status', 'venue', 'track'} & r.keys()) for r in rows))
|
|
publication = next(r for r in rows if r['key'].endswith('/paper'))
|
|
volume = next(r for r in rows if r['key'] == 'conf/ifaamas/2025')
|
|
self.assertEqual(scope(publication, volume), (None, 'outside_target_family'))
|
|
self.assertEqual({p: p.read_bytes() for p in before}, before)
|
|
self.assertEqual(hashlib.sha256(Path(report['file']).read_bytes()).hexdigest(),
|
|
report['file_sha256'])
|
|
# Raw recovery files cannot accidentally enter the normalized writer.
|
|
with patch('paper_atlas.ingestion.records.open_store') as writer:
|
|
with self.assertRaisesRegex(ValueError, 'source manifest'):
|
|
ingest_file(self.root / 'never.sqlite', report['file'])
|
|
writer.assert_not_called()
|
|
self.assertFalse((self.root / 'never.sqlite').exists())
|
|
|
|
def test_checksum_mismatch_does_not_publish(self):
|
|
for options in [{'expected_sha': '0' * 64}, {'expected_dtd_md5': '0' * 32}]:
|
|
with self.subTest(options=options), self.assertRaisesRegex(ValueError, 'checksum'):
|
|
self.run_recovery(**options)
|
|
self.assertFalse((self.root / 'recovered').exists())
|
|
|
|
def test_wrong_catalogue_snapshot_and_incomplete_totals_are_rejected(self):
|
|
for changes in [{'snapshot_sha256': '0' * 64}, {'dtd_md5': '0' * 32},
|
|
{'records_scanned': 99}, {'record_types': {'proceedings': 2}}]:
|
|
manifest = {**self.manifest, **changes}
|
|
self.manifest_path.write_text(json.dumps(manifest))
|
|
with self.subTest(changes=changes), self.assertRaises(ValueError):
|
|
self.run_recovery()
|
|
self.assertFalse((self.root / 'recovered').exists())
|
|
|
|
def test_changed_family_count_does_not_publish_partial_result(self):
|
|
with sqlite3.connect(self.catalogue) as c:
|
|
c.execute("UPDATE family_counts SET count=count+1 WHERE family='ifaamas' AND record_type='inproceedings'")
|
|
with self.assertRaisesRegex(ValueError, 'family totals'):
|
|
self.run_recovery()
|
|
self.assertFalse((self.root / 'recovered').exists())
|
|
|
|
def test_changed_volume_metadata_is_detected(self):
|
|
with sqlite3.connect(self.catalogue) as c:
|
|
row = json.loads(c.execute("SELECT metadata_json FROM proceedings WHERE key='conf/ifaamas/2025'").fetchone()[0])
|
|
row['fields']['title'] = ['Wrong volume']
|
|
c.execute("UPDATE proceedings SET metadata_json=? WHERE key='conf/ifaamas/2025'", (json.dumps(row),))
|
|
with self.assertRaisesRegex(ValueError, 'proceedings differs'):
|
|
self.run_recovery()
|
|
self.assertFalse((self.root / 'recovered').exists())
|
|
|
|
def test_truncated_gzip_never_publishes(self):
|
|
self.archive.write_bytes(self.archive.read_bytes()[:-8])
|
|
revised_sha = hashlib.sha256(self.archive.read_bytes()).hexdigest()
|
|
self.manifest_path.write_text(json.dumps({**self.manifest, 'snapshot_sha256': revised_sha}))
|
|
with self.assertRaises((EOFError, OSError)):
|
|
self.run_recovery(expected_sha=revised_sha)
|
|
self.assertFalse((self.root / 'recovered').exists())
|
|
|
|
def test_existing_output_is_preserved_and_unknown_family_is_rejected(self):
|
|
out = self.root / 'recovered'
|
|
out.mkdir()
|
|
sentinel = out / 'keep'
|
|
sentinel.write_text('existing')
|
|
with self.assertRaisesRegex(ValueError, 'already exist'):
|
|
self.run_recovery()
|
|
self.assertEqual(sentinel.read_text(), 'existing')
|
|
for names in [[], ['../ifaamas'], ['not-a-supported-family'], [None]]:
|
|
with self.subTest(names=names), self.assertRaises(ValueError):
|
|
recover(self.archive, self.dtd, self.catalogue, self.root / 'other', names)
|
|
|
|
def test_requested_family_missing_from_catalogue_is_explicit(self):
|
|
with self.assertRaisesRegex(ValueError, 'absent from complete catalogue'):
|
|
recover(self.archive, self.dtd, self.catalogue, self.root / 'recovered',
|
|
['ifaamas', 'icml'], expected_sha=self.sha, expected_dtd_md5=self.md5)
|
|
self.assertFalse((self.root / 'recovered').exists())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|