83 lines
4.7 KiB
Python
83 lines
4.7 KiB
Python
import json
|
|
from contextlib import closing
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
from paper_atlas.ingestion.arxiv import ingest
|
|
from paper_atlas.db.store import Store
|
|
from paper_atlas.ingestion.arxiv import refresh_categories
|
|
|
|
|
|
class ImportTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.root = Path(self.tmp.name)
|
|
self.db = self.root/'papers.sqlite'
|
|
self.archive = self.root/'metadata.zip'
|
|
|
|
def write(self, rows):
|
|
with zipfile.ZipFile(self.archive,'w') as z:
|
|
z.writestr('arxiv-metadata-oai-snapshot.json','\n'.join(json.dumps(r) for r in rows))
|
|
|
|
def row(self, identifier, doi=None):
|
|
return dict(id=identifier,title='A shared title',abstract='Searchable abstract',authors_parsed=[['Doe','Jane','']],categories='cs.AI cs.LG',doi=doi,versions=[])
|
|
|
|
def test_identity_provenance_and_idempotency(self):
|
|
self.write([self.row('2601.00001','10.0000/example'),self.row('2601.00002','10.0000/example')])
|
|
ingest(self.db,self.archive)
|
|
ingest(self.db,self.archive)
|
|
with closing(sqlite3.connect(self.db)) as c, c:
|
|
self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0],1)
|
|
self.assertEqual(c.execute('SELECT count(*) FROM source_records').fetchone()[0],2)
|
|
self.assertEqual(c.execute("SELECT count(*) FROM papers_fts WHERE papers_fts MATCH 'Searchable'").fetchone()[0],1)
|
|
self.assertEqual(c.execute('SELECT count(*) FROM decisions').fetchone()[0],0)
|
|
self.assertEqual(c.execute('PRAGMA foreign_key_check').fetchall(),[])
|
|
self.assertEqual(c.execute('PRAGMA integrity_check').fetchone()[0],'ok')
|
|
|
|
def test_title_alone_does_not_merge(self):
|
|
self.write([self.row('2601.00001'),self.row('2601.00002')])
|
|
ingest(self.db,self.archive)
|
|
with closing(sqlite3.connect(self.db)) as c, c:
|
|
self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0],2)
|
|
|
|
def test_platform_and_custom_tags_remain_distinct(self):
|
|
self.write([self.row('2601.00001')])
|
|
ingest(self.db,self.archive)
|
|
with closing(sqlite3.connect(self.db)) as c, c:
|
|
paper = c.execute('SELECT id FROM papers').fetchone()[0]
|
|
tag = c.execute("INSERT INTO tags(namespace,value) VALUES('custom:topic','cs.AI')").lastrowid
|
|
c.execute('INSERT INTO paper_tags VALUES(?,?,NULL,?)',(paper,tag,'manual:1'))
|
|
ingest(self.db,self.archive)
|
|
with closing(sqlite3.connect(self.db)) as c, c:
|
|
self.assertEqual(c.execute("SELECT count(*) FROM paper_tags pt JOIN tags t ON pt.tag_id=t.id WHERE t.value='cs.AI'").fetchone()[0],2)
|
|
|
|
def test_merged_records_retain_separate_category_evidence_across_refresh(self):
|
|
one=self.row('2601.00001','10.0000/example');two=self.row('2601.00002','10.0000/example')
|
|
self.write([one,two]);ingest(self.db,self.archive)
|
|
with closing(sqlite3.connect(self.db)) as c:
|
|
self.assertEqual(c.execute('SELECT count(*) FROM paper_tags').fetchone()[0],4)
|
|
self.assertEqual(c.execute('SELECT count(DISTINCT source_record_id) FROM paper_tags').fetchone()[0],2)
|
|
one['categories']='cs.AI';two['categories']='cs.LG';self.write([one,two]);ingest(self.db,self.archive)
|
|
with closing(sqlite3.connect(self.db)) as c:
|
|
actual=c.execute('SELECT r.external_id,t.value FROM paper_tags pt JOIN source_records r ON r.id=pt.source_record_id JOIN tags t ON t.id=pt.tag_id ORDER BY r.external_id').fetchall()
|
|
self.assertEqual(actual,[('2601.00001','cs.AI'),('2601.00002','cs.LG')])
|
|
|
|
def test_migrate_legacy_categories_without_removing_custom_evidence(self):
|
|
store=Store(self.db);self.addCleanup(store.conn.close)
|
|
paper,record=store.record('arxiv','2601.00001','https://arxiv.org/abs/2601.00001','A title','AB',[])
|
|
for namespace,key in [('arxiv:category','arxiv:categories'),('custom:topic','manual:1')]:
|
|
tag=store.conn.execute('INSERT INTO tags(namespace,value) VALUES(?,?)',(namespace,'cs.AI')).lastrowid
|
|
store.conn.execute('INSERT INTO paper_tags VALUES(?,?,?,?)',(paper,tag,record,key))
|
|
refresh_categories(store,paper,record,'cs.AI cs.LG');refresh_categories(store,paper,record,'cs.AI cs.LG')
|
|
self.assertEqual(store.conn.execute('SELECT count(*) FROM paper_tags').fetchone()[0],3)
|
|
self.assertEqual(store.conn.execute("SELECT count(*) FROM paper_tags WHERE evidence_key='arxiv:categories'").fetchone()[0],0)
|
|
self.assertEqual(store.conn.execute("SELECT count(*) FROM paper_tags WHERE evidence_key='manual:1'").fetchone()[0],1)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|