298 lines
10 KiB
Python
298 lines
10 KiB
Python
|
|
"""Tests for the Hermes Piper TTS plugin.
|
||
|
|
|
||
|
|
Tests cover:
|
||
|
|
- Manifest structure (plugin.yaml)
|
||
|
|
- Monkey-patch application and idempotency
|
||
|
|
- Error handling when tts_tool is not importable
|
||
|
|
- register() context wiring
|
||
|
|
- on_session_start hook behavior
|
||
|
|
|
||
|
|
Run with:
|
||
|
|
cd /opt/data/projects/hermes-piper-plugin
|
||
|
|
python -m pytest tests/ -v
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import builtins
|
||
|
|
import importlib
|
||
|
|
import importlib.util
|
||
|
|
import logging
|
||
|
|
import sys
|
||
|
|
import types
|
||
|
|
from pathlib import Path
|
||
|
|
from unittest.mock import MagicMock, patch
|
||
|
|
|
||
|
|
import yaml
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
|
||
|
|
# =============================================================================
|
||
|
|
# Helpers
|
||
|
|
# =============================================================================
|
||
|
|
|
||
|
|
|
||
|
|
def _make_tools_tts(provider: str = "edge") -> types.ModuleType:
|
||
|
|
"""Create a mock tools.tts_tool submodule and set up sys.modules.
|
||
|
|
|
||
|
|
The parent 'tools' module must have ``tts_tool`` as an attribute because
|
||
|
|
Python's submodule import binds on the parent object.
|
||
|
|
"""
|
||
|
|
tts_mod = types.ModuleType("tools.tts_tool")
|
||
|
|
tts_mod.__package__ = "tools"
|
||
|
|
tts_mod.__name__ = "tools.tts_tool"
|
||
|
|
tts_mod.DEFAULT_PROVIDER = provider
|
||
|
|
|
||
|
|
tools_pkg = types.ModuleType("tools")
|
||
|
|
tools_pkg.__path__ = ["/tmp/fake-tools"]
|
||
|
|
tools_pkg.__package__ = "tools"
|
||
|
|
tools_pkg.__name__ = "tools"
|
||
|
|
tools_pkg.tts_tool = tts_mod
|
||
|
|
|
||
|
|
sys.modules["tools"] = tools_pkg
|
||
|
|
sys.modules["tools.tts_tool"] = tts_mod
|
||
|
|
return tts_mod
|
||
|
|
|
||
|
|
|
||
|
|
def _clean_tools_tts():
|
||
|
|
"""Remove mock tools/tools.tts_tool from sys.modules."""
|
||
|
|
for key in ("tools", "tools.tts_tool", "tools.tts_tool.fallback"):
|
||
|
|
sys.modules.pop(key, None)
|
||
|
|
|
||
|
|
|
||
|
|
def _load_plugin_module(name: str) -> types.ModuleType:
|
||
|
|
"""Load the plugin __init__.py as a fresh module with given name."""
|
||
|
|
source = Path(__file__).resolve().parent.parent / "__init__.py"
|
||
|
|
assert source.exists(), f"Plugin init not found at {source}"
|
||
|
|
spec = importlib.util.spec_from_file_location(name, str(source))
|
||
|
|
mod = importlib.util.module_from_spec(spec)
|
||
|
|
spec.loader.exec_module(mod)
|
||
|
|
# Reset _PATCHED for test isolation (register() isn't called at module load)
|
||
|
|
mod._PATCHED = False
|
||
|
|
mod.logger = logging.getLogger(f"test_{name}")
|
||
|
|
return mod
|
||
|
|
|
||
|
|
|
||
|
|
# =============================================================================
|
||
|
|
# Fixtures
|
||
|
|
# =============================================================================
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def plugin_dir() -> Path:
|
||
|
|
return Path(__file__).resolve().parent.parent
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def manifest(plugin_dir: Path) -> dict:
|
||
|
|
manifest_path = plugin_dir / "plugin.yaml"
|
||
|
|
assert manifest_path.exists(), f"plugin.yaml not found at {manifest_path}"
|
||
|
|
with open(manifest_path) as f:
|
||
|
|
return yaml.safe_load(f)
|
||
|
|
|
||
|
|
|
||
|
|
# =============================================================================
|
||
|
|
# Manifest tests
|
||
|
|
# =============================================================================
|
||
|
|
|
||
|
|
|
||
|
|
class TestManifest:
|
||
|
|
"""plugin.yaml must be present and structurally valid."""
|
||
|
|
|
||
|
|
def test_manifest_exists(self, manifest):
|
||
|
|
"""plugin.yaml parses as valid YAML."""
|
||
|
|
assert isinstance(manifest, dict)
|
||
|
|
|
||
|
|
def test_manifest_required_fields(self, manifest):
|
||
|
|
"""Required fields: name, version, description, hooks."""
|
||
|
|
assert "name" in manifest and manifest["name"]
|
||
|
|
assert "version" in manifest and manifest["version"]
|
||
|
|
assert "description" in manifest and manifest["description"]
|
||
|
|
|
||
|
|
def test_manifest_hooks(self, manifest):
|
||
|
|
"""Plugin must declare on_session_start hook."""
|
||
|
|
hooks = manifest.get("hooks", [])
|
||
|
|
assert isinstance(hooks, list)
|
||
|
|
assert "on_session_start" in hooks
|
||
|
|
|
||
|
|
def test_manifest_requires_pip(self, manifest):
|
||
|
|
"""Piper-tts should be listed as a pip dependency."""
|
||
|
|
deps = manifest.get("requires_pip", [])
|
||
|
|
assert "piper-tts" in deps, (
|
||
|
|
"piper-tts must be listed in requires_pip so the plugin system "
|
||
|
|
"can prompt users to install it"
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
# =============================================================================
|
||
|
|
# Monkey-patch tests
|
||
|
|
# =============================================================================
|
||
|
|
|
||
|
|
|
||
|
|
class TestPiperPatch:
|
||
|
|
"""Test the core monkey-patch logic."""
|
||
|
|
|
||
|
|
def teardown_method(self):
|
||
|
|
_clean_tools_tts()
|
||
|
|
|
||
|
|
def test_patch_changes_default_provider(self):
|
||
|
|
"""_apply_piper_patch() must set DEFAULT_PROVIDER to 'piper'."""
|
||
|
|
mock_tts = _make_tools_tts("edge")
|
||
|
|
mod = _load_plugin_module("test_patch_provider")
|
||
|
|
|
||
|
|
mod._apply_piper_patch()
|
||
|
|
|
||
|
|
assert mock_tts.DEFAULT_PROVIDER == "piper", (
|
||
|
|
f"Expected DEFAULT_PROVIDER='piper', got "
|
||
|
|
f"'{mock_tts.DEFAULT_PROVIDER}'"
|
||
|
|
)
|
||
|
|
assert mod._PATCHED is True
|
||
|
|
|
||
|
|
def test_patch_idempotent(self):
|
||
|
|
"""Calling _apply_piper_patch() twice should not error."""
|
||
|
|
mock_tts = _make_tools_tts("edge")
|
||
|
|
mod = _load_plugin_module("test_idempotent")
|
||
|
|
|
||
|
|
# First call
|
||
|
|
mod._apply_piper_patch()
|
||
|
|
assert mod._PATCHED is True
|
||
|
|
assert mock_tts.DEFAULT_PROVIDER == "piper"
|
||
|
|
|
||
|
|
# Modify back to test second call is no-op
|
||
|
|
mock_tts.DEFAULT_PROVIDER = "elevenlabs"
|
||
|
|
|
||
|
|
# Second call — should short-circuit on _PATCHED
|
||
|
|
mod._apply_piper_patch()
|
||
|
|
|
||
|
|
# Should NOT have changed (flag prevents reapply)
|
||
|
|
assert mock_tts.DEFAULT_PROVIDER == "elevenlabs"
|
||
|
|
|
||
|
|
def test_patch_import_error_safe(self):
|
||
|
|
"""If tools.tts_tool is not importable, patch must not crash.
|
||
|
|
|
||
|
|
We simulate the failure by patching the import inside the function
|
||
|
|
to raise ImportError. This verifies the ``except ImportError``
|
||
|
|
branch is reached and _PATCHED stays False.
|
||
|
|
"""
|
||
|
|
mod = _load_plugin_module("test_import_error")
|
||
|
|
|
||
|
|
original_import = builtins.__import__
|
||
|
|
|
||
|
|
def broken_import(name, *args, **kw):
|
||
|
|
if name == "tools.tts_tool":
|
||
|
|
raise ImportError(f"No module named {name}")
|
||
|
|
return original_import(name, *args, **kw)
|
||
|
|
|
||
|
|
with patch("builtins.__import__", side_effect=broken_import):
|
||
|
|
mod._apply_piper_patch()
|
||
|
|
assert mod._PATCHED is False # patch was deferred
|
||
|
|
|
||
|
|
def test_patch_generic_exception_safe(self):
|
||
|
|
"""If something goes wrong during patching, catch and log."""
|
||
|
|
mock_tts = _make_tools_tts("edge")
|
||
|
|
mod = _load_plugin_module("test_exception")
|
||
|
|
|
||
|
|
# Should not raise — the patch uses getattr with default and setattr
|
||
|
|
mod._apply_piper_patch()
|
||
|
|
|
||
|
|
# With our mock, the patch should succeed
|
||
|
|
assert mod._PATCHED is True
|
||
|
|
assert mock_tts.DEFAULT_PROVIDER == "piper"
|
||
|
|
|
||
|
|
|
||
|
|
# =============================================================================
|
||
|
|
# on_session_start hook tests
|
||
|
|
# =============================================================================
|
||
|
|
|
||
|
|
|
||
|
|
class TestSessionStartHook:
|
||
|
|
"""Test the on_session_start lifecycle hook."""
|
||
|
|
|
||
|
|
def teardown_method(self):
|
||
|
|
_clean_tools_tts()
|
||
|
|
|
||
|
|
def test_hook_does_not_raise(self):
|
||
|
|
"""_on_session_start must not raise even if tts_tool is missing."""
|
||
|
|
mod = _load_plugin_module("test_hook_no_raise")
|
||
|
|
mod._on_session_start() # should not raise
|
||
|
|
|
||
|
|
def test_hook_with_patch_applied(self):
|
||
|
|
"""_on_session_start should work with patch already applied."""
|
||
|
|
mock_tts = _make_tools_tts("edge")
|
||
|
|
mod = _load_plugin_module("test_hook_applied")
|
||
|
|
mod._apply_piper_patch()
|
||
|
|
|
||
|
|
assert mock_tts.DEFAULT_PROVIDER == "piper"
|
||
|
|
|
||
|
|
# Hook should run without error
|
||
|
|
mod._on_session_start()
|
||
|
|
|
||
|
|
|
||
|
|
# =============================================================================
|
||
|
|
# register() tests
|
||
|
|
# =============================================================================
|
||
|
|
|
||
|
|
|
||
|
|
class TestRegister:
|
||
|
|
"""Test the plugin's register() entry point."""
|
||
|
|
|
||
|
|
def teardown_method(self):
|
||
|
|
_clean_tools_tts()
|
||
|
|
|
||
|
|
def test_register_applies_patch(self):
|
||
|
|
"""register() should apply the piper patch immediately."""
|
||
|
|
mock_tts = _make_tools_tts("edge")
|
||
|
|
mod = _load_plugin_module("test_register_applies")
|
||
|
|
ctx = MagicMock()
|
||
|
|
|
||
|
|
mod.register(ctx)
|
||
|
|
|
||
|
|
assert mock_tts.DEFAULT_PROVIDER == "piper"
|
||
|
|
ctx.register_hook.assert_called_once_with(
|
||
|
|
"on_session_start", mod._on_session_start
|
||
|
|
)
|
||
|
|
|
||
|
|
def test_register_no_error_if_missing_deps(self):
|
||
|
|
"""register() must not crash if tools.tts_tool is not available."""
|
||
|
|
mod = _load_plugin_module("test_register_nodeps")
|
||
|
|
ctx = MagicMock()
|
||
|
|
|
||
|
|
# Should not raise
|
||
|
|
mod.register(ctx)
|
||
|
|
|
||
|
|
# Should still have registered the hook even if patch was deferred
|
||
|
|
ctx.register_hook.assert_called_once()
|
||
|
|
|
||
|
|
|
||
|
|
# =============================================================================
|
||
|
|
# Plugin structure tests
|
||
|
|
# =============================================================================
|
||
|
|
|
||
|
|
|
||
|
|
class TestPluginStructure:
|
||
|
|
"""Verify the plugin directory has all required files."""
|
||
|
|
|
||
|
|
def test_required_files_exist(self, plugin_dir):
|
||
|
|
"""Plugin must have plugin.yaml, __init__.py, and README.md."""
|
||
|
|
assert (plugin_dir / "plugin.yaml").exists(), "Missing plugin.yaml"
|
||
|
|
assert (plugin_dir / "__init__.py").exists(), "Missing __init__.py"
|
||
|
|
assert (plugin_dir / "README.md").exists(), "Missing README.md"
|
||
|
|
|
||
|
|
def test_init_has_register_function(self, plugin_dir):
|
||
|
|
"""__init__.py must expose a register() function."""
|
||
|
|
spec = importlib.util.spec_from_file_location(
|
||
|
|
"test_register_check", str(plugin_dir / "__init__.py")
|
||
|
|
)
|
||
|
|
mod = importlib.util.module_from_spec(spec)
|
||
|
|
spec.loader.exec_module(mod)
|
||
|
|
assert hasattr(mod, "register"), (
|
||
|
|
"__init__.py must define a register(ctx) function"
|
||
|
|
)
|
||
|
|
assert callable(mod.register), "register must be callable"
|
||
|
|
|
||
|
|
def test_plugin_yaml_valid(self, manifest):
|
||
|
|
"""plugin.yaml must contain valid plugin metadata."""
|
||
|
|
assert isinstance(manifest.get("name"), str)
|
||
|
|
assert isinstance(manifest.get("version"), str)
|
||
|
|
assert isinstance(manifest.get("hooks"), list)
|