Spaces:
Build error
Build error
| """Unit tests for ISpeechSynthesisService interface contract.""" | |
| import pytest | |
| from abc import ABC | |
| from unittest.mock import Mock | |
| from typing import Iterator | |
| from src.domain.interfaces.speech_synthesis import ISpeechSynthesisService | |
| from src.domain.models.speech_synthesis_request import SpeechSynthesisRequest | |
| from src.domain.models.audio_content import AudioContent | |
| from src.domain.models.audio_chunk import AudioChunk | |
| from src.domain.models.text_content import TextContent | |
| from src.domain.models.voice_settings import VoiceSettings | |
| class TestISpeechSynthesisService: | |
| """Test cases for ISpeechSynthesisService interface contract.""" | |
| def test_interface_is_abstract(self): | |
| """Test that ISpeechSynthesisService is an abstract base class.""" | |
| assert issubclass(ISpeechSynthesisService, ABC) | |
| # Should not be able to instantiate directly | |
| with pytest.raises(TypeError): | |
| ISpeechSynthesisService() # type: ignore | |
| def test_interface_has_required_methods(self): | |
| """Test that interface defines the required abstract methods.""" | |
| # Check that both methods exist and are abstract | |
| assert hasattr(ISpeechSynthesisService, 'synthesize') | |
| assert hasattr(ISpeechSynthesisService, 'synthesize_stream') | |
| assert getattr(ISpeechSynthesisService.synthesize, '__isabstractmethod__', False) | |
| assert getattr(ISpeechSynthesisService.synthesize_stream, '__isabstractmethod__', False) | |
| def test_synthesize_method_signature(self): | |
| """Test that the synthesize method has the correct signature.""" | |
| import inspect | |
| method = ISpeechSynthesisService.synthesize | |
| signature = inspect.signature(method) | |
| # Check parameter names | |
| params = list(signature.parameters.keys()) | |
| expected_params = ['self', 'request'] | |
| assert params == expected_params | |
| # Check return annotation | |
| assert signature.return_annotation == "'AudioContent'" | |
| def test_synthesize_stream_method_signature(self): | |
| """Test that the synthesize_stream method has the correct signature.""" | |
| import inspect | |
| method = ISpeechSynthesisService.synthesize_stream | |
| signature = inspect.signature(method) | |
| # Check parameter names | |
| params = list(signature.parameters.keys()) | |
| expected_params = ['self', 'request'] | |
| assert params == expected_params | |
| # Check return annotation | |
| assert signature.return_annotation == "Iterator['AudioChunk']" | |
| def test_concrete_implementation_must_implement_methods(self): | |
| """Test that concrete implementations must implement both abstract methods.""" | |
| class IncompleteImplementation(ISpeechSynthesisService): | |
| def synthesize(self, request): | |
| return AudioContent(data=b"test", format="wav", sample_rate=22050, duration=1.0) | |
| # Missing synthesize_stream method | |
| # Should not be able to instantiate without implementing all abstract methods | |
| with pytest.raises(TypeError, match="Can't instantiate abstract class"): | |
| IncompleteImplementation() # type: ignore | |
| def test_concrete_implementation_with_both_methods(self): | |
| """Test that concrete implementation with both methods can be instantiated.""" | |
| class ConcreteImplementation(ISpeechSynthesisService): | |
| def synthesize(self, request): | |
| return AudioContent(data=b"synthesized", format="wav", sample_rate=22050, duration=1.0) | |
| def synthesize_stream(self, request): | |
| yield AudioChunk(data=b"chunk1", format="wav", sample_rate=22050, chunk_index=0, is_final=True) | |
| # Should be able to instantiate | |
| implementation = ConcreteImplementation() | |
| assert isinstance(implementation, ISpeechSynthesisService) | |
| def test_synthesize_method_contract_with_mock(self): | |
| """Test the synthesize method contract using a mock implementation.""" | |
| class MockImplementation(ISpeechSynthesisService): | |
| def __init__(self): | |
| self.mock_synthesize = Mock() | |
| self.mock_synthesize_stream = Mock() | |
| def synthesize(self, request): | |
| return self.mock_synthesize(request) | |
| def synthesize_stream(self, request): | |
| return self.mock_synthesize_stream(request) | |
| # Create test data | |
| text_content = TextContent(text="Hello world", language="en") | |
| voice_settings = VoiceSettings(voice_id="test_voice", speed=1.0, language="en") | |
| request = SpeechSynthesisRequest( | |
| text_content=text_content, | |
| voice_settings=voice_settings | |
| ) | |
| expected_result = AudioContent( | |
| data=b"synthesized_audio", | |
| format="wav", | |
| sample_rate=22050, | |
| duration=2.0 | |
| ) | |
| # Setup mock | |
| implementation = MockImplementation() | |
| implementation.mock_synthesize.return_value = expected_result | |
| # Call method | |
| result = implementation.synthesize(request) | |
| # Verify call and result | |
| implementation.mock_synthesize.assert_called_once_with(request) | |
| assert result == expected_result | |
| def test_synthesize_stream_method_contract_with_mock(self): | |
| """Test the synthesize_stream method contract using a mock implementation.""" | |
| class MockImplementation(ISpeechSynthesisService): | |
| def __init__(self): | |
| self.mock_synthesize = Mock() | |
| self.mock_synthesize_stream = Mock() | |
| def synthesize(self, request): | |
| return self.mock_synthesize(request) | |
| def synthesize_stream(self, request): | |
| return self.mock_synthesize_stream(request) | |
| # Create test data | |
| text_content = TextContent(text="Hello world", language="en") | |
| voice_settings = VoiceSettings(voice_id="test_voice", speed=1.0, language="en") | |
| request = SpeechSynthesisRequest( | |
| text_content=text_content, | |
| voice_settings=voice_settings | |
| ) | |
| expected_chunks = [ | |
| AudioChunk(data=b"chunk1", format="wav", sample_rate=22050, chunk_index=0), | |
| AudioChunk(data=b"chunk2", format="wav", sample_rate=22050, chunk_index=1, is_final=True) | |
| ] | |
| # Setup mock | |
| implementation = MockImplementation() | |
| implementation.mock_synthesize_stream.return_value = iter(expected_chunks) | |
| # Call method | |
| result = implementation.synthesize_stream(request) | |
| # Verify call and result | |
| implementation.mock_synthesize_stream.assert_called_once_with(request) | |
| chunks = list(result) | |
| assert chunks == expected_chunks | |
| def test_interface_docstring_requirements(self): | |
| """Test that the interface methods have proper documentation.""" | |
| synthesize_method = ISpeechSynthesisService.synthesize | |
| stream_method = ISpeechSynthesisService.synthesize_stream | |
| # Check synthesize method docstring | |
| assert synthesize_method.__doc__ is not None | |
| synthesize_doc = synthesize_method.__doc__ | |
| assert "Synthesize speech from text" in synthesize_doc | |
| assert "Args:" in synthesize_doc | |
| assert "Returns:" in synthesize_doc | |
| assert "Raises:" in synthesize_doc | |
| assert "SpeechSynthesisException" in synthesize_doc | |
| # Check synthesize_stream method docstring | |
| assert stream_method.__doc__ is not None | |
| stream_doc = stream_method.__doc__ | |
| assert "Synthesize speech from text as a stream" in stream_doc | |
| assert "Args:" in stream_doc | |
| assert "Returns:" in stream_doc | |
| assert "Iterator[AudioChunk]" in stream_doc | |
| assert "Raises:" in stream_doc | |
| assert "SpeechSynthesisException" in stream_doc | |
| def test_interface_type_hints(self): | |
| """Test that the interface uses proper type hints.""" | |
| synthesize_method = ISpeechSynthesisService.synthesize | |
| stream_method = ISpeechSynthesisService.synthesize_stream | |
| # Check synthesize method annotations | |
| synthesize_annotations = getattr(synthesize_method, '__annotations__', {}) | |
| assert 'request' in synthesize_annotations | |
| assert 'return' in synthesize_annotations | |
| assert synthesize_annotations['request'] == "'SpeechSynthesisRequest'" | |
| assert synthesize_annotations['return'] == "'AudioContent'" | |
| # Check synthesize_stream method annotations | |
| stream_annotations = getattr(stream_method, '__annotations__', {}) | |
| assert 'request' in stream_annotations | |
| assert 'return' in stream_annotations | |
| assert stream_annotations['request'] == "'SpeechSynthesisRequest'" | |
| assert stream_annotations['return'] == "Iterator['AudioChunk']" | |
| def test_multiple_implementations_possible(self): | |
| """Test that multiple implementations of the interface are possible.""" | |
| class KokoroImplementation(ISpeechSynthesisService): | |
| def synthesize(self, request): | |
| return AudioContent(data=b"kokoro_audio", format="wav", sample_rate=22050, duration=1.0) | |
| def synthesize_stream(self, request): | |
| yield AudioChunk(data=b"kokoro_chunk", format="wav", sample_rate=22050, chunk_index=0, is_final=True) | |
| class DiaImplementation(ISpeechSynthesisService): | |
| def synthesize(self, request): | |
| return AudioContent(data=b"dia_audio", format="wav", sample_rate=22050, duration=1.0) | |
| def synthesize_stream(self, request): | |
| yield AudioChunk(data=b"dia_chunk", format="wav", sample_rate=22050, chunk_index=0, is_final=True) | |
| kokoro = KokoroImplementation() | |
| dia = DiaImplementation() | |
| assert isinstance(kokoro, ISpeechSynthesisService) | |
| assert isinstance(dia, ISpeechSynthesisService) | |
| assert type(kokoro) != type(dia) | |
| def test_interface_methods_can_be_called_polymorphically(self): | |
| """Test that interface methods can be called polymorphically.""" | |
| class TestImplementation(ISpeechSynthesisService): | |
| def __init__(self, audio_data, chunk_data): | |
| self.audio_data = audio_data | |
| self.chunk_data = chunk_data | |
| def synthesize(self, request): | |
| return AudioContent(data=self.audio_data, format="wav", sample_rate=22050, duration=1.0) | |
| def synthesize_stream(self, request): | |
| yield AudioChunk(data=self.chunk_data, format="wav", sample_rate=22050, chunk_index=0, is_final=True) | |
| # Create different implementations | |
| implementations = [ | |
| TestImplementation(b"audio1", b"chunk1"), | |
| TestImplementation(b"audio2", b"chunk2") | |
| ] | |
| # Test polymorphic usage | |
| text_content = TextContent(text="test", language="en") | |
| voice_settings = VoiceSettings(voice_id="test", speed=1.0, language="en") | |
| request = SpeechSynthesisRequest(text_content=text_content, voice_settings=voice_settings) | |
| # Test synthesize method | |
| audio_results = [] | |
| for impl in implementations: | |
| result = impl.synthesize(request) | |
| audio_results.append(result.data) | |
| assert audio_results == [b"audio1", b"audio2"] | |
| # Test synthesize_stream method | |
| chunk_results = [] | |
| for impl in implementations: | |
| chunks = list(impl.synthesize_stream(request)) | |
| chunk_results.append(chunks[0].data) | |
| assert chunk_results == [b"chunk1", b"chunk2"] | |
| def test_interface_inheritance_chain(self): | |
| """Test the inheritance chain of the interface.""" | |
| # Check that it inherits from ABC | |
| assert ABC in ISpeechSynthesisService.__mro__ | |
| # Check that it's at the right position in MRO | |
| mro = ISpeechSynthesisService.__mro__ | |
| assert mro[0] == ISpeechSynthesisService | |
| assert ABC in mro | |
| def test_stream_method_returns_iterator(self): | |
| """Test that synthesize_stream returns an iterator.""" | |
| class StreamingImplementation(ISpeechSynthesisService): | |
| def synthesize(self, request): | |
| return AudioContent(data=b"audio", format="wav", sample_rate=22050, duration=1.0) | |
| def synthesize_stream(self, request): | |
| for i in range(3): | |
| yield AudioChunk( | |
| data=f"chunk{i}".encode(), | |
| format="wav", | |
| sample_rate=22050, | |
| chunk_index=i, | |
| is_final=(i == 2) | |
| ) | |
| impl = StreamingImplementation() | |
| text_content = TextContent(text="test", language="en") | |
| voice_settings = VoiceSettings(voice_id="test", speed=1.0, language="en") | |
| request = SpeechSynthesisRequest(text_content=text_content, voice_settings=voice_settings) | |
| # Get the iterator | |
| stream = impl.synthesize_stream(request) | |
| # Verify it's an iterator | |
| assert hasattr(stream, '__iter__') | |
| assert hasattr(stream, '__next__') | |
| # Verify we can iterate through chunks | |
| chunks = list(stream) | |
| assert len(chunks) == 3 | |
| for i, chunk in enumerate(chunks): | |
| assert chunk.data == f"chunk{i}".encode() | |
| assert chunk.chunk_index == i | |
| assert chunk.is_final == (i == 2) | |
| def test_implementation_can_handle_different_formats(self): | |
| """Test that implementations can handle different output formats.""" | |
| class MultiFormatImplementation(ISpeechSynthesisService): | |
| def synthesize(self, request): | |
| format_data = { | |
| "wav": b"wav_audio_data", | |
| "mp3": b"mp3_audio_data", | |
| "flac": b"flac_audio_data", | |
| "ogg": b"ogg_audio_data" | |
| } | |
| audio_data = format_data.get(request.output_format, b"default_audio_data") | |
| return AudioContent( | |
| data=audio_data, | |
| format=request.output_format, | |
| sample_rate=request.effective_sample_rate, | |
| duration=1.0 | |
| ) | |
| def synthesize_stream(self, request): | |
| yield AudioChunk( | |
| data=f"{request.output_format}_chunk".encode(), | |
| format=request.output_format, | |
| sample_rate=request.effective_sample_rate, | |
| chunk_index=0, | |
| is_final=True | |
| ) | |
| impl = MultiFormatImplementation() | |
| text_content = TextContent(text="test", language="en") | |
| voice_settings = VoiceSettings(voice_id="test", speed=1.0, language="en") | |
| # Test different formats | |
| formats = ["wav", "mp3", "flac", "ogg"] | |
| for fmt in formats: | |
| request = SpeechSynthesisRequest( | |
| text_content=text_content, | |
| voice_settings=voice_settings, | |
| output_format=fmt | |
| ) | |
| # Test synthesize | |
| audio = impl.synthesize(request) | |
| assert audio.format == fmt | |
| assert audio.data == f"{fmt}_audio_data".encode() | |
| # Test synthesize_stream | |
| chunks = list(impl.synthesize_stream(request)) | |
| assert len(chunks) == 1 | |
| assert chunks[0].format == fmt | |
| assert chunks[0].data == f"{fmt}_chunk".encode() |