42 lines
2.1 KiB
Python
42 lines
2.1 KiB
Python
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from paper_atlas.db.store import Store
|
|
from paper_atlas.ingestion.maintenance import merge
|
|
|
|
|
|
class MaintenanceTests(unittest.TestCase):
|
|
def fixture(self,tmp,status):
|
|
db=Path(tmp)/'papers.sqlite';archive=Path(tmp)/'metadata.zip';archive.write_bytes(b'not read by schema-only maintenance')
|
|
store=Store(db);source=store.source('arxiv','https://arxiv.org')
|
|
store.conn.execute('INSERT INTO ingest_runs(source_id,status) VALUES(?,?)',(source,status));store.conn.commit();store.conn.close()
|
|
return db,archive
|
|
|
|
def test_completed_archive_is_not_started_again_after_maintenance(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
db,archive=self.fixture(tmp,'complete')
|
|
with patch('paper_atlas.ingestion.maintenance.property_value',return_value='inactive'),patch('paper_atlas.ingestion.maintenance.subprocess.run') as run:
|
|
merge(db,archive,[])
|
|
run.assert_not_called()
|
|
|
|
def test_inactive_incomplete_arxiv_requires_investigation(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
db,archive=self.fixture(tmp,'failed')
|
|
with patch('paper_atlas.ingestion.maintenance.property_value',return_value='inactive'),patch('paper_atlas.ingestion.maintenance.subprocess.run') as run:
|
|
with self.assertRaisesRegex(RuntimeError,'neither active nor verified complete'):
|
|
merge(db,archive,[])
|
|
run.assert_not_called()
|
|
|
|
def test_arxiv_finishing_at_the_stop_boundary_is_not_restarted(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
db,archive=self.fixture(tmp,'complete')
|
|
with patch('paper_atlas.ingestion.maintenance.property_value',return_value='active'),patch('paper_atlas.ingestion.maintenance.subprocess.run') as run:
|
|
merge(db,archive,[])
|
|
self.assertEqual(run.call_args_list[0].args[0],['systemctl','stop','paper-atlas-arxiv-import.service'])
|
|
self.assertEqual(run.call_count,1)
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|