File size: 4,209 Bytes
a4e4eb4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 | """Handler regression tests; no model weights or endpoint required."""
import unittest
from types import SimpleNamespace
from unittest.mock import Mock
from handler import EndpointHandler
class LanguageTests(unittest.TestCase):
def setUp(self):
self.handler = EndpointHandler.__new__(EndpointHandler)
self.handler.device = "cpu"
self.handler.torch_dtype = None
self.handler.model = Mock()
self.handler.model.generation_config = SimpleNamespace(language=None)
self.features = Mock()
self.features.to.return_value = self.features
self.handler.processor = Mock(return_value=SimpleNamespace(input_features=self.features))
self.handler.processor.batch_decode.return_value = [" test transcription "]
self.handler._load_audio_from_base64 = Mock(return_value=[0.0] * 16000)
self.handler._load_audio_from_bytes = Mock(return_value=[0.0] * 16000)
def assert_success(self, request):
self.assertEqual(self.handler(request), {
"transcription": "test transcription",
"audio_duration_seconds": 1.0,
"model": "AImpower/StutteredSpeechASR",
})
def test_languages_and_json_audio_formats(self):
for language in ("en", "english", "<|en|>", "zh", " en "):
for audio in ({"inputs": "audio"}, {"inputs": {"audio": "audio"}},
{"audio": "audio"}, {"inputs": b"audio"}):
with self.subTest(language=language, audio=audio):
self.assert_success({**audio, "parameters": {"language": language}})
args, kwargs = self.handler.model.generate.call_args
self.assertEqual(args, (self.features,))
self.assertEqual(kwargs["language"], language.strip())
self.assertEqual(kwargs["task"], "transcribe")
self.assertIsNot(kwargs["generation_config"], self.handler.model.generation_config)
def test_default_requests_keep_existing_generation_call(self):
for request in ({"inputs": "audio"}, {"inputs": "audio", "parameters": {}},
{"inputs": "audio", "parameters": {"language": None}},
b"audio", bytearray(b"audio")):
with self.subTest(request=request):
self.assert_success(request)
self.handler.model.generate.assert_called_with(self.features)
def test_language_does_not_leak_between_requests(self):
def generate(features, **kwargs):
if "generation_config" in kwargs:
kwargs["generation_config"].language = kwargs["language"]
return [[1]]
self.handler.model.generate.side_effect = generate
for language in ("en", "zh"):
self.assert_success({"inputs": "audio", "parameters": {"language": language}})
self.assertIsNone(self.handler.model.generation_config.language)
self.assert_success({"inputs": "audio"})
self.handler.model.generate.assert_called_with(self.features)
def test_invalid_parameter_shapes_return_existing_error_format(self):
for parameters in (None, [], "en", {"language": ""}, {"language": " "},
{"language": 123}, {"language": ["en"]}):
with self.subTest(parameters=parameters):
result = self.handler({"inputs": "audio", "parameters": parameters})
self.assertIsNone(result["transcription"])
self.assertIn("error", result)
self.handler.model.generate.assert_not_called()
def test_unsupported_language_error_is_returned(self):
self.handler.model.generate.side_effect = ValueError("Unsupported language: invalid")
result = self.handler({"inputs": "audio", "parameters": {"language": "invalid"}})
self.assertEqual(result, {"error": "Unsupported language: invalid", "transcription": None})
def test_unrelated_parameters_are_not_forwarded(self):
self.assert_success({"inputs": "audio", "parameters": {"max_new_tokens": 1}})
self.handler.model.generate.assert_called_with(self.features)
if __name__ == "__main__":
unittest.main()
|