import io import json import wave from unittest.mock import MagicMock, patch from django.contrib.auth.models import User from django.test import TestCase, override_settings from feeds import logins from . import chapters, transcription from .models import TranscriptionSettings SETTINGS = '/podcasts/transcription/settings/' TEST = '/podcasts/transcription/test/' def fake_response(status=200): r = MagicMock() r.status_code = status r.headers = {} r.raw = None # nothing to introspect for the peer check return r class SettingsTests(TestCase): def setUp(self): self.user = User.objects.create_user(username='alice', password='pw12345678') self.client.force_login(self.user) def post(self, data, url=SETTINGS): return self.client.post(url, json.dumps(data), content_type='application/json') def test_requires_login(self): self.client.logout() self.assertEqual(self.client.get(SETTINGS).status_code, 401) self.assertEqual(self.post({}, TEST).status_code, 401) def test_defaults(self): self.assertEqual(self.client.get(SETTINGS).json(), { 'engine': 'builtin', 'url': '', 'model': '', 'language': '', 'parallel': 3, 'has_token': False, 'builtin_available': False}) self.assertFalse(TranscriptionSettings.objects.exists()) @override_settings(WHISPER_URL='http://whisper:8000/v1') def test_builtin_available_follows_the_environment_setting(self): self.assertTrue(self.client.get(SETTINGS).json()['builtin_available']) def test_roundtrip(self): r = self.post({'engine': 'custom', 'url': 'https://whisper.example.org/v1/', 'model': 'large-v3', 'language': 'de', 'parallel': 5, 'token': 'sk-secret'}) self.assertEqual(r.status_code, 200) want = {'engine': 'custom', 'url': 'https://whisper.example.org/v1', 'model': 'large-v3', 'language': 'de', 'parallel': 5, 'has_token': True, 'builtin_available': False} self.assertEqual(r.json(), want) self.assertEqual(self.client.get(SETTINGS).json(), want) def test_token_is_write_only_and_stored_encrypted(self): self.post({'engine': 'custom', 'url': 'https://w.example.org/v1', 'token': 'sk-secret'}) conf = TranscriptionSettings.objects.get(user=self.user) self.assertTrue(conf.token_encrypted) self.assertNotIn('sk-secret', conf.token_encrypted) self.assertEqual(transcription.token_of(conf), 'sk-secret') for body in (self.client.get(SETTINGS).content, self.post({'model': 'm'}).content): self.assertNotIn(b'sk-secret', body) self.assertNotIn(conf.token_encrypted.encode(), body) # Not sending a token keeps it; an empty one removes it. conf.refresh_from_db() self.assertTrue(conf.token_encrypted) self.assertFalse(self.post({'token': ''}).json()['has_token']) conf.refresh_from_db() self.assertEqual(conf.token_encrypted, '') def test_validation(self): bad = [ {'engine': 'cloud'}, {'parallel': 0}, {'parallel': 9}, {'parallel': '3'}, {'parallel': True}, {'url': 'ftp://x.org/v1'}, {'url': 'javascript:alert(1)'}, {'url': 'whisper.local/v1'}, {'language': 'Deutsch'}, {'model': 5}, {'token': 5}, {'engine': 'custom'}, # a custom instance needs a url ] for body in bad: self.assertEqual(self.post(body).status_code, 400, body) self.assertEqual(self.client.post(SETTINGS, 'nope', content_type='application/json').status_code, 400) self.assertFalse(TranscriptionSettings.objects.exists()) for body in ({'parallel': 1}, {'parallel': 8}, {'language': ''}, {'language': 'pt-BR'}): self.assertEqual(self.post(body).status_code, 200, body) def test_settings_are_per_user(self): self.post({'parallel': 7}) other = User.objects.create_user(username='bob', password='pw12345678') self.client.force_login(other) self.assertEqual(self.client.get(SETTINGS).json()['parallel'], 3) class TestEndpointTests(TestCase): def setUp(self): self.user = User.objects.create_user(username='alice', password='pw12345678') self.client.force_login(self.user) def configure(self, **kw): TranscriptionSettings.objects.update_or_create(user=self.user, defaults=kw) def test_custom_sends_a_second_of_silence(self): self.configure(engine='custom', url='https://whisper.example.org/v1', model='large-v3', language='de', token_encrypted=logins.encrypt('sk-secret')) calls = [] def post(url, **kw): calls.append((url, kw)) return fake_response() with patch('podcasts.chapters.assert_safe_url'), patch('podcasts.chapters.requests.post', post): r = self.client.post(TEST) self.assertEqual(r.json(), {'ok': True, 'model': 'large-v3', 'error': ''}) url, kw = calls[0] self.assertEqual(url, 'https://whisper.example.org/v1/audio/transcriptions') self.assertEqual(kw['headers']['Authorization'], 'Bearer sk-secret') self.assertEqual(kw['data']['model'], 'large-v3') self.assertEqual(kw['data']['language'], 'de') name, wav, mime = kw['files']['file'] self.assertEqual(mime, 'audio/wav') with wave.open(io.BytesIO(wav)) as w: self.assertEqual((w.getframerate(), w.getnchannels(), w.getnframes()), (16000, 1, 16000)) def test_custom_private_address_is_refused(self): self.configure(engine='custom', url='http://127.0.0.1:9000/v1') with patch('podcasts.chapters.requests.post') as post: r = self.client.post(TEST) self.assertFalse(r.json()['ok']) self.assertTrue(r.json()['error']) post.assert_not_called() def test_custom_http_error_is_reported(self): self.configure(engine='custom', url='https://whisper.example.org/v1') with patch('podcasts.chapters.assert_safe_url'), \ patch('podcasts.chapters.requests.post', return_value=fake_response(401)): r = self.client.post(TEST) self.assertEqual((r.json()['ok'], r.json()['error']), (False, 'HTTP 401')) @override_settings(WHISPER_URL='http://whisper:8000/v1') def test_builtin_is_exempt_from_the_address_check(self): self.configure(engine='builtin') with patch('podcasts.transcription.requests.post', return_value=fake_response()) as post, \ patch('podcasts.chapters.assert_safe_url', side_effect=chapters.WebDAVInputError('private')): r = self.client.post(TEST) self.assertTrue(r.json()['ok']) self.assertEqual(post.call_args.args[0], 'http://whisper:8000/v1/audio/transcriptions') self.assertNotIn('Authorization', post.call_args.kwargs['headers']) def test_builtin_without_instance(self): r = self.client.post(TEST) self.assertFalse(r.json()['ok']) self.assertTrue(r.json()['error'])