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.
This commit is contained in:
Artem Akymenko
2026-07-12 16:20:05 +03:00
parent 79b3d26f66
commit 0f568120f4
10 changed files with 1477 additions and 0 deletions
+5
View File
@@ -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.
"""
+231
View File
@@ -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,
)
@@ -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()
+106
View File
@@ -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()
+85
View File
@@ -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__
@@ -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)
+290
View File
@@ -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"
+146
View File
@@ -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)
+135
View File
@@ -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()
+207
View File
@@ -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"