| """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() |
|
|
