99 lines
3.9 KiB
Python
99 lines
3.9 KiB
Python
import os
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
import audio_cleanup as cleanup
|
|
|
|
|
|
class CleanupTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
self.outputs = self.root / 'outputs'
|
|
self.outputs.mkdir()
|
|
self.db = self.root / 'jobs.db3'
|
|
self.voice_db = self.root / 'voice.db3'
|
|
for name, value in [('DATABASE', self.db), ('OUTPUTS', self.outputs), ('VOICE_DATABASE', self.voice_db)]:
|
|
p = patch.object(cleanup, name, value)
|
|
p.start()
|
|
self.addCleanup(p.stop)
|
|
with sqlite3.connect(self.db) as db:
|
|
db.execute('CREATE TABLE jobs(id TEXT, output_path TEXT, status TEXT)')
|
|
with sqlite3.connect(self.voice_db) as db:
|
|
db.execute('CREATE TABLE jobs(status TEXT)')
|
|
cleanup.initialize()
|
|
self.local = self.root / 'job/audiobook.m4a'
|
|
self.local.parent.mkdir()
|
|
self.local.write_bytes(b'complete mp4')
|
|
self.output = self.outputs / 'audiobook_test.m4b'
|
|
self.output.write_bytes(self.local.read_bytes())
|
|
self.work = self.outputs / 'audiobook_test'
|
|
self.work.mkdir()
|
|
self.wav = self.outputs / ('longform_cache/' + 'a' * 20 + '.wav')
|
|
self.wav.parent.mkdir()
|
|
self.wav.write_bytes(b'chapter')
|
|
self.segment = self.wav.parent / ('segments/' + 'b' * 20 + '.wav')
|
|
self.segment.parent.mkdir()
|
|
self.segment.write_bytes(b'segment')
|
|
os.utime(self.segment, (1, 1))
|
|
(self.work / 'concat.txt').write_text("file '" + str(self.wav) + "'\n")
|
|
with cleanup.connect() as db:
|
|
db.execute('INSERT INTO jobs VALUES(?,?,?)', ('job', str(self.local), 'ready'))
|
|
|
|
def acknowledge(self):
|
|
with cleanup.connect() as db:
|
|
db.execute("DELETE FROM jobs WHERE id='job'")
|
|
|
|
def test_keeps_everything_until_phone_acknowledges_then_cleans(self):
|
|
other = self.outputs / 'audiobook_other.m4b'
|
|
other.write_bytes(b'other book')
|
|
cleanup.remember_output(self.local, self.output)
|
|
cleanup.sweep()
|
|
self.assertTrue(self.output.exists())
|
|
self.assertTrue(self.wav.exists())
|
|
self.acknowledge()
|
|
cleanup.sweep()
|
|
for path in [self.output, self.work, self.wav, self.segment]:
|
|
self.assertFalse(path.exists(), path)
|
|
self.assertTrue(other.exists())
|
|
self.assertTrue(self.local.exists())
|
|
cleanup.sweep() # restart/retry is idempotent
|
|
|
|
def test_active_generation_defers_cleanup(self):
|
|
cleanup.remember_output(self.local, self.output)
|
|
self.acknowledge()
|
|
with cleanup.connect() as db:
|
|
db.execute("INSERT INTO jobs VALUES('next', '', 'synthesizing')")
|
|
cleanup.sweep()
|
|
self.assertTrue(self.output.exists())
|
|
with cleanup.connect() as db:
|
|
db.execute("DELETE FROM jobs WHERE id='next'")
|
|
cleanup.sweep()
|
|
self.assertFalse(self.output.exists())
|
|
|
|
def test_legacy_worker_is_matched_by_content_not_title(self):
|
|
cleanup.capture_legacy_outputs()
|
|
self.acknowledge()
|
|
self.local.unlink()
|
|
cleanup.sweep()
|
|
self.assertFalse(self.output.exists())
|
|
|
|
def test_rejects_path_escape_and_retains_acknowledgement_on_error(self):
|
|
cleanup.remember_output(self.local, self.output)
|
|
self.acknowledge()
|
|
outside = self.root / 'keep.wav'
|
|
outside.write_bytes(b'keep')
|
|
(self.work / 'concat.txt').write_text("file '" + str(outside) + "'\n")
|
|
with self.assertRaises(ValueError):
|
|
cleanup.sweep()
|
|
self.assertTrue(outside.exists())
|
|
with cleanup.connect() as db:
|
|
self.assertEqual(0, db.execute('SELECT completed FROM delivered_audio_cleanup').fetchone()[0])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|