58 lines
2.6 KiB
Python
58 lines
2.6 KiB
Python
import json
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
from import_arxiv import ingest
|
|
|
|
|
|
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 sqlite3.connect(self.db) as 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 sqlite3.connect(self.db) as 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 sqlite3.connect(self.db) as 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 sqlite3.connect(self.db) as 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)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|