821 lines
37 KiB
Python
821 lines
37 KiB
Python
"""The transcription pipeline (podcasts/transcribe.py) against a fake Whisper instance, the
|
|
endpoints around it and the notice in /feeds/notifications/."""
|
|
import email
|
|
import io
|
|
import json
|
|
import math
|
|
import os
|
|
import struct
|
|
import subprocess
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import wave
|
|
from datetime import timedelta
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from unittest.mock import patch
|
|
|
|
from django.contrib.auth.models import User
|
|
from django.db import OperationalError
|
|
from django.test import TestCase, TransactionTestCase, override_settings
|
|
from django.utils import timezone
|
|
|
|
from feeds.models import Item, Source
|
|
|
|
from . import transcribe, transcription, youtube
|
|
from .models import (
|
|
EpisodeArticle, EpisodeTranscript, PodcastEpisode, PodcastFeed, TranscriptionJob, TranscriptionSettings,
|
|
)
|
|
|
|
RATE = 16000
|
|
TOTAL = 120
|
|
# Where the long pauses are (start, length): the pieces should be cut in these. The short
|
|
# ones (0.3 s every 4 s) are only there to tempt.
|
|
LONG_PAUSES = [(36.5, 1.6), (83.0, 1.4)]
|
|
SEGMENTS = [{'start': 0.0, 'end': 2.0, 'text': ' Hallo '}, {'start': 2.0, 'end': 5.5, 'text': 'Welt.'}]
|
|
|
|
|
|
def make_audio():
|
|
"""~2 minutes of tones with pauses, as mp3 bytes (and the wav as a check of the length)."""
|
|
samples = []
|
|
t = 0.0
|
|
pauses = [(s, s + n) for s, n in LONG_PAUSES]
|
|
for i in range(RATE * TOTAL):
|
|
t = i / RATE
|
|
silent = any(a <= t < b for a, b in pauses) or (t % 4.0) > 3.7
|
|
value = 0 if silent else int(9000 * math.sin(2 * math.pi * (180 + 40 * ((i // 8000) % 5)) * t))
|
|
samples.append(value)
|
|
buf = io.BytesIO()
|
|
with wave.open(buf, 'wb') as w:
|
|
w.setnchannels(1)
|
|
w.setsampwidth(2)
|
|
w.setframerate(RATE)
|
|
w.writeframes(struct.pack('<%dh' % len(samples), *samples))
|
|
import imageio_ffmpeg
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
src, dst = os.path.join(tmp, 'a.wav'), os.path.join(tmp, 'a.mp3')
|
|
open(src, 'wb').write(buf.getvalue())
|
|
subprocess.run([imageio_ffmpeg.get_ffmpeg_exe(), '-loglevel', 'error', '-i', src, '-b:a', '64k', dst], check=True)
|
|
return open(dst, 'rb').read()
|
|
|
|
|
|
_AUDIO = []
|
|
|
|
|
|
def audio():
|
|
if not _AUDIO:
|
|
_AUDIO.append(make_audio())
|
|
return _AUDIO[0]
|
|
|
|
|
|
def decoded_seconds(data, suffix):
|
|
"""How long the uploaded piece is, by decoding it."""
|
|
import imageio_ffmpeg
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = os.path.join(tmp, 'p' + suffix)
|
|
open(path, 'wb').write(data)
|
|
out = subprocess.run([imageio_ffmpeg.get_ffmpeg_exe(), '-loglevel', 'error', '-i', path, '-f', 's16le',
|
|
'-ac', '1', '-ar', str(RATE), '-'], stdout=subprocess.PIPE, check=True).stdout
|
|
return len(out) / 2 / RATE
|
|
|
|
|
|
class Fake:
|
|
"""A local HTTP server: GET /ep.mp3 is the test audio, POST /v1/audio/transcriptions a Whisper."""
|
|
|
|
def __init__(self, fail_first=0, fail_status=500, delay=0, on_call=None):
|
|
self.fail_first, self.fail_status, self.delay, self.on_call = fail_first, fail_status, delay, on_call
|
|
self.calls = [] # {'fields', 'filename', 'body', 'auth'}
|
|
self.lock = threading.Lock()
|
|
fake = self
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def log_message(self, *a):
|
|
pass
|
|
|
|
def do_GET(self):
|
|
if self.path == '/missing.mp3':
|
|
self.send_response(404)
|
|
self.end_headers()
|
|
return
|
|
body = audio_holder.data or audio()
|
|
self.send_response(200)
|
|
self.send_header('Content-Type', 'audio/mpeg')
|
|
self.send_header('Content-Length', str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def do_POST(self):
|
|
raw = self.rfile.read(int(self.headers['Content-Length']))
|
|
msg = email.message_from_bytes(
|
|
b'Content-Type: ' + self.headers['Content-Type'].encode() + b'\r\n\r\n' + raw)
|
|
fields, filename, body = {}, '', b''
|
|
for part in msg.get_payload():
|
|
name = part.get_param('name', header='content-disposition')
|
|
if part.get_filename():
|
|
filename, body = part.get_filename(), part.get_payload(decode=True)
|
|
else:
|
|
fields[name] = part.get_payload(decode=True).decode()
|
|
with fake.lock:
|
|
fake.calls.append({'fields': fields, 'filename': filename, 'body': body,
|
|
'auth': self.headers.get('Authorization')})
|
|
n = len(fake.calls)
|
|
if fake.on_call:
|
|
fake.on_call(n)
|
|
if fake.delay:
|
|
time.sleep(fake.delay)
|
|
if n <= fake.fail_first or fake.fail_first < 0:
|
|
self.send_response(fake.fail_status)
|
|
self.send_header('Content-Length', '0')
|
|
self.end_headers()
|
|
return
|
|
out = json.dumps({'language': 'german', 'segments': SEGMENTS}).encode()
|
|
self.send_response(200)
|
|
self.send_header('Content-Type', 'application/json')
|
|
self.send_header('Content-Length', str(len(out)))
|
|
self.end_headers()
|
|
self.wfile.write(out)
|
|
|
|
self.server = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
|
|
self.server.daemon_threads = True
|
|
self.port = self.server.server_address[1]
|
|
threading.Thread(target=self.server.serve_forever, daemon=True).start()
|
|
|
|
@property
|
|
def whisper(self):
|
|
return f'http://127.0.0.1:{self.port}/v1'
|
|
|
|
def url(self, path='/ep.mp3'):
|
|
return f'http://127.0.0.1:{self.port}{path}'
|
|
|
|
def stop(self):
|
|
self.server.shutdown()
|
|
self.server.server_close()
|
|
|
|
|
|
def leftovers():
|
|
base = tempfile.gettempdir()
|
|
return {d for d in os.listdir(base) if d.startswith('diora-transcribe-')}
|
|
|
|
|
|
FAST = dict(CHUNK_SECONDS=40, CUT_WINDOW=8, MIN_TAIL=10, RETRY_PAUSES=(0, 0, 0))
|
|
|
|
|
|
def wait_for(cond, timeout=30):
|
|
end = time.time() + timeout
|
|
while time.time() < end:
|
|
if cond():
|
|
return True
|
|
time.sleep(0.05)
|
|
return False
|
|
|
|
|
|
class PipelineBase(TransactionTestCase):
|
|
def setUp(self):
|
|
for k, v in FAST.items():
|
|
patcher = patch.object(transcribe, k, v)
|
|
patcher.start()
|
|
self.addCleanup(patcher.stop)
|
|
self.user = User.objects.create_user(username='alice', password='pw12345678')
|
|
self.feed = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/f', title='Der Podcast',
|
|
author='Anna Autorin')
|
|
self.fake = Fake()
|
|
self.addCleanup(self.fake.stop)
|
|
self.before = leftovers()
|
|
|
|
def job(self, audio_path='/ep.mp3', **kw):
|
|
n = PodcastEpisode.objects.count()
|
|
ep = PodcastEpisode.objects.create(feed=self.feed, guid=f'g{n}', title=f'Folge {n}',
|
|
audio_url=self.fake.url(audio_path), duration_seconds=TOTAL)
|
|
return TranscriptionJob.objects.create(episode=ep, state=TranscriptionJob.QUEUED, **kw)
|
|
|
|
def reload(self, job):
|
|
for _ in range(50): # the in-memory test database may be locked for a moment by the worker
|
|
try:
|
|
return TranscriptionJob.objects.get(pk=job.pk)
|
|
except OperationalError:
|
|
time.sleep(0.05)
|
|
return TranscriptionJob.objects.get(pk=job.pk)
|
|
|
|
def assertClean(self):
|
|
self.assertEqual(leftovers(), self.before)
|
|
|
|
|
|
class PlanTests(PipelineBase):
|
|
def test_cuts_land_in_the_long_pauses(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
src, pcm = os.path.join(tmp, 'a.mp3'), os.path.join(tmp, 'a.pcm')
|
|
open(src, 'wb').write(audio())
|
|
transcribe._decode(src, pcm)
|
|
total = os.path.getsize(pcm) / 2 / RATE
|
|
pieces = transcribe.plan(total, transcribe.pauses_of(pcm))
|
|
self.assertAlmostEqual(total, TOTAL, delta=0.5)
|
|
self.assertEqual(len(pieces), 3)
|
|
cuts = [p[1] for p in pieces[:-1]]
|
|
for cut, (start, length) in zip(cuts, LONG_PAUSES):
|
|
self.assertTrue(start <= cut <= start + length, f'{cut} not in the pause {start}+{length}')
|
|
self.assertEqual(pieces[0][0], 0)
|
|
self.assertEqual(pieces[-1][1], total)
|
|
self.assertEqual([a[1] for a in pieces[:-1]], [b[0] for b in pieces[1:]])
|
|
|
|
def test_plan_without_pauses_cuts_at_the_target(self):
|
|
self.assertEqual(transcribe.plan(100, []), [(0.0, 40), (40, 80), (80, 100)])
|
|
self.assertEqual(transcribe.plan(30, []), [(0.0, 30)])
|
|
# a short tail is joined to the last piece
|
|
self.assertEqual(transcribe.plan(85, []), [(0.0, 40), (40, 85)])
|
|
|
|
|
|
@override_settings(WHISPER_URL='')
|
|
class RunTests(PipelineBase):
|
|
def configure(self, **kw):
|
|
conf = TranscriptionSettings(user=self.user, **kw)
|
|
conf.save()
|
|
return conf
|
|
|
|
def test_builtin_run_end_to_end(self):
|
|
with override_settings(WHISPER_URL=self.fake.whisper):
|
|
self.configure(language='de', parallel=2)
|
|
job = self.job()
|
|
started = time.time()
|
|
transcribe.run(job.pk)
|
|
took = time.time() - started
|
|
print(f'\n[pipeline run of the {TOTAL} s test audio took {took:.1f} s]')
|
|
job = self.reload(job)
|
|
self.assertEqual((job.state, job.error), ('done', ''))
|
|
self.assertEqual((job.chunks_done, job.chunks_total), (3, 3))
|
|
self.assertIsNotNone(job.finished_at)
|
|
self.assertEqual(len(self.fake.calls), 3)
|
|
for c in self.fake.calls:
|
|
self.assertEqual(c['fields']['model'], 'whisper-1')
|
|
self.assertEqual(c['fields']['language'], 'de')
|
|
self.assertEqual(c['fields']['response_format'], 'verbose_json')
|
|
self.assertEqual(c['fields']['prompt'], 'Der Podcast. Folge 0. Anna Autorin')
|
|
self.assertTrue(c['filename'].endswith('.flac'))
|
|
self.assertIsNone(c['auth']) # the token only goes to custom instances
|
|
# the uploaded pieces are the planned ones: the same total, cut at the pauses
|
|
lengths = sorted(decoded_seconds(c['body'], '.flac') for c in self.fake.calls)
|
|
self.assertAlmostEqual(sum(lengths), TOTAL, delta=0.5)
|
|
data = job.data
|
|
self.assertEqual(data['source'], {'type': 'whisper', 'model': 'whisper-1', 'language': 'de'})
|
|
segs = data['segments']
|
|
starts = [s['start'] for s in segs]
|
|
self.assertEqual(starts, sorted(starts))
|
|
self.assertTrue(all(s['speaker'] == '' for s in segs))
|
|
# every piece answered "Hallo Welt." at 0 to 5.5 s: its start is where the piece starts,
|
|
# which lies on a cut in a long pause, and the first at zero
|
|
self.assertEqual(starts[0], 0)
|
|
cuts = starts[1:]
|
|
self.assertEqual(len(cuts), 2)
|
|
for cut, (p, n) in zip(cuts, LONG_PAUSES):
|
|
self.assertTrue(p <= cut <= p + n, f'{cut} not in {p}+{n}')
|
|
self.assertEqual(segs[0]['text'], 'Hallo Welt.')
|
|
self.assertAlmostEqual(segs[0]['end'], 5.5)
|
|
self.assertClean()
|
|
|
|
def test_language_empty_takes_what_whisper_detects_and_custom_gets_opus_and_token(self):
|
|
conf = self.configure(engine='custom', url=self.fake.whisper, model='large-v3', language='')
|
|
conf.token_encrypted = transcription.logins.encrypt('tok')
|
|
conf.save()
|
|
job = self.job()
|
|
with patch.object(transcribe.chapters, 'assert_safe_url'), \
|
|
patch.object(transcribe.chapters, '_assert_peer_is_safe'):
|
|
transcribe.run(job.pk)
|
|
job = self.reload(job)
|
|
self.assertEqual((job.state, job.error), ('done', ''))
|
|
self.assertEqual(job.data['source'], {'type': 'whisper', 'model': 'large-v3', 'language': 'german'})
|
|
for c in self.fake.calls:
|
|
self.assertTrue(c['filename'].endswith('.ogg'))
|
|
self.assertNotIn('language', c['fields'])
|
|
self.assertEqual(c['auth'], 'Bearer tok')
|
|
self.assertLess(len(c['body']), decoded_seconds(c['body'], '.ogg') * 6500) # 32 kbit/s: about 4 kB a second, ogg and VBR on top
|
|
self.assertAlmostEqual(sum(decoded_seconds(c['body'], '.ogg') for c in self.fake.calls), TOTAL, delta=0.5)
|
|
self.assertClean()
|
|
|
|
def test_custom_instance_on_a_private_address_is_refused(self):
|
|
self.configure(engine='custom', url=self.fake.whisper)
|
|
job = self.job()
|
|
transcribe.run(job.pk)
|
|
job = self.reload(job)
|
|
self.assertEqual(job.state, 'error')
|
|
self.assertIn('Whisper', job.error)
|
|
self.assertEqual(self.fake.calls, [])
|
|
|
|
def test_unreachable_builtin_waits_and_tries_again(self):
|
|
import socket
|
|
sock = socket.socket()
|
|
sock.bind(('127.0.0.1', 0))
|
|
port = sock.getsockname()[1]
|
|
sock.close() # nothing listens there now
|
|
with override_settings(WHISPER_URL=f'http://127.0.0.1:{port}/v1'):
|
|
self.configure()
|
|
job = self.job()
|
|
transcribe.run(job.pk)
|
|
job = self.reload(job)
|
|
self.assertEqual(job.state, 'queued')
|
|
self.assertIn('nicht erreichbar', job.error)
|
|
self.assertGreater(transcribe._not_before[job.pk], time.time())
|
|
self.assertEqual(self.fake.calls, []) # nothing downloaded while the instance is away
|
|
self.assertEqual(transcribe._next_job(), transcribe.WAIT)
|
|
transcribe._not_before[job.pk] = 0 # due again: the instance is back
|
|
with override_settings(WHISPER_URL=self.fake.whisper):
|
|
transcribe.run(job.pk)
|
|
job = self.reload(job)
|
|
self.assertEqual((job.state, job.error), ('done', ''))
|
|
transcribe._not_before.clear()
|
|
|
|
def test_waiting_gives_up_after_two_days(self):
|
|
from datetime import timedelta
|
|
from django.utils import timezone
|
|
with override_settings(WHISPER_URL='http://127.0.0.1:9/v1'):
|
|
self.configure()
|
|
job = self.job()
|
|
TranscriptionJob.objects.filter(pk=job.pk).update(created_at=timezone.now() - timedelta(days=3))
|
|
transcribe.run(job.pk)
|
|
job = self.reload(job)
|
|
self.assertEqual(job.state, 'error')
|
|
self.assertIn('zwei Tage', job.error)
|
|
|
|
@override_settings(WHISPER_URL='')
|
|
def test_retries_then_succeeds(self):
|
|
self.fake.fail_first = 2
|
|
with override_settings(WHISPER_URL=self.fake.whisper):
|
|
self.configure(parallel=1)
|
|
job = self.job()
|
|
transcribe.run(job.pk)
|
|
self.assertEqual(self.reload(job).state, 'done')
|
|
self.assertEqual(len(self.fake.calls), 3 + 2)
|
|
|
|
def test_gives_up_after_three_retries(self):
|
|
self.fake.fail_first = -1
|
|
with override_settings(WHISPER_URL=self.fake.whisper):
|
|
self.configure(parallel=1)
|
|
job = self.job()
|
|
transcribe.run(job.pk)
|
|
job = self.reload(job)
|
|
self.assertEqual(job.state, 'error')
|
|
self.assertEqual(job.error, 'Whisper: HTTP 500')
|
|
self.assertEqual(len(self.fake.calls), 4) # the first try and three more
|
|
self.assertIsNone(job.data)
|
|
self.assertClean()
|
|
|
|
def test_a_refusal_is_not_retried(self):
|
|
self.fake.fail_first, self.fake.fail_status = -1, 401
|
|
with override_settings(WHISPER_URL=self.fake.whisper):
|
|
self.configure(parallel=1)
|
|
job = self.job()
|
|
transcribe.run(job.pk)
|
|
job = self.reload(job)
|
|
self.assertEqual((job.state, job.error), ('error', 'Whisper: HTTP 401'))
|
|
self.assertEqual(len(self.fake.calls), 1)
|
|
|
|
def test_download_failure(self):
|
|
with override_settings(WHISPER_URL=self.fake.whisper):
|
|
job = self.job('/missing.mp3')
|
|
transcribe.run(job.pk)
|
|
job = self.reload(job)
|
|
self.assertEqual((job.state, job.error), ('error', 'Download: HTTP 404'))
|
|
self.assertClean()
|
|
|
|
def test_not_audio(self):
|
|
with override_settings(WHISPER_URL=self.fake.whisper), patch.object(audio_holder, 'data', b'<html>no</html>'):
|
|
job = self.job()
|
|
transcribe.run(job.pk)
|
|
job = self.reload(job)
|
|
self.assertEqual(job.state, 'error')
|
|
self.assertTrue(job.error.startswith('ffmpeg:'), job.error)
|
|
self.assertClean()
|
|
|
|
def test_no_builtin_instance(self):
|
|
job = self.job()
|
|
transcribe.run(job.pk)
|
|
self.assertEqual(self.reload(job).error, 'Keine eingebaute Instanz eingerichtet')
|
|
|
|
def test_cancel_while_running(self):
|
|
job = self.job()
|
|
|
|
def cancel(n):
|
|
if n == 1:
|
|
TranscriptionJob.objects.filter(pk=job.pk).delete()
|
|
from django.db import connection
|
|
connection.close()
|
|
self.fake.on_call, self.fake.delay = cancel, 0.2
|
|
with override_settings(WHISPER_URL=self.fake.whisper):
|
|
self.configure(parallel=1)
|
|
transcribe.run(job.pk)
|
|
self.assertFalse(TranscriptionJob.objects.filter(pk=job.pk).exists())
|
|
self.assertLessEqual(len(self.fake.calls), 1) # no further piece was sent
|
|
self.assertClean()
|
|
|
|
def test_cancelled_while_waiting_is_not_started(self):
|
|
job = self.job()
|
|
job.delete()
|
|
transcribe.run(job.pk)
|
|
self.assertEqual(self.fake.calls, [])
|
|
|
|
|
|
class _Holder:
|
|
"""Lets a test swap the audio the fake server sends."""
|
|
data = None
|
|
|
|
|
|
audio_holder = _Holder()
|
|
|
|
|
|
class QueueTests(PipelineBase):
|
|
"""The runner and its queue. The work itself (_transcribe) is replaced by a stand-in the test
|
|
holds back, so only the runner thread writes to the database while the test looks at it."""
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.gate = threading.Event()
|
|
self.started = []
|
|
patcher = patch.object(transcribe, '_resumed', True)
|
|
patcher.start()
|
|
self.addCleanup(patcher.stop)
|
|
work = patch.object(transcribe, '_transcribe', side_effect=self.work)
|
|
work.start()
|
|
self.addCleanup(work.stop)
|
|
self.addCleanup(self.gate.set)
|
|
self.addCleanup(lambda: wait_for(lambda: transcribe._runner is None))
|
|
|
|
def work(self, job_id):
|
|
self.started.append(job_id)
|
|
self.gate.wait(30)
|
|
return {'segments': [{'start': 0, 'end': 1, 'speaker': '', 'text': 'x'}], 'source': {'type': 'whisper'}}
|
|
|
|
def test_one_job_at_a_time_oldest_first(self):
|
|
first, second, third = self.job(), self.job(), self.job()
|
|
transcribe.kick()
|
|
self.assertTrue(wait_for(lambda: self.started == [first.pk]))
|
|
transcribe.kick() # a second runner doesn't start
|
|
time.sleep(0.3)
|
|
self.assertEqual(self.started, [first.pk])
|
|
self.assertEqual([self.reload(j).state for j in (first, second, third)], ['downloading', 'queued', 'queued'])
|
|
self.gate.set()
|
|
self.assertTrue(wait_for(lambda: transcribe._runner is None))
|
|
self.assertEqual(self.started, [first.pk, second.pk, third.pk])
|
|
jobs = [self.reload(j) for j in (first, second, third)]
|
|
self.assertEqual([j.state for j in jobs], ['done'] * 3)
|
|
self.assertEqual(sorted(j.finished_at for j in jobs), [j.finished_at for j in jobs])
|
|
|
|
def test_a_job_queued_while_the_runner_works_is_taken_too(self):
|
|
first = self.job()
|
|
transcribe.kick()
|
|
self.assertTrue(wait_for(lambda: self.started == [first.pk]))
|
|
second = self.job()
|
|
transcribe.kick()
|
|
self.gate.set()
|
|
self.assertTrue(wait_for(lambda: transcribe._runner is None))
|
|
self.assertEqual(self.started, [first.pk, second.pk])
|
|
|
|
def test_a_cancelled_waiting_job_is_skipped(self):
|
|
first, second, third = self.job(), self.job(), self.job()
|
|
transcribe.kick()
|
|
self.assertTrue(wait_for(lambda: self.started == [first.pk]))
|
|
second.delete()
|
|
self.gate.set()
|
|
self.assertTrue(wait_for(lambda: transcribe._runner is None))
|
|
self.assertEqual(self.started, [first.pk, third.pk])
|
|
|
|
def test_a_crashing_job_does_not_stop_the_queue(self):
|
|
first, second = self.job(), self.job()
|
|
self.gate.set()
|
|
calls = []
|
|
|
|
def work(job_id):
|
|
calls.append(job_id)
|
|
if job_id == first.pk:
|
|
raise RuntimeError('boom')
|
|
return self.work(job_id)
|
|
with patch.object(transcribe, '_transcribe', side_effect=work):
|
|
transcribe.kick()
|
|
self.assertTrue(wait_for(lambda: transcribe._runner is None))
|
|
self.assertEqual(calls, [first.pk, second.pk])
|
|
self.assertEqual((self.reload(first).state, self.reload(first).error), ('error', 'RuntimeError: boom'))
|
|
self.assertEqual(self.reload(second).state, 'done')
|
|
|
|
def test_stuck_jobs_start_again_after_a_restart(self):
|
|
self.gate.set()
|
|
job = self.job()
|
|
TranscriptionJob.objects.filter(pk=job.pk).update(state='transcribing', chunks_done=2, chunks_total=3)
|
|
with patch.object(transcribe, '_resumed', False):
|
|
transcribe.kick()
|
|
self.assertTrue(wait_for(lambda: transcribe._runner is None))
|
|
self.assertEqual(self.started, [job.pk])
|
|
self.assertEqual(self.reload(job).state, 'done')
|
|
|
|
def test_nothing_queued_starts_no_runner(self):
|
|
transcribe.kick()
|
|
self.assertIsNone(transcribe._runner)
|
|
|
|
|
|
# --------------------------------------------------------------------------- endpoints
|
|
|
|
class EndpointBase(TestCase):
|
|
def setUp(self):
|
|
self.user = User.objects.create_user(username='alice', password='pw12345678')
|
|
self.client.force_login(self.user)
|
|
self.feed = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/f', title='Der Podcast')
|
|
self.ep = self.episode('e1')
|
|
kick = patch.object(transcribe, 'kick')
|
|
self.kick = kick.start()
|
|
self.addCleanup(kick.stop)
|
|
|
|
def episode(self, guid, **kw):
|
|
kw.setdefault('audio_url', f'https://example.org/{guid}.mp3')
|
|
return PodcastEpisode.objects.create(feed=self.feed, guid=guid, title=f'Folge {guid}', duration_seconds=600, **kw)
|
|
|
|
def post(self, url, body=None, **extra):
|
|
return self.client.post(url, json.dumps(body or {}), content_type='application/json', **extra)
|
|
|
|
def done_job(self, ep, **kw):
|
|
kw.setdefault('data', {'segments': [{'start': 0, 'end': 4, 'speaker': '', 'text': 'Hallo.'}],
|
|
'source': {'type': 'whisper', 'model': 'whisper-1', 'language': 'de'}})
|
|
return TranscriptionJob.objects.create(episode=ep, state='done', finished_at=timezone.now(), **kw)
|
|
|
|
|
|
class TranscribeEndpointTests(EndpointBase):
|
|
def url(self, ep=None):
|
|
return f'/podcasts/episodes/{(ep or self.ep).id}/transcribe/'
|
|
|
|
def test_requires_login_and_ownership(self):
|
|
other = User.objects.create_user(username='bob', password='pw12345678')
|
|
self.client.force_login(other)
|
|
self.assertEqual(self.client.get(self.url()).status_code, 404)
|
|
self.assertEqual(self.post(self.url()).status_code, 404)
|
|
self.client.logout()
|
|
self.assertEqual(self.client.get(self.url()).status_code, 401)
|
|
|
|
def test_none_before_a_job(self):
|
|
self.assertEqual(self.client.get(self.url()).json(), {
|
|
'state': 'none', 'chunks_done': 0, 'chunks_total': 0, 'error': '', 'want_article': False,
|
|
'started_at': None, 'finished_at': None})
|
|
|
|
@patch('podcasts.transcription._too_large', return_value=False)
|
|
def test_post_queues_a_job_and_starts_the_runner(self, _):
|
|
r = self.post(self.url())
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertEqual((r.json()['state'], r.json()['want_article']), ('queued', False))
|
|
self.assertEqual(TranscriptionJob.objects.get(episode=self.ep).state, 'queued')
|
|
self.kick.assert_called()
|
|
|
|
@patch('podcasts.transcription._too_large', return_value=False)
|
|
def test_post_again_only_raises_the_article_wish(self, _):
|
|
self.post(self.url())
|
|
first = TranscriptionJob.objects.get(episode=self.ep).pk
|
|
r = self.post(self.url(), {'article': True})
|
|
self.assertTrue(r.json()['want_article'])
|
|
r = self.post(self.url(), {'article': False}) # never lowers it
|
|
self.assertTrue(r.json()['want_article'])
|
|
self.assertEqual(TranscriptionJob.objects.filter(episode=self.ep).count(), 1)
|
|
self.assertEqual(TranscriptionJob.objects.get(episode=self.ep).pk, first)
|
|
|
|
@patch('podcasts.transcription._too_large', return_value=False)
|
|
def test_post_with_article_true_sets_the_wish(self, _):
|
|
self.assertTrue(self.post(self.url(), {'article': True}).json()['want_article'])
|
|
|
|
def test_youtube_is_refused(self):
|
|
ep = self.episode('yt', audio_url=youtube.WATCH + 'abc123')
|
|
r = self.post(self.url(ep))
|
|
self.assertEqual((r.status_code, r.json()), (400, {'error': 'youtube'}))
|
|
self.assertFalse(TranscriptionJob.objects.exists())
|
|
|
|
@patch('podcasts.transcription._too_large', return_value=True)
|
|
def test_too_large_is_refused(self, _):
|
|
r = self.post(self.url())
|
|
self.assertEqual((r.status_code, r.json()), (400, {'error': 'too large'}))
|
|
self.assertFalse(TranscriptionJob.objects.exists())
|
|
|
|
def test_the_size_check_reads_the_content_length(self):
|
|
from . import silence
|
|
resp = type('R', (), {'headers': {'Content-Length': str(silence.MAX_BYTES + 1)}})()
|
|
with patch('podcasts.transcription.requests.head', return_value=resp):
|
|
self.assertTrue(transcription._too_large('https://example.org/x.mp3'))
|
|
resp.headers = {'Content-Length': '1000'}
|
|
with patch('podcasts.transcription.requests.head', return_value=resp):
|
|
self.assertFalse(transcription._too_large('https://example.org/x.mp3'))
|
|
with patch('podcasts.transcription.requests.head', side_effect=transcription.requests.ConnectionError):
|
|
self.assertFalse(transcription._too_large('https://example.org/x.mp3'))
|
|
|
|
def test_a_whisper_transcript_is_done_already(self):
|
|
self.done_job(self.ep)
|
|
r = self.post(self.url(), {'article': True})
|
|
self.assertEqual((r.json()['state'], r.json()['want_article']), ('done', True))
|
|
self.kick.assert_not_called()
|
|
self.assertTrue(TranscriptionJob.objects.get(episode=self.ep).want_article)
|
|
|
|
def test_a_feed_transcript_is_done_already(self):
|
|
self.ep.transcripts = [{'url': 'https://example.org/t.vtt', 'type': 'text/vtt'}]
|
|
self.ep.save()
|
|
r = self.post(self.url(), {'article': True})
|
|
self.assertEqual((r.json()['state'], r.json()['want_article']), ('done', True))
|
|
job = TranscriptionJob.objects.get(episode=self.ep)
|
|
self.assertIsNone(job.data)
|
|
self.kick.assert_not_called()
|
|
|
|
@patch('podcasts.transcription._too_large', return_value=False)
|
|
def test_after_an_error_post_starts_again_at_the_back(self, _):
|
|
old = TranscriptionJob.objects.create(episode=self.ep, state='error', error='Whisper: HTTP 500')
|
|
r = self.post(self.url(), {'article': True})
|
|
self.assertEqual((r.json()['state'], r.json()['error']), ('queued', ''))
|
|
new = TranscriptionJob.objects.get(episode=self.ep)
|
|
self.assertGreater(new.pk, old.pk)
|
|
|
|
@patch('podcasts.transcription._too_large', return_value=False)
|
|
def test_delete_cancels_queued_and_running_but_not_a_finished_one(self, _):
|
|
self.post(self.url())
|
|
self.assertEqual(self.client.delete(self.url()).json()['state'], 'none')
|
|
self.assertFalse(TranscriptionJob.objects.exists())
|
|
TranscriptionJob.objects.create(episode=self.ep, state='transcribing', chunks_total=5, chunks_done=2)
|
|
self.assertEqual(self.client.delete(self.url()).json()['state'], 'none')
|
|
self.assertFalse(TranscriptionJob.objects.exists())
|
|
self.done_job(self.ep)
|
|
self.assertEqual(self.client.delete(self.url()).json()['state'], 'done')
|
|
self.assertTrue(TranscriptionJob.objects.exists())
|
|
|
|
def test_get_shows_progress(self):
|
|
TranscriptionJob.objects.create(episode=self.ep, state='transcribing', chunks_total=38, chunks_done=7,
|
|
started_at=timezone.now())
|
|
d = self.client.get(self.url()).json()
|
|
self.assertEqual((d['state'], d['chunks_done'], d['chunks_total']), ('transcribing', 7, 38))
|
|
self.assertIsNotNone(d['started_at'])
|
|
self.kick.assert_called() # a restart's first request starts the queue again
|
|
|
|
|
|
class TranscriptTests(EndpointBase):
|
|
def transcript(self, ep=None):
|
|
return self.client.get(f'/podcasts/episodes/{(ep or self.ep).id}/transcript/')
|
|
|
|
def detail(self, ep=None):
|
|
return self.client.get(f'/podcasts/episodes/{(ep or self.ep).id}/').json()
|
|
|
|
def test_404_as_before_without_any(self):
|
|
self.assertEqual((self.transcript().status_code, self.transcript().json()), (404, {'error': 'no transcript'}))
|
|
d = self.detail()
|
|
self.assertEqual((d['has_transcript'], d['transcript_origin'], d['transcription'], d['article_item_id']),
|
|
(False, None, None, None))
|
|
|
|
def test_falls_back_to_the_whisper_transcript(self):
|
|
job = self.done_job(self.ep)
|
|
r = self.transcript()
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertEqual(r.json(), job.data)
|
|
d = self.detail()
|
|
self.assertEqual((d['has_transcript'], d['transcript_origin']), (True, 'whisper'))
|
|
self.assertEqual(d['transcription']['state'], 'done')
|
|
|
|
def test_a_running_job_has_no_transcript_yet(self):
|
|
TranscriptionJob.objects.create(episode=self.ep, state='transcribing', data=None)
|
|
self.assertEqual(self.transcript().status_code, 404)
|
|
d = self.detail()
|
|
self.assertEqual((d['has_transcript'], d['transcript_origin'], d['transcription']['state']),
|
|
(False, None, 'transcribing'))
|
|
|
|
def feed_transcript(self, error=''):
|
|
ep = self.ep
|
|
ep.transcripts = [{'url': 'https://example.org/t.vtt', 'type': 'text/vtt'}]
|
|
ep.save()
|
|
EpisodeTranscript.objects.create(
|
|
episode=ep, sources=['https://example.org/t.vtt'], error=error,
|
|
data=None if error else {'segments': [{'start': 0, 'end': 1, 'speaker': 'A', 'text': 'Aus dem Feed'}],
|
|
'source': {'url': 'https://example.org/t.vtt'}})
|
|
|
|
def test_the_feeds_transcript_wins(self):
|
|
self.done_job(self.ep)
|
|
self.feed_transcript()
|
|
self.assertEqual(self.transcript().json()['segments'][0]['text'], 'Aus dem Feed')
|
|
self.assertEqual(self.detail()['transcript_origin'], 'feed')
|
|
|
|
def test_an_unusable_feed_transcript_gives_way(self):
|
|
self.done_job(self.ep)
|
|
self.feed_transcript(error='invalid')
|
|
self.assertEqual(self.transcript().json()['segments'][0]['text'], 'Hallo.')
|
|
self.assertEqual(self.detail()['transcript_origin'], 'whisper')
|
|
|
|
def test_a_feed_transcript_that_cannot_be_loaded_now_gives_way_too(self):
|
|
self.done_job(self.ep)
|
|
self.ep.transcripts = [{'url': 'https://example.org/t.vtt', 'type': 'text/vtt'}]
|
|
self.ep.save()
|
|
from . import chapters
|
|
with patch('podcasts.views.transcripts_mod.load', side_effect=chapters.FetchUnavailable('x')):
|
|
self.assertEqual(self.transcript().json()['segments'][0]['text'], 'Hallo.')
|
|
|
|
def test_article_item_id(self):
|
|
source = Source.objects.create(user=self.user, url='diora://podcast-artikel', title='Podcast-Artikel',
|
|
kind='blog')
|
|
item = Item.objects.create(source=source, user=self.user, kind='blog', guid_hash='x', title='A',
|
|
published=timezone.now())
|
|
EpisodeArticle.objects.create(episode=self.ep, item=item)
|
|
self.assertEqual(self.detail()['article_item_id'], item.id)
|
|
|
|
|
|
class WishTests(EndpointBase):
|
|
PENDING = '/podcasts/transcription/pending/'
|
|
|
|
def claim(self, ep=None, method='post', **kw):
|
|
return getattr(self.client, method)(
|
|
f'/podcasts/episodes/{(ep or self.ep).id}/article/claim/', json.dumps(kw.pop('body', {})),
|
|
content_type='application/json', **kw)
|
|
|
|
def test_pending_lists_open_wishes_only(self):
|
|
wished = self.done_job(self.ep, want_article=True)
|
|
e2, e3, e4, e5 = (self.episode(g) for g in ('e2', 'e3', 'e4', 'e5'))
|
|
self.done_job(e2) # no wish
|
|
TranscriptionJob.objects.create(episode=e3, state='transcribing', want_article=True) # not done
|
|
self.done_job(e4, want_article=True, claimed_by='tablet', claimed_until=timezone.now() + timedelta(minutes=5))
|
|
job5 = self.done_job(e5, want_article=True)
|
|
source = Source.objects.create(user=self.user, url='diora://podcast-artikel', title='P', kind='blog')
|
|
item = Item.objects.create(source=source, user=self.user, kind='blog', guid_hash='x', title='A',
|
|
published=timezone.now())
|
|
EpisodeArticle.objects.create(episode=e5, item=item) # written already
|
|
r = self.client.get(self.PENDING).json()
|
|
self.assertEqual(r, [{'episode_id': self.ep.id, 'title': 'Folge e1', 'podcast': 'Der Podcast',
|
|
'duration_seconds': 600}])
|
|
self.assertTrue(wished.want_article and job5.want_article)
|
|
|
|
def test_pending_is_per_user_and_needs_login(self):
|
|
self.done_job(self.ep, want_article=True)
|
|
other = User.objects.create_user(username='bob', password='pw12345678')
|
|
self.client.force_login(other)
|
|
self.assertEqual(self.client.get(self.PENDING).json(), [])
|
|
self.client.logout()
|
|
self.assertEqual(self.client.get(self.PENDING).status_code, 401)
|
|
|
|
def test_an_expired_claim_is_open_again(self):
|
|
self.done_job(self.ep, want_article=True, claimed_by='tablet',
|
|
claimed_until=timezone.now() - timedelta(seconds=1))
|
|
self.assertEqual(len(self.client.get(self.PENDING).json()), 1)
|
|
|
|
def test_claim_blocks_the_other_device_until_it_expires(self):
|
|
job = self.done_job(self.ep, want_article=True)
|
|
r = self.claim(body={'device': 'phone'})
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertEqual(r.json()['claimed_by'], 'phone')
|
|
job.refresh_from_db()
|
|
left = job.claimed_until - timezone.now()
|
|
self.assertTrue(timedelta(minutes=14) < left <= timedelta(minutes=15))
|
|
self.assertEqual(self.client.get(self.PENDING).json(), [])
|
|
r = self.claim(body={'device': 'desktop'})
|
|
self.assertEqual(r.status_code, 409)
|
|
self.assertEqual(r.json()['error'], 'claimed')
|
|
self.assertEqual(self.claim(body={'device': 'phone'}).status_code, 200) # the holder renews
|
|
TranscriptionJob.objects.filter(pk=job.pk).update(claimed_until=timezone.now() - timedelta(seconds=1))
|
|
self.assertEqual(self.claim(body={'device': 'desktop'}).status_code, 200)
|
|
job.refresh_from_db()
|
|
self.assertEqual(job.claimed_by, 'desktop')
|
|
|
|
def test_the_device_may_come_in_a_header(self):
|
|
self.done_job(self.ep, want_article=True)
|
|
self.assertEqual(self.claim(HTTP_X_DIORA_DEVICE='phone').status_code, 200)
|
|
self.assertEqual(self.claim(HTTP_X_DIORA_DEVICE='desktop').status_code, 409)
|
|
|
|
def test_a_device_name_is_required(self):
|
|
self.done_job(self.ep, want_article=True)
|
|
self.assertEqual(self.claim().status_code, 400)
|
|
|
|
def test_release(self):
|
|
self.done_job(self.ep, want_article=True)
|
|
self.claim(body={'device': 'phone'})
|
|
self.assertEqual(self.claim(method='delete', body={'device': 'desktop'}).status_code, 409)
|
|
self.assertEqual(self.claim(method='delete', body={'device': 'phone'}).status_code, 200)
|
|
self.assertEqual(len(self.client.get(self.PENDING).json()), 1)
|
|
|
|
def test_no_wish_no_claim(self):
|
|
self.done_job(self.ep) # done, but no article wanted
|
|
self.assertEqual(self.claim(body={'device': 'phone'}).status_code, 404)
|
|
self.assertEqual(self.claim(self.episode('e9'), body={'device': 'phone'}).status_code, 404)
|
|
|
|
|
|
class NotificationTests(EndpointBase):
|
|
def get(self, **q):
|
|
return self.client.get('/feeds/notifications/', q).json()
|
|
|
|
def test_first_time_only_the_cursor(self):
|
|
j = self.done_job(self.ep, want_article=True)
|
|
d = self.get()
|
|
self.assertEqual((d['transcripts'], d['t_last'], d['items']), ([], j.id, []))
|
|
self.assertEqual(d['last'], 0)
|
|
|
|
def test_jobs_done_after_the_cursor(self):
|
|
e2, e3 = self.episode('e2'), self.episode('e3')
|
|
a = self.done_job(self.ep)
|
|
b = self.done_job(e2, want_article=True)
|
|
TranscriptionJob.objects.create(episode=e3, state='transcribing') # not done: no notice
|
|
d = self.get(t_after=a.id)
|
|
self.assertEqual(d['transcripts'], [{'job_id': b.id, 'episode_id': e2.id, 'title': 'Folge e2',
|
|
'podcast': 'Der Podcast', 'want_article': True}])
|
|
self.assertEqual(d['t_last'], b.id)
|
|
self.assertEqual(self.get(t_after=b.id)['transcripts'], [])
|
|
self.assertEqual([t['job_id'] for t in self.get(t_after=0)['transcripts']], [a.id, b.id])
|
|
|
|
def test_only_the_users_own_and_not_feed_only_jobs(self):
|
|
other = User.objects.create_user(username='bob', password='pw12345678')
|
|
feed = PodcastFeed.objects.create(user=other, rss_url='https://example.org/o', title='O')
|
|
ep = PodcastEpisode.objects.create(feed=feed, guid='o', title='O', audio_url='https://example.org/o.mp3')
|
|
self.done_job(ep)
|
|
self.done_job(self.episode('e2'), data=None) # only the wish for an article from a feed transcript
|
|
d = self.get(t_after=0)
|
|
self.assertEqual((d['transcripts'], d['t_last']), ([], 0))
|
|
|
|
def test_items_still_work_beside_it(self):
|
|
d = self.get(after=0)
|
|
self.assertEqual((d['items'], d['last'], d['transcripts'], d['t_last']), ([], 0, [], 0))
|