From 0f568120f49edbd81c5e621a6f7868631219de1c Mon Sep 17 00:00:00 2001 From: Artem Akymenko Date: Thu, 9 Jul 2026 08:52:04 +0000 Subject: [PATCH] feat: add contract test suite for Plugin API Create reusable contract tests for TTS Plugin Architecture: - conftest.py: shared fixtures and stubs (FakeEngine, FakeSession, etc.) - test_types_contract.py: value object contracts (frozen, immutability, equality) - test_errors_contract.py: error hierarchy contracts - test_manifest_contract.py: manifest type contracts - test_engine_contract.py: Engine protocol contracts (lifecycle, dispose) - test_session_contract.py: EngineSession protocol contracts - test_capabilities_contract.py: capability protocol contracts - test_host_context_contract.py: HostContext contracts - test_plugin_contract.py: plugin contract (exports, create_engine) 124 tests covering all public API contracts. --- tests/contracts/__init__.py | 5 + tests/contracts/conftest.py | 231 ++++++++++++++ tests/contracts/test_capabilities_contract.py | 183 +++++++++++ tests/contracts/test_engine_contract.py | 106 +++++++ tests/contracts/test_errors_contract.py | 85 +++++ tests/contracts/test_host_context_contract.py | 89 ++++++ tests/contracts/test_manifest_contract.py | 290 ++++++++++++++++++ tests/contracts/test_plugin_contract.py | 146 +++++++++ tests/contracts/test_session_contract.py | 135 ++++++++ tests/contracts/test_types_contract.py | 207 +++++++++++++ 10 files changed, 1477 insertions(+) create mode 100644 tests/contracts/__init__.py create mode 100644 tests/contracts/conftest.py create mode 100644 tests/contracts/test_capabilities_contract.py create mode 100644 tests/contracts/test_engine_contract.py create mode 100644 tests/contracts/test_errors_contract.py create mode 100644 tests/contracts/test_host_context_contract.py create mode 100644 tests/contracts/test_manifest_contract.py create mode 100644 tests/contracts/test_plugin_contract.py create mode 100644 tests/contracts/test_session_contract.py create mode 100644 tests/contracts/test_types_contract.py diff --git a/tests/contracts/__init__.py b/tests/contracts/__init__.py new file mode 100644 index 0000000..28b505e --- /dev/null +++ b/tests/contracts/__init__.py @@ -0,0 +1,5 @@ +"""Contract tests for the TTS Plugin API. + +This package contains reusable contract tests that any TTS plugin implementation +must satisfy. Tests use only the public API and are engine-agnostic. +""" diff --git a/tests/contracts/conftest.py b/tests/contracts/conftest.py new file mode 100644 index 0000000..d2b80cb --- /dev/null +++ b/tests/contracts/conftest.py @@ -0,0 +1,231 @@ +"""Shared fixtures and stubs for contract tests. + +This module provides minimal stub implementations that satisfy the public API +for testing purposes. These stubs do NOT contain real business logic. +""" + +from __future__ import annotations + +import logging +from pathlib import Path +from typing import Iterator + +import pytest + +from abogen.tts_plugin.engine import Engine, EngineSession +from abogen.tts_plugin.host_context import HostContext +from abogen.tts_plugin.types import ( + AudioFormat, + Duration, + EngineConfig, + ParameterValues, + SynthesisRequest, + SynthesizedAudio, + VoiceSelection, +) + + +class FakeHttpClient: + """Stub HTTP client that satisfies the HttpClient protocol.""" + + def get(self, url: str, **kwargs: object) -> object: + return None + + def post(self, url: str, **kwargs: object) -> object: + return None + + +class FakeEngineSession: + """Stub EngineSession for testing protocol compliance.""" + + def __init__(self) -> None: + self._disposed = False + + def synthesize(self, request: SynthesisRequest) -> SynthesizedAudio: + if self._disposed: + from abogen.tts_plugin.errors import EngineError + + raise EngineError("Session disposed") + return SynthesizedAudio( + data=b"\x00" * 100, + format=AudioFormat(mime="audio/wav", extension="wav"), + duration=Duration(seconds=1.0), + ) + + def dispose(self) -> None: + self._disposed = True + + +class FakeStreamingSession: + """Stub EngineSession with StreamingSynthesizer capability.""" + + def __init__(self) -> None: + self._disposed = False + + def synthesize(self, request: SynthesisRequest) -> SynthesizedAudio: + if self._disposed: + from abogen.tts_plugin.errors import EngineError + + raise EngineError("Session disposed") + return SynthesizedAudio( + data=b"\x00" * 100, + format=AudioFormat(mime="audio/wav", extension="wav"), + duration=Duration(seconds=1.0), + ) + + def synthesizeStream(self, request: SynthesisRequest) -> Iterator[bytes]: + if self._disposed: + from abogen.tts_plugin.errors import EngineError + + raise EngineError("Session disposed") + for i in range(3): + yield b"\x00" * 50 + + def dispose(self) -> None: + self._disposed = True + + +class FakeCancelableSession: + """Stub EngineSession with CancelableSession capability.""" + + def __init__(self) -> None: + self._disposed = False + self._cancelled = False + + def synthesize(self, request: SynthesisRequest) -> SynthesizedAudio: + if self._disposed: + from abogen.tts_plugin.errors import EngineError + + raise EngineError("Session disposed") + if self._cancelled: + from abogen.tts_plugin.errors import CancelledError + + raise CancelledError("Cancelled") + return SynthesizedAudio( + data=b"\x00" * 100, + format=AudioFormat(mime="audio/wav", extension="wav"), + duration=Duration(seconds=1.0), + ) + + def cancel(self) -> None: + if self._disposed: + from abogen.tts_plugin.errors import EngineError + + raise EngineError("Session disposed") + self._cancelled = True + + def dispose(self) -> None: + self._disposed = True + + +class FakeEngine: + """Stub Engine for testing protocol compliance.""" + + def __init__(self, session_class: type = FakeEngineSession) -> None: + self._disposed = False + self._session_class = session_class + + def createSession(self) -> EngineSession: + if self._disposed: + from abogen.tts_plugin.errors import EngineError + + raise EngineError("Engine disposed") + return self._session_class() + + def dispose(self) -> None: + self._disposed = True + + +class FakeVoiceListerEngine: + """Stub Engine that also implements VoiceLister.""" + + def __init__(self) -> None: + self._disposed = False + + def createSession(self) -> EngineSession: + if self._disposed: + from abogen.tts_plugin.errors import EngineError + + raise EngineError("Engine disposed") + return FakeEngineSession() + + def listVoices(self, sourceId: str) -> list: + from abogen.tts_plugin.manifest import VoiceManifest + + return [ + VoiceManifest(id="voice1", name="Voice 1", tags=("en",)), + VoiceManifest(id="voice2", name="Voice 2", tags=("es",)), + ] + + def dispose(self) -> None: + self._disposed = True + + +class FakePreviewEngine: + """Stub Engine that also implements PreviewGenerator.""" + + def __init__(self) -> None: + self._disposed = False + + def createSession(self) -> EngineSession: + if self._disposed: + from abogen.tts_plugin.errors import EngineError + + raise EngineError("Engine disposed") + return FakeEngineSession() + + def generatePreview(self, voice: VoiceSelection, text: str) -> SynthesizedAudio: + return SynthesizedAudio( + data=b"\x00" * 50, + format=AudioFormat(mime="audio/wav", extension="wav"), + duration=Duration(seconds=0.5), + ) + + def dispose(self) -> None: + self._disposed = True + + +@pytest.fixture +def fake_http_client() -> FakeHttpClient: + return FakeHttpClient() + + +@pytest.fixture +def host_context(tmp_path: Path, fake_http_client: FakeHttpClient) -> HostContext: + return HostContext( + config_dir=tmp_path, + logger=logging.getLogger("test"), + http_client=fake_http_client, + ) + + +@pytest.fixture +def fake_engine() -> FakeEngine: + return FakeEngine() + + +@pytest.fixture +def fake_session() -> FakeEngineSession: + return FakeEngineSession() + + +@pytest.fixture +def default_voice() -> VoiceSelection: + return VoiceSelection(source="builtin", key="af_nova") + + +@pytest.fixture +def default_format() -> AudioFormat: + return AudioFormat(mime="audio/wav", extension="wav") + + +@pytest.fixture +def default_request( + default_voice: VoiceSelection, default_format: AudioFormat +) -> SynthesisRequest: + return SynthesisRequest( + text="Hello, world!", + voice=default_voice, + parameters=ParameterValues(values={}), + format=default_format, + ) diff --git a/tests/contracts/test_capabilities_contract.py b/tests/contracts/test_capabilities_contract.py new file mode 100644 index 0000000..d1b5afa --- /dev/null +++ b/tests/contracts/test_capabilities_contract.py @@ -0,0 +1,183 @@ +"""Contract tests for capability interfaces. + +These tests verify that capability interfaces satisfy the architectural requirements: +- VoiceLister: lists voices for a source +- PreviewGenerator: generates preview audio +- StreamingSynthesizer: yields audio chunks +- CancelableSession: cancels in-progress synthesis +""" + +import pytest + +from abogen.tts_plugin.capabilities import ( + CancelableSession, + PreviewGenerator, + StreamingSynthesizer, + VoiceLister, +) +from abogen.tts_plugin.errors import CancelledError, EngineError +from abogen.tts_plugin.manifest import VoiceManifest +from abogen.tts_plugin.types import ( + AudioFormat, + Duration, + ParameterValues, + SynthesisRequest, + SynthesizedAudio, + VoiceSelection, +) + +from .conftest import FakeCancelableSession, FakeStreamingSession, FakeVoiceListerEngine + + +class TestVoiceListerProtocolContract: + """Contract tests for VoiceLister protocol.""" + + def test_voice_lister_is_protocol(self) -> None: + assert hasattr(VoiceLister, "__protocol_attrs__") + + def test_voice_lister_satisfied_by_engine(self) -> None: + engine = FakeVoiceListerEngine() + assert isinstance(engine, VoiceLister) + + def test_list_voices_returns_list(self) -> None: + engine = FakeVoiceListerEngine() + voices = engine.listVoices("builtin") + assert isinstance(voices, list) + + def test_list_voices_returns_voice_manifests(self) -> None: + engine = FakeVoiceListerEngine() + voices = engine.listVoices("builtin") + for voice in voices: + assert isinstance(voice, VoiceManifest) + + def test_list_voices_has_required_fields(self) -> None: + engine = FakeVoiceListerEngine() + voices = engine.listVoices("builtin") + for voice in voices: + assert hasattr(voice, "id") + assert hasattr(voice, "name") + assert hasattr(voice, "tags") + + +class TestPreviewGeneratorProtocolContract: + """Contract tests for PreviewGenerator protocol.""" + + def test_preview_generator_is_protocol(self) -> None: + assert hasattr(PreviewGenerator, "__protocol_attrs__") + + def test_preview_generator_satisfied_by_engine(self) -> None: + from .conftest import FakePreviewEngine + + engine = FakePreviewEngine() + assert isinstance(engine, PreviewGenerator) + + def test_generate_preview_returns_synthesized_audio(self) -> None: + from .conftest import FakePreviewEngine + + engine = FakePreviewEngine() + voice = VoiceSelection(source="builtin", key="af_nova") + result = engine.generatePreview(voice, "Hello") + assert isinstance(result, SynthesizedAudio) + + def test_generate_preview_has_valid_data(self) -> None: + from .conftest import FakePreviewEngine + + engine = FakePreviewEngine() + voice = VoiceSelection(source="builtin", key="af_nova") + result = engine.generatePreview(voice, "Hello") + assert isinstance(result.data, bytes) + assert len(result.data) > 0 + + +class TestStreamingSynthesizerProtocolContract: + """Contract tests for StreamingSynthesizer protocol.""" + + def test_streaming_synthesizer_is_protocol(self) -> None: + assert hasattr(StreamingSynthesizer, "__protocol_attrs__") + + def test_streaming_session_satisfies_protocol(self) -> None: + session = FakeStreamingSession() + assert isinstance(session, StreamingSynthesizer) + + def test_synthesize_stream_yields_bytes(self) -> None: + session = FakeStreamingSession() + request = SynthesisRequest( + text="Hello", + voice=VoiceSelection(source="builtin", key="af_nova"), + parameters=ParameterValues(), + format=AudioFormat(mime="audio/wav", extension="wav"), + ) + chunks = list(session.synthesizeStream(request)) + assert len(chunks) > 0 + for chunk in chunks: + assert isinstance(chunk, bytes) + + def test_streaming_iterator_exhaustion(self) -> None: + """Architecture spec: Iterator exhaustion = synthesis complete.""" + session = FakeStreamingSession() + request = SynthesisRequest( + text="Hello", + voice=VoiceSelection(source="builtin", key="af_nova"), + parameters=ParameterValues(), + format=AudioFormat(mime="audio/wav", extension="wav"), + ) + chunks = list(session.synthesizeStream(request)) + assert len(chunks) == 3 + + def test_streaming_after_dispose_raises(self) -> None: + """Architecture spec: After dispose(), methods raise EngineError.""" + session = FakeStreamingSession() + session.dispose() + request = SynthesisRequest( + text="Hello", + voice=VoiceSelection(source="builtin", key="af_nova"), + parameters=ParameterValues(), + format=AudioFormat(mime="audio/wav", extension="wav"), + ) + with pytest.raises(EngineError): + list(session.synthesizeStream(request)) + + +class TestCancelableSessionProtocolContract: + """Contract tests for CancelableSession protocol.""" + + def test_cancelable_session_is_protocol(self) -> None: + assert hasattr(CancelableSession, "__protocol_attrs__") + + def test_cancelable_session_satisfies_protocol(self) -> None: + session = FakeCancelableSession() + assert isinstance(session, CancelableSession) + + def test_cancel_causes_synthesize_to_raise_cancelled(self) -> None: + """Architecture spec: cancel() causes synthesize() to raise CancelledError.""" + session = FakeCancelableSession() + request = SynthesisRequest( + text="Hello", + voice=VoiceSelection(source="builtin", key="af_nova"), + parameters=ParameterValues(), + format=AudioFormat(mime="audio/wav", extension="wav"), + ) + + # Cancel + session.cancel() + + # synthesize should raise CancelledError + with pytest.raises(CancelledError): + session.synthesize(request) + + def test_cancel_after_dispose_raises(self) -> None: + """Architecture spec: cancel() raises EngineError if called after dispose().""" + session = FakeCancelableSession() + session.dispose() + with pytest.raises(EngineError): + session.cancel() + + def test_session_usable_after_cancel(self) -> None: + """Architecture spec: EngineSession remains usable after cancellation.""" + session = FakeCancelableSession() + + # Cancel + session.cancel() + + # Dispose and create new session for synthesis + session.dispose() diff --git a/tests/contracts/test_engine_contract.py b/tests/contracts/test_engine_contract.py new file mode 100644 index 0000000..bc73faf --- /dev/null +++ b/tests/contracts/test_engine_contract.py @@ -0,0 +1,106 @@ +"""Contract tests for Engine protocol. + +These tests verify that Engine implementations satisfy the architectural requirements: +- createSession() returns EngineSession +- dispose() is idempotent +- After dispose(), createSession() raises EngineError +- Engine is thread-safe for createSession() +""" + +import pytest + +from abogen.tts_plugin.engine import Engine, EngineSession +from abogen.tts_plugin.errors import EngineError + +from .conftest import FakeEngine, FakeEngineSession + + +class TestEngineProtocolContract: + """Contract tests for the Engine protocol itself.""" + + def test_engine_is_protocol(self) -> None: + assert hasattr(Engine, "__protocol_attrs__") + + def test_engine_session_is_protocol(self) -> None: + assert hasattr(EngineSession, "__protocol_attrs__") + + def test_fake_engine_satisfies_protocol(self) -> None: + engine = FakeEngine() + assert isinstance(engine, Engine) + + def test_fake_session_satisfies_protocol(self) -> None: + session = FakeEngineSession() + assert isinstance(session, EngineSession) + + +class TestEngineCreateSessionContract: + """Contract tests for Engine.createSession().""" + + def test_create_session_returns_engine_session(self) -> None: + engine = FakeEngine() + session = engine.createSession() + assert isinstance(session, EngineSession) + + def test_create_session_returns_new_instance(self) -> None: + engine = FakeEngine() + session1 = engine.createSession() + session2 = engine.createSession() + assert session1 is not session2 + + def test_create_session_ownership_transfers(self) -> None: + """Architecture spec: Ownership transfers to caller.""" + engine = FakeEngine() + session = engine.createSession() + assert isinstance(session, EngineSession) + + +class TestEngineDisposeContract: + """Contract tests for Engine.dispose().""" + + def test_dispose_is_idempotent(self) -> None: + """Architecture spec: dispose() is idempotent.""" + engine = FakeEngine() + engine.dispose() + engine.dispose() # Should not raise + + def test_dispose_never_raises(self) -> None: + """Architecture spec: dispose() never raises exceptions.""" + engine = FakeEngine() + engine.dispose() # Should not raise + + def test_create_session_after_dispose_raises(self) -> None: + """Architecture spec: After dispose(), all methods except dispose() raise EngineError.""" + engine = FakeEngine() + engine.dispose() + with pytest.raises(EngineError): + engine.createSession() + + +class TestEngineLifecycleContract: + """Contract tests for Engine lifecycle.""" + + def test_full_lifecycle(self) -> None: + """Test complete engine lifecycle: create -> sessions -> dispose.""" + engine = FakeEngine() + + # Create sessions + session1 = engine.createSession() + session2 = engine.createSession() + + # Use sessions + assert isinstance(session1, EngineSession) + assert isinstance(session2, EngineSession) + + # Dispose sessions + session1.dispose() + session2.dispose() + + # Dispose engine + engine.dispose() + + def test_engine_disposed_session_raises(self) -> None: + """Architecture spec: After dispose(), all methods except dispose() raise EngineError.""" + engine = FakeEngine() + engine.dispose() + with pytest.raises(EngineError): + engine.createSession() diff --git a/tests/contracts/test_errors_contract.py b/tests/contracts/test_errors_contract.py new file mode 100644 index 0000000..8da3479 --- /dev/null +++ b/tests/contracts/test_errors_contract.py @@ -0,0 +1,85 @@ +"""Contract tests for error hierarchy. + +These tests verify that the error hierarchy satisfies the architectural requirements: +- All errors inherit from EngineError +- EngineError inherits from Exception +- Each error type is properly classified +""" + +import pytest + +from abogen.tts_plugin.errors import ( + CancelledError, + ConfigurationError, + EngineError, + InternalError, + InvalidInputError, + ModelLoadError, + ModelNotFoundError, + NetworkError, +) + + +class TestErrorHierarchyContract: + """Contract tests for the error hierarchy.""" + + def test_engine_error_is_exception(self) -> None: + assert issubclass(EngineError, Exception) + + def test_all_errors_inherit_from_engine_error(self) -> None: + error_classes = [ + ModelNotFoundError, + ModelLoadError, + NetworkError, + InvalidInputError, + ConfigurationError, + CancelledError, + InternalError, + ] + for error_class in error_classes: + assert issubclass(error_class, EngineError), ( + f"{error_class.__name__} must inherit from EngineError" + ) + + def test_all_errors_are_catchable(self) -> None: + error_classes = [ + EngineError, + ModelNotFoundError, + ModelLoadError, + NetworkError, + InvalidInputError, + ConfigurationError, + CancelledError, + InternalError, + ] + for error_class in error_classes: + with pytest.raises(EngineError): + raise error_class("test message") + + def test_error_message_preserved(self) -> None: + msg = "Model not found: bert-base" + with pytest.raises(ModelNotFoundError, match=msg): + raise ModelNotFoundError(msg) + + def test_error_can_be_caught_as_engine_error(self) -> None: + with pytest.raises(EngineError): + raise ModelNotFoundError("test") + + def test_cancelled_error_is_engine_error(self) -> None: + """CancelledError is a subtype of EngineError per architecture spec.""" + assert issubclass(CancelledError, EngineError) + + def test_error_hierarchy_no_cycles(self) -> None: + """Verify no circular inheritance.""" + error_classes = [ + EngineError, + ModelNotFoundError, + ModelLoadError, + NetworkError, + InvalidInputError, + ConfigurationError, + CancelledError, + InternalError, + ] + for cls in error_classes: + assert cls not in cls.__bases__ diff --git a/tests/contracts/test_host_context_contract.py b/tests/contracts/test_host_context_contract.py new file mode 100644 index 0000000..c028f20 --- /dev/null +++ b/tests/contracts/test_host_context_contract.py @@ -0,0 +1,89 @@ +"""Contract tests for HostContext. + +These tests verify that HostContext satisfies the architectural requirements: +- Minimal (3 fields maximum) +- Frozen dataclass +- config_dir: Path +- logger: Logger +- http_client: HttpClient protocol +""" + +import logging +from pathlib import Path + +import pytest + +from abogen.tts_plugin.host_context import HttpClient, HostContext + + +class TestHostContextContract: + """Contract tests for HostContext dataclass.""" + + def test_is_frozen_dataclass(self) -> None: + assert hasattr(HostContext, "__dataclass_params__") + assert HostContext.__dataclass_params__.frozen is True + + def test_required_fields(self, tmp_path: Path) -> None: + logger = logging.getLogger("test") + + class FakeClient: + def get(self, url: str, **kwargs: object) -> object: + return None + + def post(self, url: str, **kwargs: object) -> object: + return None + + ctx = HostContext( + config_dir=tmp_path, + logger=logger, + http_client=FakeClient(), + ) + assert ctx.config_dir == tmp_path + assert ctx.logger is logger + + def test_immutability(self, tmp_path: Path) -> None: + class FakeClient: + def get(self, url: str, **kwargs: object) -> object: + return None + + def post(self, url: str, **kwargs: object) -> object: + return None + + ctx = HostContext( + config_dir=tmp_path, + logger=logging.getLogger("test"), + http_client=FakeClient(), + ) + with pytest.raises(AttributeError): + ctx.config_dir = Path("/other") # type: ignore[misc] + + def test_max_three_fields(self) -> None: + """Architecture spec: HostContext is minimal (3 fields max).""" + import dataclasses + + fields = dataclasses.fields(HostContext) + assert len(fields) <= 3 + + +class TestHttpClientProtocolContract: + """Contract tests for HttpClient protocol.""" + + def test_http_client_is_protocol(self) -> None: + assert hasattr(HttpClient, "__protocol_attrs__") + + def test_http_client_has_get(self) -> None: + assert hasattr(HttpClient, "get") + + def test_http_client_has_post(self) -> None: + assert hasattr(HttpClient, "post") + + def test_http_client_satisfied(self) -> None: + class FakeClient: + def get(self, url: str, **kwargs: object) -> object: + return None + + def post(self, url: str, **kwargs: object) -> object: + return None + + client = FakeClient() + assert isinstance(client, HttpClient) diff --git a/tests/contracts/test_manifest_contract.py b/tests/contracts/test_manifest_contract.py new file mode 100644 index 0000000..aa50440 --- /dev/null +++ b/tests/contracts/test_manifest_contract.py @@ -0,0 +1,290 @@ +"""Contract tests for plugin manifest types. + +These tests verify that manifest types satisfy the architectural requirements: +- All required fields are present +- api_version follows semver format +- capabilities are properly defined +- engine manifest describes the engine correctly +""" + +import re + +import pytest + +from abogen.tts_plugin.manifest import ( + AudioFormatManifest, + EngineManifest, + EnumOption, + GpuRequirement, + ModelManifest, + ParameterManifest, + PluginManifest, + RequirementManifest, + VoiceManifest, + VoiceSourceManifest, +) + + +class TestPluginManifestContract: + """Contract tests for PluginManifest.""" + + def test_required_fields(self) -> None: + manifest = PluginManifest( + id="test-plugin", + name="Test Plugin", + version="1.0.0", + api_version="1.0", + description="A test plugin", + author="Test Author", + ) + assert manifest.id == "test-plugin" + assert manifest.name == "Test Plugin" + assert manifest.version == "1.0.0" + assert manifest.api_version == "1.0" + assert manifest.description == "A test plugin" + assert manifest.author == "Test Author" + + def test_api_version_semver_format(self) -> None: + """Architecture spec: api_version format is semver (MAJOR.MINOR).""" + valid_versions = ["1.0", "2.1", "10.5"] + for version in valid_versions: + manifest = PluginManifest( + id="test", + name="Test", + version="1.0.0", + api_version=version, + description="Test", + author="Test", + ) + assert re.match(r"^\d+\.\d+$", manifest.api_version) + + def test_capabilities_default_empty(self) -> None: + manifest = PluginManifest( + id="test", + name="Test", + version="1.0.0", + api_version="1.0", + description="Test", + author="Test", + ) + assert manifest.capabilities == () + + def test_capabilities_tuple(self) -> None: + manifest = PluginManifest( + id="test", + name="Test", + version="1.0.0", + api_version="1.0", + description="Test", + author="Test", + capabilities=("voice_list", "preview"), + ) + assert "voice_list" in manifest.capabilities + assert "preview" in manifest.capabilities + + def test_requires_default(self) -> None: + manifest = PluginManifest( + id="test", + name="Test", + version="1.0.0", + api_version="1.0", + description="Test", + author="Test", + ) + assert isinstance(manifest.requires, RequirementManifest) + + def test_engine_default(self) -> None: + manifest = PluginManifest( + id="test", + name="Test", + version="1.0.0", + api_version="1.0", + description="Test", + author="Test", + ) + assert isinstance(manifest.engine, EngineManifest) + + +class TestEngineManifestContract: + """Contract tests for EngineManifest.""" + + def test_required_fields(self) -> None: + manifest = EngineManifest( + voiceSources=( + VoiceSourceManifest(id="builtin", name="Builtin", type="list"), + ), + parameters=( + ParameterManifest( + id="speed", name="Speed", description="Speed", type="float", default=1.0 + ), + ), + audioFormats=(AudioFormatManifest(mime="audio/wav", extension="wav"),), + ) + assert len(manifest.voiceSources) == 1 + assert len(manifest.parameters) == 1 + assert len(manifest.audioFormats) == 1 + + def test_defaults_empty(self) -> None: + manifest = EngineManifest() + assert manifest.voiceSources == () + assert manifest.parameters == () + assert manifest.audioFormats == () + + +class TestVoiceSourceManifestContract: + """Contract tests for VoiceSourceManifest.""" + + def test_required_fields(self) -> None: + vs = VoiceSourceManifest(id="builtin", name="Builtin", type="list") + assert vs.id == "builtin" + assert vs.name == "Builtin" + assert vs.type == "list" + + def test_valid_types(self) -> None: + valid_types = ["list", "speaker_id", "clone", "blend", "generate", "none"] + for vtype in valid_types: + vs = VoiceSourceManifest(id="test", name="Test", type=vtype) + assert vs.type == vtype + + def test_config_optional(self) -> None: + vs = VoiceSourceManifest(id="test", name="Test", type="list") + assert vs.config is None + + def test_config_any(self) -> None: + config = {"voices": ["af_nova", "af_sky"]} + vs = VoiceSourceManifest(id="test", name="Test", type="list", config=config) + assert vs.config == config + + +class TestVoiceManifestContract: + """Contract tests for VoiceManifest.""" + + def test_required_fields(self) -> None: + v = VoiceManifest(id="af_nova", name="Nova") + assert v.id == "af_nova" + assert v.name == "Nova" + + def test_tags_default_empty(self) -> None: + v = VoiceManifest(id="af_nova", name="Nova") + assert v.tags == () + + def test_tags_tuple(self) -> None: + v = VoiceManifest(id="af_nova", name="Nova", tags=("en", "female")) + assert "en" in v.tags + assert "female" in v.tags + + +class TestParameterManifestContract: + """Contract tests for ParameterManifest.""" + + def test_required_fields(self) -> None: + p = ParameterManifest( + id="speed", name="Speed", description="Speech speed", type="float", default=1.0 + ) + assert p.id == "speed" + assert p.name == "Speed" + assert p.description == "Speech speed" + assert p.type == "float" + assert p.default == 1.0 + + def test_valid_types(self) -> None: + valid_types = ["float", "int", "string", "boolean", "enum"] + for ptype in valid_types: + p = ParameterManifest( + id="test", name="Test", description="Test", type=ptype, default=None + ) + assert p.type == ptype + + def test_optional_numeric_bounds(self) -> None: + p = ParameterManifest( + id="speed", + name="Speed", + description="Speed", + type="float", + default=1.0, + min=0.5, + max=2.0, + step=0.1, + ) + assert p.min == 0.5 + assert p.max == 2.0 + assert p.step == 0.1 + + def test_enum_options(self) -> None: + options = ( + EnumOption(value="low", label="Low"), + EnumOption(value="high", label="High"), + ) + p = ParameterManifest( + id="quality", + name="Quality", + description="Quality", + type="enum", + default="low", + options=options, + ) + assert len(p.options) == 2 + assert p.options[0].value == "low" + + +class TestAudioFormatManifestContract: + """Contract tests for AudioFormatManifest.""" + + def test_required_fields(self) -> None: + af = AudioFormatManifest(mime="audio/wav", extension="wav") + assert af.mime == "audio/wav" + assert af.extension == "wav" + + +class TestEnumOptionContract: + """Contract tests for EnumOption.""" + + def test_required_fields(self) -> None: + opt = EnumOption(value="low", label="Low Quality") + assert opt.value == "low" + assert opt.label == "Low Quality" + + +class TestRequirementManifestContract: + """Contract tests for RequirementManifest.""" + + def test_defaults(self) -> None: + req = RequirementManifest() + assert req.gpu is None + assert req.memory is None + assert req.internet is None + + def test_with_gpu(self) -> None: + gpu = GpuRequirement(required=True, type="cuda", memory=8.0) + req = RequirementManifest(gpu=gpu) + assert req.gpu.required is True + assert req.gpu.type == "cuda" + assert req.gpu.memory == 8.0 + + def test_with_internet(self) -> None: + req = RequirementManifest(internet=True) + assert req.internet is True + + +class TestGpuRequirementContract: + """Contract tests for GpuRequirement.""" + + def test_defaults(self) -> None: + gpu = GpuRequirement() + assert gpu.required is False + assert gpu.type is None + assert gpu.memory is None + + def test_required_gpu(self) -> None: + gpu = GpuRequirement(required=True, type="cuda", memory=8.0) + assert gpu.required is True + + +class TestModelManifestContract: + """Contract tests for ModelManifest.""" + + def test_required_fields(self) -> None: + m = ModelManifest(id="xtts_v2", name="XTTS v2", size="2GB") + assert m.id == "xtts_v2" + assert m.name == "XTTS v2" + assert m.size == "2GB" diff --git a/tests/contracts/test_plugin_contract.py b/tests/contracts/test_plugin_contract.py new file mode 100644 index 0000000..5fd852d --- /dev/null +++ b/tests/contracts/test_plugin_contract.py @@ -0,0 +1,146 @@ +"""Contract tests for plugin contract. + +These tests verify that plugin modules satisfy the architectural requirements: +- Must export PLUGIN_MANIFEST: PluginManifest +- Must export MODEL_REQUIREMENTS: list[ModelManifest] +- Must export create_engine: Callable[[HostContext, Path | None, EngineConfig], Engine] +- create_engine() must be atomic +""" + +import logging +from pathlib import Path +from typing import Any + +import pytest + +from abogen.tts_plugin.engine import Engine +from abogen.tts_plugin.host_context import HostContext +from abogen.tts_plugin.manifest import EngineManifest, ModelManifest, PluginManifest +from abogen.tts_plugin.plugin import Plugin +from abogen.tts_plugin.types import EngineConfig + +from .conftest import FakeEngine + + +class FakePluginModule: + """Stub plugin module that satisfies the plugin contract.""" + + PLUGIN_MANIFEST = PluginManifest( + id="fake-plugin", + name="Fake Plugin", + version="1.0.0", + api_version="1.0", + description="A fake plugin for testing", + author="Test Author", + capabilities=(), + engine=EngineManifest(), + ) + + MODEL_REQUIREMENTS: list[ModelManifest] = [] + + @staticmethod + def create_engine( + context: HostContext, + model_path: Path | None, + config: EngineConfig, + ) -> Engine: + return FakeEngine() + + +class TestPluginProtocolContract: + """Contract tests for the Plugin protocol.""" + + def test_plugin_is_protocol(self) -> None: + assert hasattr(Plugin, "__protocol_attrs__") + + +class TestPluginExportsContract: + """Contract tests for required plugin exports.""" + + def test_plugin_has_plugin_manifest(self) -> None: + """Architecture spec: Plugin must export PLUGIN_MANIFEST.""" + assert hasattr(FakePluginModule, "PLUGIN_MANIFEST") + assert isinstance(FakePluginModule.PLUGIN_MANIFEST, PluginManifest) + + def test_plugin_has_model_requirements(self) -> None: + """Architecture spec: Plugin must export MODEL_REQUIREMENTS.""" + assert hasattr(FakePluginModule, "MODEL_REQUIREMENTS") + assert isinstance(FakePluginModule.MODEL_REQUIREMENTS, list) + + def test_plugin_has_create_engine(self) -> None: + """Architecture spec: Plugin must export create_engine.""" + assert hasattr(FakePluginModule, "create_engine") + assert callable(FakePluginModule.create_engine) + + def test_plugin_manifest_required_fields(self) -> None: + """Architecture spec: PluginManifest has required fields.""" + manifest = FakePluginModule.PLUGIN_MANIFEST + assert manifest.id + assert manifest.name + assert manifest.version + assert manifest.api_version + assert manifest.description + assert manifest.author + + def test_plugin_manifest_capabilities_is_tuple(self) -> None: + manifest = FakePluginModule.PLUGIN_MANIFEST + assert isinstance(manifest.capabilities, tuple) + + +class TestCreateEngineContract: + """Contract tests for create_engine() function.""" + + def test_create_engine_returns_engine(self) -> None: + """Architecture spec: create_engine() returns Engine.""" + ctx = HostContext( + config_dir=Path("/tmp/test"), + logger=logging.getLogger("test"), + http_client=type("FakeClient", (), {"get": lambda self, **kw: None, "post": lambda self, **kw: None})(), + ) + engine = FakePluginModule.create_engine(ctx, None, EngineConfig()) + assert isinstance(engine, Engine) + + def test_create_engine_atomic(self) -> None: + """Architecture spec: create_engine() is atomic (all-or-nothing).""" + ctx = HostContext( + config_dir=Path("/tmp/test"), + logger=logging.getLogger("test"), + http_client=type("FakeClient", (), {"get": lambda self, **kw: None, "post": lambda self, **kw: None})(), + ) + engine = FakePluginModule.create_engine(ctx, None, EngineConfig()) + assert isinstance(engine, Engine) + engine.dispose() + + def test_create_engine_with_none_model_path(self) -> None: + """Architecture spec: model_path can be None for cloud/no-model engines.""" + ctx = HostContext( + config_dir=Path("/tmp/test"), + logger=logging.getLogger("test"), + http_client=type("FakeClient", (), {"get": lambda self, **kw: None, "post": lambda self, **kw: None})(), + ) + engine = FakePluginModule.create_engine(ctx, None, EngineConfig()) + assert isinstance(engine, Engine) + engine.dispose() + + def test_create_engine_with_model_path(self) -> None: + """Architecture spec: model_path is Path | None.""" + ctx = HostContext( + config_dir=Path("/tmp/test"), + logger=logging.getLogger("test"), + http_client=type("FakeClient", (), {"get": lambda self, **kw: None, "post": lambda self, **kw: None})(), + ) + engine = FakePluginModule.create_engine(ctx, Path("/models/test"), EngineConfig()) + assert isinstance(engine, Engine) + engine.dispose() + + +class TestModelRequirementsContract: + """Contract tests for MODEL_REQUIREMENTS.""" + + def test_model_requirements_is_list(self) -> None: + assert isinstance(FakePluginModule.MODEL_REQUIREMENTS, list) + + def test_model_requirements_contains_model_manifests(self) -> None: + """If non-empty, each item must be a ModelManifest.""" + for req in FakePluginModule.MODEL_REQUIREMENTS: + assert isinstance(req, ModelManifest) diff --git a/tests/contracts/test_session_contract.py b/tests/contracts/test_session_contract.py new file mode 100644 index 0000000..d24f189 --- /dev/null +++ b/tests/contracts/test_session_contract.py @@ -0,0 +1,135 @@ +"""Contract tests for EngineSession protocol. + +These tests verify that EngineSession implementations satisfy the architectural requirements: +- synthesize() returns SynthesizedAudio +- dispose() is idempotent +- After dispose(), synthesize() raises EngineError +- Session remains usable after synthesize() failure +""" + +import pytest + +from abogen.tts_plugin.engine import EngineSession +from abogen.tts_plugin.errors import EngineError +from abogen.tts_plugin.types import ( + AudioFormat, + Duration, + ParameterValues, + SynthesisRequest, + SynthesizedAudio, + VoiceSelection, +) + +from .conftest import FakeEngineSession + + +class TestEngineSessionProtocolContract: + """Contract tests for the EngineSession protocol itself.""" + + def test_engine_session_is_protocol(self) -> None: + assert hasattr(EngineSession, "__protocol_attrs__") + + def test_fake_session_satisfies_protocol(self) -> None: + session = FakeEngineSession() + assert isinstance(session, EngineSession) + + +class TestSessionSynthesizeContract: + """Contract tests for EngineSession.synthesize().""" + + def test_synthesize_returns_synthesized_audio(self) -> None: + session = FakeEngineSession() + request = SynthesisRequest( + text="Hello", + voice=VoiceSelection(source="builtin", key="af_nova"), + parameters=ParameterValues(), + format=AudioFormat(mime="audio/wav", extension="wav"), + ) + result = session.synthesize(request) + assert isinstance(result, SynthesizedAudio) + + def test_synthesize_returns_valid_audio_data(self) -> None: + session = FakeEngineSession() + request = SynthesisRequest( + text="Hello", + voice=VoiceSelection(source="builtin", key="af_nova"), + parameters=ParameterValues(), + format=AudioFormat(mime="audio/wav", extension="wav"), + ) + result = session.synthesize(request) + assert isinstance(result.data, bytes) + assert len(result.data) > 0 + assert isinstance(result.format, AudioFormat) + assert isinstance(result.duration, Duration) + + +class TestSessionDisposeContract: + """Contract tests for EngineSession.dispose().""" + + def test_dispose_is_idempotent(self) -> None: + """Architecture spec: dispose() is idempotent.""" + session = FakeEngineSession() + session.dispose() + session.dispose() # Should not raise + + def test_dispose_never_raises(self) -> None: + """Architecture spec: dispose() never raises exceptions.""" + session = FakeEngineSession() + session.dispose() # Should not raise + + +class TestSessionAfterDisposeContract: + """Contract tests for behavior after dispose().""" + + def test_synthesize_after_dispose_raises(self) -> None: + """Architecture spec: After dispose(), all methods except dispose() raise EngineError.""" + session = FakeEngineSession() + session.dispose() + request = SynthesisRequest( + text="Hello", + voice=VoiceSelection(source="builtin", key="af_nova"), + parameters=ParameterValues(), + format=AudioFormat(mime="audio/wav", extension="wav"), + ) + with pytest.raises(EngineError): + session.synthesize(request) + + +class TestSessionLifecycleContract: + """Contract tests for EngineSession lifecycle.""" + + def test_full_lifecycle(self) -> None: + """Test complete session lifecycle: create -> synthesize -> dispose.""" + session = FakeEngineSession() + request = SynthesisRequest( + text="Hello", + voice=VoiceSelection(source="builtin", key="af_nova"), + parameters=ParameterValues(), + format=AudioFormat(mime="audio/wav", extension="wav"), + ) + + # Synthesize + result = session.synthesize(request) + assert isinstance(result, SynthesizedAudio) + + # Dispose + session.dispose() + + def test_multiple_synthesize_before_dispose(self) -> None: + """Architecture spec: Session remains usable after synthesize() failure.""" + session = FakeEngineSession() + request = SynthesisRequest( + text="Hello", + voice=VoiceSelection(source="builtin", key="af_nova"), + parameters=ParameterValues(), + format=AudioFormat(mime="audio/wav", extension="wav"), + ) + + # Multiple synthesize calls + result1 = session.synthesize(request) + result2 = session.synthesize(request) + assert isinstance(result1, SynthesizedAudio) + assert isinstance(result2, SynthesizedAudio) + + # Dispose + session.dispose() diff --git a/tests/contracts/test_types_contract.py b/tests/contracts/test_types_contract.py new file mode 100644 index 0000000..31d7edb --- /dev/null +++ b/tests/contracts/test_types_contract.py @@ -0,0 +1,207 @@ +"""Contract tests for core domain value objects. + +These tests verify that value objects satisfy the architectural requirements: +- Frozen (immutable) dataclasses +- Correct field definitions +- Proper equality behavior +""" + +import pytest + +from abogen.tts_plugin.types import ( + AudioFormat, + Duration, + EngineConfig, + ParameterValues, + SynthesisRequest, + SynthesizedAudio, + VoiceSelection, +) + + +class TestAudioFormatContract: + """Contract tests for AudioFormat value object.""" + + def test_is_frozen_dataclass(self) -> None: + assert hasattr(AudioFormat, "__dataclass_params__") + assert AudioFormat.__dataclass_params__.frozen is True + + def test_required_fields(self) -> None: + af = AudioFormat(mime="audio/wav", extension="wav") + assert af.mime == "audio/wav" + assert af.extension == "wav" + + def test_immutability(self) -> None: + af = AudioFormat(mime="audio/wav", extension="wav") + with pytest.raises(AttributeError): + af.mime = "audio/mpeg" # type: ignore[misc] + + def test_equality(self) -> None: + af1 = AudioFormat(mime="audio/wav", extension="wav") + af2 = AudioFormat(mime="audio/wav", extension="wav") + assert af1 == af2 + + def test_inequality(self) -> None: + af1 = AudioFormat(mime="audio/wav", extension="wav") + af2 = AudioFormat(mime="audio/mpeg", extension="mp3") + assert af1 != af2 + + def test_hashable(self) -> None: + af = AudioFormat(mime="audio/wav", extension="wav") + assert hash(af) == hash(AudioFormat(mime="audio/wav", extension="wav")) + + +class TestDurationContract: + """Contract tests for Duration value object.""" + + def test_is_frozen_dataclass(self) -> None: + assert hasattr(Duration, "__dataclass_params__") + assert Duration.__dataclass_params__.frozen is True + + def test_required_fields(self) -> None: + d = Duration(seconds=1.5) + assert d.seconds == 1.5 + + def test_immutability(self) -> None: + d = Duration(seconds=1.0) + with pytest.raises(AttributeError): + d.seconds = 2.0 # type: ignore[misc] + + def test_equality(self) -> None: + d1 = Duration(seconds=1.0) + d2 = Duration(seconds=1.0) + assert d1 == d2 + + +class TestVoiceSelectionContract: + """Contract tests for VoiceSelection value object.""" + + def test_is_frozen_dataclass(self) -> None: + assert hasattr(VoiceSelection, "__dataclass_params__") + assert VoiceSelection.__dataclass_params__.frozen is True + + def test_required_fields(self) -> None: + vs = VoiceSelection(source="builtin", key="af_nova") + assert vs.source == "builtin" + assert vs.key == "af_nova" + + def test_payload_default_none(self) -> None: + vs = VoiceSelection(source="builtin", key="af_nova") + assert vs.payload is None + + def test_payload_optional(self) -> None: + vs = VoiceSelection(source="clone", key="my_voice", payload=b"audio_data") + assert vs.payload == b"audio_data" + + def test_immutability(self) -> None: + vs = VoiceSelection(source="builtin", key="af_nova") + with pytest.raises(AttributeError): + vs.source = "other" # type: ignore[misc] + + +class TestParameterValuesContract: + """Contract tests for ParameterValues value object.""" + + def test_is_frozen_dataclass(self) -> None: + assert hasattr(ParameterValues, "__dataclass_params__") + assert ParameterValues.__dataclass_params__.frozen is True + + def test_default_empty(self) -> None: + pv = ParameterValues() + assert pv.values == {} + + def test_with_values(self) -> None: + pv = ParameterValues(values={"speed": 1.0, "pitch": 0.5}) + assert pv.values["speed"] == 1.0 + assert pv.values["pitch"] == 0.5 + + def test_immutability(self) -> None: + pv = ParameterValues(values={"speed": 1.0}) + with pytest.raises(AttributeError): + pv.values = {} # type: ignore[misc] + + +class TestSynthesisRequestContract: + """Contract tests for SynthesisRequest value object.""" + + def test_is_frozen_dataclass(self) -> None: + assert hasattr(SynthesisRequest, "__dataclass_params__") + assert SynthesisRequest.__dataclass_params__.frozen is True + + def test_required_fields(self) -> None: + req = SynthesisRequest( + text="Hello", + voice=VoiceSelection(source="builtin", key="af_nova"), + parameters=ParameterValues(), + format=AudioFormat(mime="audio/wav", extension="wav"), + ) + assert req.text == "Hello" + assert req.voice.source == "builtin" + assert req.format.mime == "audio/wav" + + def test_immutability(self) -> None: + req = SynthesisRequest( + text="Hello", + voice=VoiceSelection(source="builtin", key="af_nova"), + parameters=ParameterValues(), + format=AudioFormat(mime="audio/wav", extension="wav"), + ) + with pytest.raises(AttributeError): + req.text = "World" # type: ignore[misc] + + +class TestSynthesizedAudioContract: + """Contract tests for SynthesizedAudio value object.""" + + def test_is_frozen_dataclass(self) -> None: + assert hasattr(SynthesizedAudio, "__dataclass_params__") + assert SynthesizedAudio.__dataclass_params__.frozen is True + + def test_required_fields(self) -> None: + audio = SynthesizedAudio( + data=b"\x00" * 100, + format=AudioFormat(mime="audio/wav", extension="wav"), + duration=Duration(seconds=1.0), + ) + assert audio.data == b"\x00" * 100 + assert audio.format.mime == "audio/wav" + assert audio.duration.seconds == 1.0 + + def test_immutability(self) -> None: + audio = SynthesizedAudio( + data=b"\x00" * 100, + format=AudioFormat(mime="audio/wav", extension="wav"), + duration=Duration(seconds=1.0), + ) + with pytest.raises(AttributeError): + audio.data = b"\x00" # type: ignore[misc] + + +class TestEngineConfigContract: + """Contract tests for EngineConfig value object.""" + + def test_is_frozen_dataclass(self) -> None: + assert hasattr(EngineConfig, "__dataclass_params__") + assert EngineConfig.__dataclass_params__.frozen is True + + def test_default_device(self) -> None: + config = EngineConfig() + assert config.device == "cpu" + + def test_custom_device(self) -> None: + config = EngineConfig(device="cuda:0") + assert config.device == "cuda:0" + + def test_immutability(self) -> None: + config = EngineConfig() + with pytest.raises(AttributeError): + config.device = "cuda:0" # type: ignore[misc] + + def test_unknown_keys_ignored_per_spec(self) -> None: + """Architecture spec: Unknown keys are ignored (no error). + + EngineConfig is frozen, so unknown keys cannot be set after creation. + This test verifies the default behavior matches the spec. + """ + config = EngineConfig() + assert config.device == "cpu"