69 lines
4.7 KiB
Python
69 lines
4.7 KiB
Python
import gzip
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
|
|
from paper_atlas.collectors.dblp import catalogue,records
|
|
|
|
|
|
class DblpCatalogueTests(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 "é">')
|
|
|
|
def archive(self,body,doctype='<!DOCTYPE dblp SYSTEM "dblp.dtd">'):
|
|
path=self.root/'dblp.xml.gz'
|
|
with gzip.open(path,'wb') as stream:stream.write(('<?xml version="1.0"?>'+doctype+'<dblp>'+body+'</dblp>').encode())
|
|
return path
|
|
|
|
def test_streamed_entities_nested_titles_and_candidate_scope_remain_explicit(self):
|
|
path=self.archive('<proceedings key="conf/aaai/2025"><title>AAAI Workshop proceedings</title><year>2025</year></proceedings><inproceedings key="conf/aaai/paper"><author>René</author><title>Learn <i>deep</i> models</title><year>2025</year><booktitle>AAAI Workshop</booktitle><crossref>conf/aaai/2025</crossref><ee>https://doi.org/10.1234/example</ee></inproceedings><inproceedings key="conf/unrelated/paper"><title>Other</title><year>2020</year></inproceedings>')
|
|
manifest=catalogue(path,self.dtd,self.root/'out',hashlib.sha256(path.read_bytes()).hexdigest(),hashlib.md5(self.dtd.read_bytes()).hexdigest())
|
|
self.assertEqual(manifest['records_scanned'],3);self.assertEqual(manifest['candidate_records'],1)
|
|
row=json.loads(Path(manifest['file']).read_text())
|
|
self.assertEqual(row['fields']['author'],['René']);self.assertEqual(row['fields']['title'],['Learn deep models'])
|
|
self.assertNotIn('status',row);self.assertNotIn('venue',row)
|
|
with sqlite3.connect(manifest['catalogue']) as connection:
|
|
self.assertEqual(connection.execute('SELECT sum(count) FROM volume_counts').fetchone()[0],2)
|
|
self.assertEqual(connection.execute('SELECT count(*) FROM proceedings').fetchone()[0],1)
|
|
|
|
def test_remote_external_dtd_is_rejected_without_fetching(self):
|
|
path=self.archive('',doctype='<!DOCTYPE dblp SYSTEM "https://example.org/evil.dtd">')
|
|
with self.assertRaisesRegex(ValueError,'pinned external'):list(records(path,self.dtd.read_bytes()))
|
|
|
|
def test_real_monthly_dtd_name_and_latin1_encoding(self):
|
|
path=self.root/'monthly.xml.gz'
|
|
xml='<?xml version="1.0" encoding="ISO-8859-1"?><!DOCTYPE dblp SYSTEM "dblp-2023-06-28.dtd"><dblp><inproceedings key="conf/aaai/one"><author>René</author><title>Metadata</title></inproceedings></dblp>'
|
|
with gzip.open(path,'wb') as stream:stream.write(xml.encode('latin1'))
|
|
self.assertEqual(list(records(path,self.dtd.read_bytes()))[0]['fields']['author'],['René'])
|
|
|
|
def test_multiple_volume_fields_remain_reviewable_without_double_counting(self):
|
|
path=self.archive('<inproceedings key="conf/aaai/joint"><title>Joint publication</title><year>2024</year><booktitle>First meeting</booktitle><booktitle>Second meeting</booktitle><crossref>conf/aaai/joint2024</crossref></inproceedings>')
|
|
manifest=catalogue(path,self.dtd,self.root/'out',hashlib.sha256(path.read_bytes()).hexdigest(),hashlib.md5(self.dtd.read_bytes()).hexdigest())
|
|
with sqlite3.connect(manifest['catalogue']) as connection:
|
|
raw=json.loads(connection.execute('SELECT metadata_json FROM volume_anomalies').fetchone()[0])
|
|
self.assertEqual(raw['fields']['booktitle'],['First meeting','Second meeting'])
|
|
self.assertEqual(connection.execute('SELECT sum(count) FROM volume_counts').fetchone()[0],1)
|
|
|
|
def test_internal_entities_are_rejected(self):
|
|
path=self.archive('',doctype='<!DOCTYPE dblp SYSTEM "dblp.dtd" [<!ENTITY x "expanded">]>')
|
|
with self.assertRaisesRegex(ValueError,'pinned external'):list(records(path,self.dtd.read_bytes()))
|
|
|
|
def test_checksum_failure_preserves_existing_output(self):
|
|
path=self.archive('');out=self.root/'out';out.mkdir();target=out/'dblp-ai-candidates.jsonl';target.write_text('previous')
|
|
with self.assertRaisesRegex(ValueError,'snapshot checksum'):catalogue(path,self.dtd,out,'0'*64)
|
|
self.assertEqual(target.read_text(),'previous')
|
|
|
|
def test_truncated_gzip_cannot_publish_a_catalogue(self):
|
|
path=self.archive('<inproceedings key="conf/aaai/one"><title>One</title></inproceedings>')
|
|
path.write_bytes(path.read_bytes()[:-8]);out=self.root/'out'
|
|
with self.assertRaises((EOFError,OSError)):
|
|
catalogue(path,self.dtd,out,hashlib.sha256(path.read_bytes()).hexdigest(),hashlib.md5(self.dtd.read_bytes()).hexdigest())
|
|
self.assertFalse((out/'dblp-ai-candidates.jsonl').exists())
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|