From 6eda8516ccd1aac12f3d25e9b9ee85269fc10507 Mon Sep 17 00:00:00 2001 From: Artem Akymenko Date: Thu, 9 Jul 2026 09:01:55 +0000 Subject: [PATCH] feat: add plugin loader infrastructure Implement plugin loading and validation: - loader.py: discover, import, validate plugins - validate PLUGIN_MANIFEST, MODEL_REQUIREMENTS, create_engine - validate api_version compatibility (major must match) - validate capabilities (reject unknown) - diagnostic messages for all error cases - no partial registration after error Test plugins: - fake_plugin: minimal valid plugin for testing - missing_manifest: no PLUGIN_MANIFEST - invalid_api_version: major version mismatch - invalid_capabilities: unknown capabilities - missing_create_engine: no create_engine function - import_error: raises ImportError during import - missing_model_requirements: no MODEL_REQUIREMENTS 39 new tests covering all loader functionality. --- abogen/tts_plugin/loader.py | 365 +++++++++++++++ tests/contracts/test_loader_contract.py | 436 ++++++++++++++++++ tests/plugins/fake_plugin/__init__.py | 92 ++++ tests/plugins/import_error/__init__.py | 18 + tests/plugins/invalid_api_version/__init__.py | 29 ++ .../plugins/invalid_capabilities/__init__.py | 29 ++ .../plugins/missing_create_engine/__init__.py | 18 + tests/plugins/missing_manifest/__init__.py | 10 + .../missing_model_requirements/__init__.py | 28 ++ 9 files changed, 1025 insertions(+) create mode 100644 abogen/tts_plugin/loader.py create mode 100644 tests/contracts/test_loader_contract.py create mode 100644 tests/plugins/fake_plugin/__init__.py create mode 100644 tests/plugins/import_error/__init__.py create mode 100644 tests/plugins/invalid_api_version/__init__.py create mode 100644 tests/plugins/invalid_capabilities/__init__.py create mode 100644 tests/plugins/missing_create_engine/__init__.py create mode 100644 tests/plugins/missing_manifest/__init__.py create mode 100644 tests/plugins/missing_model_requirements/__init__.py diff --git a/abogen/tts_plugin/loader.py b/abogen/tts_plugin/loader.py new file mode 100644 index 0000000..fa600ab --- /dev/null +++ b/abogen/tts_plugin/loader.py @@ -0,0 +1,365 @@ +"""Plugin loader infrastructure for the TTS Plugin Architecture. + +This module provides functionality to discover, import, validate, and load +TTS plugins. It handles both valid and invalid plugins, providing diagnostic +messages for errors. + +The loader does NOT: +- Create Engine instances (that's the plugin's create_engine() responsibility) +- Manage plugin lifecycle (that's the Plugin Manager's responsibility) +- Implement any TTS engine functionality +""" + +from __future__ import annotations + +import importlib +import re +import sys +import types +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Callable + +from abogen.tts_plugin.manifest import ModelManifest, PluginManifest + + +# Host API version for compatibility checking +HOST_API_VERSION = "1.0" + + +@dataclass(frozen=True) +class PluginLoadError: + """Diagnostic information for a failed plugin load. + + Attributes: + plugin_id: Plugin identifier if available, otherwise directory name. + path: Path to the plugin directory. + errors: List of error messages describing what went wrong. + """ + + plugin_id: str + path: Path + errors: tuple[str, ...] = field(default_factory=tuple) + + +@dataclass(frozen=True) +class PluginLoadResult: + """Result of loading a plugin. + + Attributes: + success: Whether the plugin loaded successfully. + manifest: The plugin manifest if successful. + model_requirements: Model requirements if successful. + create_engine: The create_engine function if successful. + module: The plugin module if successful. + error: Error information if failed. + """ + + success: bool + manifest: PluginManifest | None = None + model_requirements: tuple[ModelManifest, ...] | None = None + create_engine: Callable[..., Any] | None = None + module: types.ModuleType | None = None + error: PluginLoadError | None = None + + +def _parse_api_version(version: str) -> tuple[int, int] | None: + """Parse an api_version string into (major, minor) tuple. + + Args: + version: Version string in format "MAJOR.MINOR". + + Returns: + Tuple of (major, minor) or None if invalid format. + """ + match = re.match(r"^(\d+)\.(\d+)$", version) + if match: + return int(match.group(1)), int(match.group(2)) + return None + + +def _check_api_version_compatibility(plugin_version: str) -> str | None: + """Check if plugin api_version is compatible with host. + + Architecture spec: + - Format: semver (MAJOR.MINOR) + - Compatibility: Host rejects plugin if major version differs + - Minor version: backward compatible, Host accepts higher minor + + Args: + plugin_version: Plugin's api_version string. + + Returns: + Error message if incompatible, None if compatible. + """ + plugin_ver = _parse_api_version(plugin_version) + if plugin_ver is None: + return f"Invalid api_version format: '{plugin_version}'. Expected format: MAJOR.MINOR" + + host_ver = _parse_api_version(HOST_API_VERSION) + if host_ver is None: + return f"Invalid host api_version format: '{HOST_API_VERSION}'" + + if plugin_ver[0] != host_ver[0]: + return ( + f"api_version major mismatch: plugin={plugin_ver[0]}, host={host_ver[0]}. " + f"Major version must match for compatibility." + ) + + return None + + +def _validate_manifest(module: types.ModuleType, plugin_dir: Path) -> list[str]: + """Validate that a plugin module has required exports. + + Args: + module: The imported plugin module. + plugin_dir: Path to the plugin directory. + + Returns: + List of error messages (empty if valid). + """ + errors: list[str] = [] + + # Check PLUGIN_MANIFEST + manifest = getattr(module, "PLUGIN_MANIFEST", None) + if manifest is None: + errors.append("Missing PLUGIN_MANIFEST export") + elif not isinstance(manifest, PluginManifest): + errors.append( + f"PLUGIN_MANIFEST must be a PluginManifest instance, " + f"got {type(manifest).__name__}" + ) + + # Check MODEL_REQUIREMENTS + model_reqs = getattr(module, "MODEL_REQUIREMENTS", None) + if model_reqs is None: + errors.append("Missing MODEL_REQUIREMENTS export") + elif not isinstance(model_reqs, list): + errors.append( + f"MODEL_REQUIREMENTS must be a list, got {type(model_reqs).__name__}" + ) + else: + for i, req in enumerate(model_reqs): + if not isinstance(req, ModelManifest): + errors.append( + f"MODEL_REQUIREMENTS[{i}] must be a ModelManifest instance, " + f"got {type(req).__name__}" + ) + + # Check create_engine + create_engine = getattr(module, "create_engine", None) + if create_engine is None: + errors.append("Missing create_engine export") + elif not callable(create_engine): + errors.append( + f"create_engine must be callable, got {type(create_engine).__name__}" + ) + + return errors + + +def _validate_capabilities(manifest: PluginManifest) -> list[str]: + """Validate plugin capabilities. + + Args: + manifest: The plugin manifest to validate. + + Returns: + List of error messages (empty if valid). + """ + errors: list[str] = [] + + # Known capabilities (can be extended) + known_capabilities = frozenset({ + "voice_list", + "preview", + "voice_clone", + "voice_blend", + "streaming", + "cancel", + }) + + for cap in manifest.capabilities: + if cap not in known_capabilities: + errors.append(f"Unknown capability: '{cap}'") + + return errors + + +def _validate_api_version(manifest: PluginManifest) -> list[str]: + """Validate api_version compatibility. + + Args: + manifest: The plugin manifest to validate. + + Returns: + List of error messages (empty if valid). + """ + errors: list[str] = [] + error = _check_api_version_compatibility(manifest.api_version) + if error: + errors.append(error) + return errors + + +def load_plugin_from_dir(plugin_dir: Path) -> PluginLoadResult: + """Load and validate a plugin from a directory. + + The plugin directory must contain an __init__.py that exports: + - PLUGIN_MANIFEST: PluginManifest + - MODEL_REQUIREMENTS: list[ModelManifest] + - create_engine: Callable + + Args: + plugin_dir: Path to the plugin directory. + + Returns: + PluginLoadResult with success status and either plugin data or error info. + """ + plugin_id = plugin_dir.name + errors: list[str] = [] + + # Check if directory exists + if not plugin_dir.exists(): + return PluginLoadResult( + success=False, + error=PluginLoadError( + plugin_id=plugin_id, + path=plugin_dir, + errors=(f"Plugin directory does not exist: {plugin_dir}",), + ), + ) + + # Check for __init__.py + init_file = plugin_dir / "__init__.py" + if not init_file.exists(): + return PluginLoadResult( + success=False, + error=PluginLoadError( + plugin_id=plugin_id, + path=plugin_dir, + errors=("Missing __init__.py in plugin directory",), + ), + ) + + # Import the module + module_name = f"abogen.tts_plugin._loaded.{plugin_id}" + try: + # Remove from cache if already imported (for testing) + if module_name in sys.modules: + del sys.modules[module_name] + + spec = importlib.util.spec_from_file_location( + module_name, init_file, submodule_search_locations=[] + ) + if spec is None or spec.loader is None: + return PluginLoadResult( + success=False, + error=PluginLoadError( + plugin_id=plugin_id, + path=plugin_dir, + errors=(f"Failed to create module spec for {init_file}",), + ), + ) + + module = importlib.util.module_from_spec(spec) + sys.modules[module_name] = module + spec.loader.exec_module(module) + except Exception as e: + # Clean up module from sys.modules on import failure + if module_name in sys.modules: + del sys.modules[module_name] + return PluginLoadResult( + success=False, + error=PluginLoadError( + plugin_id=plugin_id, + path=plugin_dir, + errors=(f"Failed to import plugin module: {e}",), + ), + ) + + # Validate manifest + manifest_errors = _validate_manifest(module, plugin_dir) + errors.extend(manifest_errors) + + # If manifest is valid, perform additional validation + manifest = getattr(module, "PLUGIN_MANIFEST", None) + if isinstance(manifest, PluginManifest): + # Validate api_version + api_errors = _validate_api_version(manifest) + errors.extend(api_errors) + + # Validate capabilities + cap_errors = _validate_capabilities(manifest) + errors.extend(cap_errors) + + # Use manifest id if available + plugin_id = manifest.id + + # Check if any errors occurred + if errors: + # Clean up module from sys.modules + if module_name in sys.modules: + del sys.modules[module_name] + + return PluginLoadResult( + success=False, + error=PluginLoadError( + plugin_id=plugin_id, + path=plugin_dir, + errors=tuple(errors), + ), + ) + + # Get MODEL_REQUIREMENTS + model_requirements = tuple(getattr(module, "MODEL_REQUIREMENTS", [])) + create_engine = getattr(module, "create_engine", None) + + return PluginLoadResult( + success=True, + manifest=manifest, + model_requirements=model_requirements, + create_engine=create_engine, + module=module, + ) + + +def discover_plugins(plugin_dirs: list[Path]) -> list[PluginLoadResult]: + """Discover and load plugins from multiple directories. + + Args: + plugin_dirs: List of directories to scan for plugins. + + Returns: + List of PluginLoadResult, one per plugin directory found. + """ + results: list[PluginLoadResult] = [] + + for plugin_dir in plugin_dirs: + if not plugin_dir.exists(): + continue + + # Scan for subdirectories (each is a potential plugin) + for item in sorted(plugin_dir.iterdir()): + if item.is_dir() and not item.name.startswith("."): + result = load_plugin_from_dir(item) + results.append(result) + + return results + + +def load_plugin( + plugin_dir: Path, +) -> PluginLoadResult: + """Load a single plugin from a directory. + + This is the main entry point for loading a plugin. + + Args: + plugin_dir: Path to the plugin directory. + + Returns: + PluginLoadResult with success status and either plugin data or error info. + """ + return load_plugin_from_dir(plugin_dir) diff --git a/tests/contracts/test_loader_contract.py b/tests/contracts/test_loader_contract.py new file mode 100644 index 0000000..9213993 --- /dev/null +++ b/tests/contracts/test_loader_contract.py @@ -0,0 +1,436 @@ +"""Comprehensive tests for the plugin loader infrastructure. + +These tests verify that the loader correctly: +- Discovers plugins in directories +- Imports plugin modules +- Validates PLUGIN_MANIFEST, MODEL_REQUIREMENTS, create_engine +- Validates api_version compatibility +- Validates capabilities +- Provides diagnostic messages for errors +- Rejects invalid plugins +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +import pytest + +from abogen.tts_plugin.loader import ( + HOST_API_VERSION, + PluginLoadError, + PluginLoadResult, + _check_api_version_compatibility, + _parse_api_version, + _validate_api_version, + _validate_capabilities, + _validate_manifest, + discover_plugins, + load_plugin, + load_plugin_from_dir, +) +from abogen.tts_plugin.manifest import ( + EngineManifest, + ModelManifest, + PluginManifest, +) + + +# ────────────────────────────────────────────────────────────── +# Path fixtures +# ────────────────────────────────────────────────────────────── + +@pytest.fixture +def plugins_dir() -> Path: + return Path(__file__).parent.parent / "plugins" + + +@pytest.fixture +def fake_plugin_dir(plugins_dir: Path) -> Path: + return plugins_dir / "fake_plugin" + + +@pytest.fixture +def missing_manifest_dir(plugins_dir: Path) -> Path: + return plugins_dir / "missing_manifest" + + +@pytest.fixture +def invalid_api_version_dir(plugins_dir: Path) -> Path: + return plugins_dir / "invalid_api_version" + + +@pytest.fixture +def invalid_capabilities_dir(plugins_dir: Path) -> Path: + return plugins_dir / "invalid_capabilities" + + +@pytest.fixture +def missing_create_engine_dir(plugins_dir: Path) -> Path: + return plugins_dir / "missing_create_engine" + + +@pytest.fixture +def import_error_dir(plugins_dir: Path) -> Path: + return plugins_dir / "import_error" + + +@pytest.fixture +def missing_model_requirements_dir(plugins_dir: Path) -> Path: + return plugins_dir / "missing_model_requirements" + + +# ────────────────────────────────────────────────────────────── +# Unit tests: _parse_api_version +# ────────────────────────────────────────────────────────────── + +class TestParseApiVersion: + def test_valid_version(self) -> None: + assert _parse_api_version("1.0") == (1, 0) + assert _parse_api_version("2.5") == (2, 5) + assert _parse_api_version("10.20") == (10, 20) + + def test_invalid_format(self) -> None: + assert _parse_api_version("1") is None + assert _parse_api_version("1.0.0") is None + assert _parse_api_version("abc") is None + assert _parse_api_version("") is None + assert _parse_api_version("1.x") is None + + +# ────────────────────────────────────────────────────────────── +# Unit tests: _check_api_version_compatibility +# ────────────────────────────────────────────────────────────── + +class TestCheckApiVersionCompatibility: + def test_compatible_version(self) -> None: + assert _check_api_version_compatibility("1.0") is None + assert _check_api_version_compatibility("1.5") is None + + def test_major_mismatch(self) -> None: + error = _check_api_version_compatibility("2.0") + assert error is not None + assert "major mismatch" in error + + def test_invalid_format(self) -> None: + error = _check_api_version_compatibility("invalid") + assert error is not None + assert "Invalid api_version format" in error + + +# ────────────────────────────────────────────────────────────── +# Unit tests: _validate_manifest +# ────────────────────────────────────────────────────────────── + +class TestValidateManifest: + def test_valid_manifest(self) -> None: + class FakeModule: + PLUGIN_MANIFEST = PluginManifest( + id="test", name="Test", version="1.0.0", + api_version="1.0", description="Test", author="Test", + ) + MODEL_REQUIREMENTS: list = [] + create_engine = lambda *a, **kw: None + + errors = _validate_manifest(FakeModule(), Path("/tmp")) + assert errors == [] + + def test_missing_manifest(self) -> None: + class FakeModule: + MODEL_REQUIREMENTS: list = [] + create_engine = lambda *a, **kw: None + + errors = _validate_manifest(FakeModule(), Path("/tmp")) + assert any("Missing PLUGIN_MANIFEST" in e for e in errors) + + def test_wrong_manifest_type(self) -> None: + class FakeModule: + PLUGIN_MANIFEST = "not a manifest" + MODEL_REQUIREMENTS: list = [] + create_engine = lambda *a, **kw: None + + errors = _validate_manifest(FakeModule(), Path("/tmp")) + assert any("PluginManifest instance" in e for e in errors) + + def test_missing_model_requirements(self) -> None: + class FakeModule: + PLUGIN_MANIFEST = PluginManifest( + id="test", name="Test", version="1.0.0", + api_version="1.0", description="Test", author="Test", + ) + create_engine = lambda *a, **kw: None + + errors = _validate_manifest(FakeModule(), Path("/tmp")) + assert any("Missing MODEL_REQUIREMENTS" in e for e in errors) + + def test_wrong_model_requirements_type(self) -> None: + class FakeModule: + PLUGIN_MANIFEST = PluginManifest( + id="test", name="Test", version="1.0.0", + api_version="1.0", description="Test", author="Test", + ) + MODEL_REQUIREMENTS = "not a list" + create_engine = lambda *a, **kw: None + + errors = _validate_manifest(FakeModule(), Path("/tmp")) + assert any("must be a list" in e for e in errors) + + def test_invalid_model_requirements_item(self) -> None: + class FakeModule: + PLUGIN_MANIFEST = PluginManifest( + id="test", name="Test", version="1.0.0", + api_version="1.0", description="Test", author="Test", + ) + MODEL_REQUIREMENTS = ["not a model manifest"] + create_engine = lambda *a, **kw: None + + errors = _validate_manifest(FakeModule(), Path("/tmp")) + assert any("ModelManifest instance" in e for e in errors) + + def test_missing_create_engine(self) -> None: + class FakeModule: + PLUGIN_MANIFEST = PluginManifest( + id="test", name="Test", version="1.0.0", + api_version="1.0", description="Test", author="Test", + ) + MODEL_REQUIREMENTS: list = [] + + errors = _validate_manifest(FakeModule(), Path("/tmp")) + assert any("Missing create_engine" in e for e in errors) + + def test_create_engine_not_callable(self) -> None: + class FakeModule: + PLUGIN_MANIFEST = PluginManifest( + id="test", name="Test", version="1.0.0", + api_version="1.0", description="Test", author="Test", + ) + MODEL_REQUIREMENTS: list = [] + create_engine = "not callable" + + errors = _validate_manifest(FakeModule(), Path("/tmp")) + assert any("must be callable" in e for e in errors) + + +# ────────────────────────────────────────────────────────────── +# Unit tests: _validate_capabilities +# ────────────────────────────────────────────────────────────── + +class TestValidateCapabilities: + def test_valid_capabilities(self) -> None: + manifest = PluginManifest( + id="test", name="Test", version="1.0.0", + api_version="1.0", description="Test", author="Test", + capabilities=("voice_list", "preview"), + ) + errors = _validate_capabilities(manifest) + assert errors == [] + + def test_unknown_capability(self) -> None: + manifest = PluginManifest( + id="test", name="Test", version="1.0.0", + api_version="1.0", description="Test", author="Test", + capabilities=("voice_list", "unknown_cap"), + ) + errors = _validate_capabilities(manifest) + assert any("unknown_cap" in e for e in errors) + + def test_empty_capabilities(self) -> None: + manifest = PluginManifest( + id="test", name="Test", version="1.0.0", + api_version="1.0", description="Test", author="Test", + capabilities=(), + ) + errors = _validate_capabilities(manifest) + assert errors == [] + + +# ────────────────────────────────────────────────────────────── +# Unit tests: _validate_api_version +# ────────────────────────────────────────────────────────────── + +class TestValidateApiVersion: + def test_compatible(self) -> None: + manifest = PluginManifest( + id="test", name="Test", version="1.0.0", + api_version="1.0", description="Test", author="Test", + ) + errors = _validate_api_version(manifest) + assert errors == [] + + def test_incompatible(self) -> None: + manifest = PluginManifest( + id="test", name="Test", version="1.0.0", + api_version="2.0", description="Test", author="Test", + ) + errors = _validate_api_version(manifest) + assert len(errors) > 0 + + +# ────────────────────────────────────────────────────────────── +# Integration tests: load_plugin_from_dir +# ────────────────────────────────────────────────────────────── + +class TestLoadPluginFromDir: + def test_load_valid_plugin(self, fake_plugin_dir: Path) -> None: + result = load_plugin_from_dir(fake_plugin_dir) + assert result.success is True + assert result.manifest is not None + assert result.manifest.id == "fake_plugin" + assert result.model_requirements is not None + assert result.create_engine is not None + assert result.module is not None + assert result.error is None + + def test_plugin_satisfies_protocol(self, fake_plugin_dir: Path) -> None: + from abogen.tts_plugin.engine import Engine + from abogen.tts_plugin.host_context import HostContext + import logging + + result = load_plugin_from_dir(fake_plugin_dir) + assert result.success is True + + # Create engine using the loaded create_engine function + ctx = HostContext( + config_dir=Path("/tmp/test"), + logger=logging.getLogger("test"), + http_client=type("FakeClient", (), {"get": lambda s, **kw: None, "post": lambda s, **kw: None})(), + ) + engine = result.create_engine(ctx, None, __import__("abogen.tts_plugin.types", fromlist=["EngineConfig"]).EngineConfig()) + assert isinstance(engine, Engine) + engine.dispose() + + def test_nonexistent_directory(self, tmp_path: Path) -> None: + result = load_plugin_from_dir(tmp_path / "nonexistent") + assert result.success is False + assert result.error is not None + assert "does not exist" in result.error.errors[0] + + def test_missing_init_file(self, tmp_path: Path) -> None: + plugin_dir = tmp_path / "no_init" + plugin_dir.mkdir() + result = load_plugin_from_dir(plugin_dir) + assert result.success is False + assert result.error is not None + assert "__init__.py" in result.error.errors[0] + + def test_import_error(self, import_error_dir: Path) -> None: + result = load_plugin_from_dir(import_error_dir) + assert result.success is False + assert result.error is not None + assert "Failed to import" in result.error.errors[0] + + +# ────────────────────────────────────────────────────────────── +# Integration tests: invalid plugins +# ────────────────────────────────────────────────────────────── + +class TestInvalidPlugins: + def test_missing_manifest(self, missing_manifest_dir: Path) -> None: + result = load_plugin_from_dir(missing_manifest_dir) + assert result.success is False + assert result.error is not None + assert any("Missing PLUGIN_MANIFEST" in e for e in result.error.errors) + + def test_invalid_api_version(self, invalid_api_version_dir: Path) -> None: + result = load_plugin_from_dir(invalid_api_version_dir) + assert result.success is False + assert result.error is not None + assert any("major mismatch" in e for e in result.error.errors) + + def test_invalid_capabilities(self, invalid_capabilities_dir: Path) -> None: + result = load_plugin_from_dir(invalid_capabilities_dir) + assert result.success is False + assert result.error is not None + assert any("Unknown capability" in e for e in result.error.errors) + + def test_missing_create_engine(self, missing_create_engine_dir: Path) -> None: + result = load_plugin_from_dir(missing_create_engine_dir) + assert result.success is False + assert result.error is not None + assert any("Missing create_engine" in e for e in result.error.errors) + + def test_missing_model_requirements(self, missing_model_requirements_dir: Path) -> None: + result = load_plugin_from_dir(missing_model_requirements_dir) + assert result.success is False + assert result.error is not None + assert any("Missing MODEL_REQUIREMENTS" in e for e in result.error.errors) + + +# ────────────────────────────────────────────────────────────── +# Integration tests: discover_plugins +# ────────────────────────────────────────────────────────────── + +class TestDiscoverPlugins: + def test_discover_from_valid_dir(self, plugins_dir: Path) -> None: + results = discover_plugins([plugins_dir]) + # Should find multiple plugins (valid and invalid) + assert len(results) > 0 + + def test_discover_includes_valid_plugin(self, plugins_dir: Path) -> None: + results = discover_plugins([plugins_dir]) + valid = [r for r in results if r.success] + assert len(valid) >= 1 + assert any(r.manifest and r.manifest.id == "fake_plugin" for r in valid) + + def test_discover_includes_invalid_plugins(self, plugins_dir: Path) -> None: + results = discover_plugins([plugins_dir]) + invalid = [r for r in results if not r.success] + assert len(invalid) >= 1 + + def test_discover_nonexistent_dir(self, tmp_path: Path) -> None: + results = discover_plugins([tmp_path / "nonexistent"]) + assert results == [] + + def test_discover_multiple_dirs(self, plugins_dir: Path, tmp_path: Path) -> None: + results = discover_plugins([plugins_dir, tmp_path / "nonexistent"]) + assert len(results) > 0 + + +# ────────────────────────────────────────────────────────────── +# Diagnostic messages tests +# ────────────────────────────────────────────────────────────── + +class TestDiagnosticMessages: + def test_error_contains_plugin_id(self, missing_manifest_dir: Path) -> None: + result = load_plugin_from_dir(missing_manifest_dir) + assert result.error is not None + assert result.error.plugin_id == "missing_manifest" + + def test_error_contains_path(self, missing_manifest_dir: Path) -> None: + result = load_plugin_from_dir(missing_manifest_dir) + assert result.error is not None + assert result.error.path == missing_manifest_dir + + def test_error_contains_messages(self, missing_manifest_dir: Path) -> None: + result = load_plugin_from_dir(missing_manifest_dir) + assert result.error is not None + assert len(result.error.errors) > 0 + + def test_multiple_errors(self, invalid_api_version_dir: Path) -> None: + # This plugin has multiple issues + result = load_plugin_from_dir(invalid_api_version_dir) + assert result.error is not None + # Should have at least the api_version error + assert len(result.error.errors) >= 1 + + +# ────────────────────────────────────────────────────────────── +# No partial registration tests +# ────────────────────────────────────────────────────────────── + +class TestNoPartialRegistration: + def test_invalid_plugin_no_manifest_attr(self, missing_manifest_dir: Path) -> None: + """After failed load, module should not remain in sys.modules.""" + result = load_plugin_from_dir(missing_manifest_dir) + assert result.success is False + # Module should not be registered + module_name = f"abogen.tts_plugin._loaded.missing_manifest" + assert module_name not in sys.modules + + def test_import_error_no_registration(self, import_error_dir: Path) -> None: + """After import error, module should not remain in sys.modules.""" + result = load_plugin_from_dir(import_error_dir) + assert result.success is False + module_name = f"abogen.tts_plugin._loaded.import_error" + assert module_name not in sys.modules diff --git a/tests/plugins/fake_plugin/__init__.py b/tests/plugins/fake_plugin/__init__.py new file mode 100644 index 0000000..e23c8d6 --- /dev/null +++ b/tests/plugins/fake_plugin/__init__.py @@ -0,0 +1,92 @@ +"""Fake plugin for testing the plugin loader. + +This is a minimal valid plugin that satisfies the Plugin API contract. +It does NOT perform any real TTS synthesis. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from abogen.tts_plugin.engine import Engine, EngineSession +from abogen.tts_plugin.errors import EngineError +from abogen.tts_plugin.host_context import HostContext +from abogen.tts_plugin.manifest import ( + AudioFormatManifest, + EngineManifest, + PluginManifest, + VoiceSourceManifest, +) +from abogen.tts_plugin.types import ( + AudioFormat, + Duration, + EngineConfig, + ParameterValues, + SynthesisRequest, + SynthesizedAudio, +) + + +class FakeSession: + """Minimal EngineSession implementation for testing.""" + + def __init__(self) -> None: + self._disposed = False + + def synthesize(self, request: SynthesisRequest) -> SynthesizedAudio: + if self._disposed: + 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 FakeEngine: + """Minimal Engine implementation for testing.""" + + def __init__(self) -> None: + self._disposed = False + + def createSession(self) -> EngineSession: + if self._disposed: + raise EngineError("Engine disposed") + return FakeSession() + + def dispose(self) -> None: + self._disposed = True + + +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( + voiceSources=( + VoiceSourceManifest(id="builtin", name="Builtin", type="list"), + ), + audioFormats=( + AudioFormatManifest(mime="audio/wav", extension="wav"), + ), + ), +) + +MODEL_REQUIREMENTS: list[Any] = [] + + +def create_engine( + context: HostContext, + model_path: Path | None, + config: EngineConfig, +) -> Engine: + """Create a fake engine instance.""" + return FakeEngine() diff --git a/tests/plugins/import_error/__init__.py b/tests/plugins/import_error/__init__.py new file mode 100644 index 0000000..80bd647 --- /dev/null +++ b/tests/plugins/import_error/__init__.py @@ -0,0 +1,18 @@ +"""Invalid plugin: raises ImportError during import.""" + +from __future__ import annotations + +# This plugin intentionally raises an ImportError +raise ImportError("Simulated import error for testing") + +# The following code will never be reached, but is here for documentation +from abogen.tts_plugin.manifest import PluginManifest + +PLUGIN_MANIFEST = PluginManifest( + id="import_error", + name="Import Error Plugin", + version="1.0.0", + api_version="1.0", + description="Plugin that fails to import", + author="Test Author", +) diff --git a/tests/plugins/invalid_api_version/__init__.py b/tests/plugins/invalid_api_version/__init__.py new file mode 100644 index 0000000..7b44b84 --- /dev/null +++ b/tests/plugins/invalid_api_version/__init__.py @@ -0,0 +1,29 @@ +"""Invalid plugin: incompatible api_version (major version mismatch).""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from abogen.tts_plugin.manifest import PluginManifest +from abogen.tts_plugin.types import EngineConfig + +# api_version "2.0" has major version 2, but host expects major version 1 +PLUGIN_MANIFEST = PluginManifest( + id="invalid_api_version", + name="Invalid API Version Plugin", + version="1.0.0", + api_version="2.0", # Major version mismatch! + description="Plugin with incompatible api_version", + author="Test Author", +) + +MODEL_REQUIREMENTS: list[Any] = [] + + +def create_engine( + context: Any, + model_path: Path | None, + config: EngineConfig, +) -> Any: + raise NotImplementedError("This plugin is invalid") diff --git a/tests/plugins/invalid_capabilities/__init__.py b/tests/plugins/invalid_capabilities/__init__.py new file mode 100644 index 0000000..525e727 --- /dev/null +++ b/tests/plugins/invalid_capabilities/__init__.py @@ -0,0 +1,29 @@ +"""Invalid plugin: unknown capabilities.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from abogen.tts_plugin.manifest import PluginManifest +from abogen.tts_plugin.types import EngineConfig + +PLUGIN_MANIFEST = PluginManifest( + id="invalid_capabilities", + name="Invalid Capabilities Plugin", + version="1.0.0", + api_version="1.0", + description="Plugin with unknown capabilities", + author="Test Author", + capabilities=("voice_list", "unknown_capability", "another_unknown"), +) + +MODEL_REQUIREMENTS: list[Any] = [] + + +def create_engine( + context: Any, + model_path: Path | None, + config: EngineConfig, +) -> Any: + raise NotImplementedError("This plugin is invalid") diff --git a/tests/plugins/missing_create_engine/__init__.py b/tests/plugins/missing_create_engine/__init__.py new file mode 100644 index 0000000..d837f95 --- /dev/null +++ b/tests/plugins/missing_create_engine/__init__.py @@ -0,0 +1,18 @@ +"""Invalid plugin: missing create_engine function.""" + +from __future__ import annotations + +from abogen.tts_plugin.manifest import PluginManifest + +PLUGIN_MANIFEST = PluginManifest( + id="missing_create_engine", + name="Missing Create Engine Plugin", + version="1.0.0", + api_version="1.0", + description="Plugin missing create_engine", + author="Test Author", +) + +MODEL_REQUIREMENTS: list = [] + +# This plugin intentionally does NOT export create_engine diff --git a/tests/plugins/missing_manifest/__init__.py b/tests/plugins/missing_manifest/__init__.py new file mode 100644 index 0000000..38af923 --- /dev/null +++ b/tests/plugins/missing_manifest/__init__.py @@ -0,0 +1,10 @@ +"""Invalid plugin: missing PLUGIN_MANIFEST.""" + +from __future__ import annotations + +# This plugin intentionally does NOT export PLUGIN_MANIFEST +MODEL_REQUIREMENTS: list = [] + + +def create_engine(context, model_path, config): + raise NotImplementedError("This plugin is invalid") diff --git a/tests/plugins/missing_model_requirements/__init__.py b/tests/plugins/missing_model_requirements/__init__.py new file mode 100644 index 0000000..50b8441 --- /dev/null +++ b/tests/plugins/missing_model_requirements/__init__.py @@ -0,0 +1,28 @@ +"""Invalid plugin: missing MODEL_REQUIREMENTS.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from abogen.tts_plugin.manifest import PluginManifest +from abogen.tts_plugin.types import EngineConfig + +PLUGIN_MANIFEST = PluginManifest( + id="missing_model_requirements", + name="Missing Model Requirements Plugin", + version="1.0.0", + api_version="1.0", + description="Plugin missing MODEL_REQUIREMENTS", + author="Test Author", +) + +# This plugin intentionally does NOT export MODEL_REQUIREMENTS + + +def create_engine( + context: Any, + model_path: Path | None, + config: EngineConfig, +) -> Any: + raise NotImplementedError("This plugin is invalid")